Loading src/common/tests/MockServicerImpl_Context.py +16 −8 Original line number Diff line number Diff line Loading @@ -14,6 +14,7 @@ import grpc, json, logging from typing import Any, Dict, Iterator, List, Set from common.Constants import DEFAULT_CONTEXT_NAME, DEFAULT_TOPOLOGY_NAME from common.proto.context_pb2 import ( Connection, ConnectionEvent, ConnectionId, ConnectionIdList, ConnectionList, Context, ContextEvent, ContextId, ContextIdList, ContextList, Loading Loading @@ -164,11 +165,18 @@ class MockServicerImpl_Context(ContextServiceServicer): def GetTopologyDetails(self, request : TopologyId, context : grpc.ServicerContext) -> TopologyDetails: LOGGER.info('[GetTopologyDetails] request={:s}'.format(grpc_message_to_json_string(request))) container_name = 'topology[{:s}]'.format(str(request.context_id.context_uuid.uuid)) _reply = get_entry(context, self.database, container_name, request.topology_uuid.uuid) context_uuid = request.context_id.context_uuid.uuid container_name = 'topology[{:s}]'.format(str(context_uuid)) topology_uuid = request.topology_uuid.uuid _reply = get_entry(context, self.database, container_name, topology_uuid) reply = TopologyDetails() reply.topology_id.CopyFrom(_reply.topology_id) reply.name = _reply.name if context_uuid == DEFAULT_CONTEXT_NAME and topology_uuid == DEFAULT_TOPOLOGY_NAME: for device in get_entries(self.database, 'device'): reply.devices.append(device) for link in get_entries(self.database, 'link'): reply.links.append(link) else: # TODO: to be improved; Mock does not associate devices/links to topologies automatically for device_id in _reply.device_ids: device = get_entry(context, self.database, 'device', device_id.device_uuid.uuid) reply.devices.append(device) Loading Loading
src/common/tests/MockServicerImpl_Context.py +16 −8 Original line number Diff line number Diff line Loading @@ -14,6 +14,7 @@ import grpc, json, logging from typing import Any, Dict, Iterator, List, Set from common.Constants import DEFAULT_CONTEXT_NAME, DEFAULT_TOPOLOGY_NAME from common.proto.context_pb2 import ( Connection, ConnectionEvent, ConnectionId, ConnectionIdList, ConnectionList, Context, ContextEvent, ContextId, ContextIdList, ContextList, Loading Loading @@ -164,11 +165,18 @@ class MockServicerImpl_Context(ContextServiceServicer): def GetTopologyDetails(self, request : TopologyId, context : grpc.ServicerContext) -> TopologyDetails: LOGGER.info('[GetTopologyDetails] request={:s}'.format(grpc_message_to_json_string(request))) container_name = 'topology[{:s}]'.format(str(request.context_id.context_uuid.uuid)) _reply = get_entry(context, self.database, container_name, request.topology_uuid.uuid) context_uuid = request.context_id.context_uuid.uuid container_name = 'topology[{:s}]'.format(str(context_uuid)) topology_uuid = request.topology_uuid.uuid _reply = get_entry(context, self.database, container_name, topology_uuid) reply = TopologyDetails() reply.topology_id.CopyFrom(_reply.topology_id) reply.name = _reply.name if context_uuid == DEFAULT_CONTEXT_NAME and topology_uuid == DEFAULT_TOPOLOGY_NAME: for device in get_entries(self.database, 'device'): reply.devices.append(device) for link in get_entries(self.database, 'link'): reply.links.append(link) else: # TODO: to be improved; Mock does not associate devices/links to topologies automatically for device_id in _reply.device_ids: device = get_entry(context, self.database, 'device', device_id.device_uuid.uuid) reply.devices.append(device) Loading