Skip to content
Snippets Groups Projects

Pre-release CI/CD fixes - Service - PathComp

Merged Lluis Gifre Renom requested to merge feat/pre-release-fixes into develop
Files
30
@@ -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,
@@ -22,7 +23,7 @@ from common.proto.context_pb2 import (
Link, LinkEvent, LinkId, LinkIdList, LinkList,
Service, ServiceEvent, ServiceFilter, ServiceId, ServiceIdList, ServiceList,
Slice, SliceEvent, SliceFilter, SliceId, SliceIdList, SliceList,
Topology, TopologyEvent, TopologyId, TopologyIdList, TopologyList)
Topology, TopologyDetails, TopologyEvent, TopologyId, TopologyIdList, TopologyList)
from common.proto.context_pb2_grpc import ContextServiceServicer
from common.tests.MockMessageBroker import (
TOPIC_CONNECTION, TOPIC_CONTEXT, TOPIC_DEVICE, TOPIC_LINK, TOPIC_SERVICE, TOPIC_SLICE, TOPIC_TOPOLOGY,
@@ -162,6 +163,29 @@ class MockServicerImpl_Context(ContextServiceServicer):
LOGGER.info('[GetTopology] reply={:s}'.format(grpc_message_to_json_string(reply)))
return reply
def GetTopologyDetails(self, request : TopologyId, context : grpc.ServicerContext) -> TopologyDetails:
LOGGER.info('[GetTopologyDetails] request={:s}'.format(grpc_message_to_json_string(request)))
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)
for link_id in _reply.link_ids:
link = get_entry(context, self.database, 'link', link_id.link_uuid.uuid)
reply.links.append(link)
LOGGER.info('[GetTopologyDetails] reply={:s}'.format(grpc_message_to_json_string(reply)))
return reply
def SetTopology(self, request: Topology, context : grpc.ServicerContext) -> TopologyId:
LOGGER.info('[SetTopology] request={:s}'.format(grpc_message_to_json_string(request)))
container_name = 'topology[{:s}]'.format(str(request.topology_id.context_id.context_uuid.uuid))
Loading