Skip to content
Snippets Groups Projects
Commit 291f0802 authored by Shayan Hajipour's avatar Shayan Hajipour
Browse files

feat: QoSProfile error handling added:

- not found and already exists errors added to server implementations
- test_crud.py updated
parent 276a8d44
No related branches found
No related tags found
2 merge requests!294Release TeraFlowSDN 4.0,!257Resolve "Create QoSProfile component"
This commit is part of merge request !249. Comments created here will be created in the context of that merge request.
......@@ -268,7 +268,12 @@ class ContextServiceServicerImpl(ContextServiceServicer, ContextPolicyServiceSer
@safe_and_metered_rpc_method(METRICS_POOL, LOGGER)
def GetQoSProfile(self, request : QoSProfileId, context : grpc.ServicerContext) -> QoSProfile:
return get_qos_profile(self.db_engine, request.qos_profile_id.uuid)
qos_profile = get_qos_profile(self.db_engine, request.qos_profile_id.uuid)
if qos_profile is None:
context.set_details(f'QoSProfile {request.qos_profile_id.uuid} not found')
context.set_code(grpc.StatusCode.NOT_FOUND)
return QoSProfile()
return qos_profile
@safe_and_metered_rpc_method(METRICS_POOL, LOGGER)
def GetQoSProfiles(self, request : Empty, context : grpc.ServicerContext) -> Iterator[QoSProfile]:
......
......@@ -103,14 +103,11 @@ def delete_qos_profile(db_engine : Engine, request : str) -> Empty:
deleted = run_transaction(sessionmaker(bind=db_engine), callback)
return Empty()
def get_qos_profile(db_engine : Engine, request : str) -> QoSProfile:
def get_qos_profile(db_engine : Engine, request : str) -> Optional[QoSProfile]:
def callback(session : Session) -> Optional[QoSProfile]:
obj : Optional[QoSProfileModel] = session.query(QoSProfileModel).filter_by(qos_profile_id=request).one_or_none()
return None if obj is None else qos_table_data_to_grpc_message(obj)
qos_profile = run_transaction(sessionmaker(bind=db_engine), callback)
if qos_profile is None:
raise NotFoundException('QoSProfile', request)
return qos_profile
return run_transaction(sessionmaker(bind=db_engine), callback)
def get_qos_profiles(db_engine : Engine, request : Empty) -> List[QoSProfile]:
def callback(session : Session) -> List[QoSProfile]:
......
......@@ -14,8 +14,9 @@
import grpc, logging
from typing import Iterator
import grpc._channel
from common.method_wrappers.Decorator import MetricsPool, safe_and_metered_rpc_method
from common.method_wrappers.ServiceExceptions import AlreadyExistsException
from common.proto.context_pb2 import Empty, QoSProfileId, QoSProfile
from common.proto.qos_profile_pb2_grpc import QoSProfileServiceServicer
from context.client.ContextClient import ContextClient
......@@ -34,10 +35,12 @@ class QoSProfileServiceServicerImpl(QoSProfileServiceServicer):
context_client = ContextClient()
try:
qos_profile_get = context_client.GetQoSProfile(request.qos_profile_id)
except grpc.RpcError:
qos_profile_get = None
if qos_profile_get:
raise AlreadyExistsException('QoSProfile', request.qos_profile_id)
context.set_details(f'QoSProfile {request.qos_profile_id.qos_profile_id.uuid} already exists')
context.set_code(grpc.StatusCode.ALREADY_EXISTS)
return QoSProfile()
except grpc._channel._InactiveRpcError as exc:
if exc.code() != grpc.StatusCode.NOT_FOUND:
raise exc
qos_profile = context_client.CreateQoSProfile(request)
return qos_profile
......@@ -46,8 +49,11 @@ class QoSProfileServiceServicerImpl(QoSProfileServiceServicer):
context_client = ContextClient()
try:
_ = context_client.GetQoSProfile(request.qos_profile_id)
except grpc.RpcError as exc:
raise exc
except grpc._channel._InactiveRpcError as exc:
if exc.code() == grpc.StatusCode.NOT_FOUND:
context.set_details(f'QoSProfile {request.qos_profile_id.qos_profile_id.uuid} not found')
context.set_code(grpc.StatusCode.NOT_FOUND)
return QoSProfile()
qos_profile = context_client.UpdateQoSProfile(request)
return qos_profile
......@@ -56,8 +62,11 @@ class QoSProfileServiceServicerImpl(QoSProfileServiceServicer):
context_client = ContextClient()
try:
_ = context_client.GetQoSProfile(request)
except grpc.RpcError as exc:
raise exc
except grpc._channel._InactiveRpcError as exc:
if exc.code() == grpc.StatusCode.NOT_FOUND:
context.set_details(f'QoSProfile {request.qos_profile_id.uuid} not found')
context.set_code(grpc.StatusCode.NOT_FOUND)
return QoSProfile()
empty = context_client.DeleteQoSProfile(request)
return empty
......@@ -66,8 +75,11 @@ class QoSProfileServiceServicerImpl(QoSProfileServiceServicer):
context_client = ContextClient()
try:
qos_profile = context_client.GetQoSProfile(request)
except grpc.RpcError as exc:
raise exc
except grpc._channel._InactiveRpcError as exc:
if exc.code() == grpc.StatusCode.NOT_FOUND:
context.set_details(f'QoSProfile {request.qos_profile_id.uuid} not found')
context.set_code(grpc.StatusCode.NOT_FOUND)
return QoSProfile()
return qos_profile
@safe_and_metered_rpc_method(METRICS_POOL, LOGGER)
......
......@@ -12,7 +12,7 @@
# See the License for the specific language governing permissions and
# limitations under the License.
from grpc import RpcError
from grpc import RpcError, StatusCode
import logging, pytest
from common.proto.context_pb2 import Empty, Uuid, QoSProfileValueUnitPair, QoSProfileId, QoSProfile
......@@ -104,6 +104,7 @@ def test_failed_create_qos_profile(qos_profile_client: QoSProfileClient):
qos_profile = create_qos_profile_from_json(qos_profile_data)
with pytest.raises(RpcError) as exc:
qos_profile_created = qos_profile_client.CreateQoSProfile(qos_profile)
assert exc.value.code() == StatusCode.ALREADY_EXISTS
def test_get_qos_profile(qos_profile_client: QoSProfileClient):
qos_profile = create_qos_profile_from_json(qos_profile_data)
......@@ -124,7 +125,13 @@ def test_update_qos_profile(qos_profile_client: QoSProfileClient):
LOGGER.info('qos_profile_data = {:s}'.format(grpc_message_to_json_string(qos_profile_updated)))
assert qos_profile_updated.packetErrorLossRate == 5
def test_delete_qos_profiles(qos_profile_client: QoSProfileClient):
def test_failed_delete_qos_profiles(qos_profile_client: QoSProfileClient):
qos_profile = create_qos_profile_from_json(qos_profile_data)
with pytest.raises(RpcError) as exc:
qos_profiles_deleted = qos_profile_client.DeleteQoSProfile(QoSProfileId(qos_profile_id=Uuid(uuid='f8b1c625-ac01-405c-b1f7-b5ee06e16282')))
\ No newline at end of file
qos_profiles_deleted = qos_profile_client.DeleteQoSProfile(QoSProfileId(qos_profile_id=Uuid(uuid='f8b1c625-ac01-405c-b1f7-b5ee06e16282')))
assert exc.value.code() == StatusCode.NOT_FOUND
def test_delete_qos_profiles(qos_profile_client: QoSProfileClient):
qos_profile = create_qos_profile_from_json(qos_profile_data)
qos_profiles_deleted = qos_profile_client.DeleteQoSProfile(qos_profile.qos_profile_id)
assert qos_profiles_deleted == Empty()
\ No newline at end of file
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment