Upcoming maintenance: Thursday 21 August @ 12:00-14:00 CEST.

Skip to content
Snippets Groups Projects
Commit 9d6b4876 authored by Lluis Gifre Renom's avatar Lluis Gifre Renom
Browse files

MockDltGateway:

- Improved error logging
parent ae916d20
No related branches found
No related tags found
2 merge requests!54Release 2.0.0,!24Integrate NFV-SDN'22 demo
......@@ -36,6 +36,10 @@ class AlreadyExistsException(Exception):
class DoesNotExistException(Exception):
pass
MSG_NOT_EXISTS = 'RecordId({:s}, {:s}, {:s}) Does Not Exist'
MSG_ALREADY_EXISTS = 'RecordId({:s}, {:s}, {:s}) Already Exists'
MSG_OPERATION_NOT_IMPLEMENTED = 'DltRecordOperationEnum({:s}) Not Implemented'
class MockServicerImpl_DltGateway(DltGatewayServiceServicer):
def __init__(self):
LOGGER.info('[__init__] Creating Servicer...')
......@@ -50,9 +54,9 @@ class MockServicerImpl_DltGateway(DltGatewayServiceServicer):
records_type : Dict[str, Dict] = records_domain.setdefault(str_type, {})
record : Optional[Dict] = records_type.get(record_uuid)
if should_exist and record is None:
raise DoesNotExistException('RecordId({:s}, {:s}, {:s})'.format(domain_uuid, str_type, record_uuid))
raise DoesNotExistException(MSG_NOT_EXISTS.format(domain_uuid, str_type, record_uuid))
elif not should_exist and record is not None:
raise AlreadyExistsException('RecordId({:s}, {:s}, {:s})'.format(domain_uuid, str_type, record_uuid))
raise AlreadyExistsException(MSG_ALREADY_EXISTS.format(domain_uuid, str_type, record_uuid))
return record
def __set_record(self, record_id : DltRecordId, should_exist : bool, data_json : str) -> None:
......@@ -62,9 +66,9 @@ class MockServicerImpl_DltGateway(DltGatewayServiceServicer):
records_type : Dict[str, Dict] = records_domain.setdefault(str_type, {})
record : Optional[Dict] = records_type.get(record_uuid)
if should_exist and record is None:
raise DoesNotExistException('RecordId({:s}, {:s}, {:s})'.format(domain_uuid, str_type, record_uuid))
raise DoesNotExistException(MSG_NOT_EXISTS.format(domain_uuid, str_type, record_uuid))
elif not should_exist and record is not None:
raise AlreadyExistsException('RecordId({:s}, {:s}, {:s})'.format(domain_uuid, str_type, record_uuid))
raise AlreadyExistsException(MSG_ALREADY_EXISTS.format(domain_uuid, str_type, record_uuid))
records_type[record_uuid] = json.loads(data_json)
def __del_record(self, record_id : DltRecordId) -> None:
......@@ -74,7 +78,7 @@ class MockServicerImpl_DltGateway(DltGatewayServiceServicer):
records_type : Dict[str, Dict] = records_domain.setdefault(str_type, {})
record : Optional[Dict] = records_type.get(record_uuid)
if record is None:
raise DoesNotExistException('RecordId({:s}, {:s}, {:s})'.format(domain_uuid, str_type, record_uuid))
raise DoesNotExistException(MSG_NOT_EXISTS.format(domain_uuid, str_type, record_uuid))
records_type.discard(record_uuid)
def __publish(self, operation : DltRecordOperationEnum, record_id : DltRecordId) -> None:
......@@ -106,7 +110,7 @@ class MockServicerImpl_DltGateway(DltGatewayServiceServicer):
self.__del_record(record_id)
else:
str_operation = DltRecordOperationEnum.Name(operation).upper().replace('DLTRECORDOPERATION_', '')
raise NotImplementedError('DltRecordOperationEnum({:s})'.format(str_operation))
raise NotImplementedError(MSG_OPERATION_NOT_IMPLEMENTED.format(str_operation))
self.__publish(operation, record_id)
response.status = DLTRECORDSTATUS_SUCCEEDED
except Exception as e: # pylint: disable=broad-except
......
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