diff --git a/src/common/tests/InMemoryObjectDatabase.py b/src/common/tests/InMemoryObjectDatabase.py
index 47623026ccd730311837d661d5c1806c7f945f79..21697a4355795775cc25112671c4e436fbbecb8c 100644
--- a/src/common/tests/InMemoryObjectDatabase.py
+++ b/src/common/tests/InMemoryObjectDatabase.py
@@ -16,11 +16,10 @@ import grpc, logging
from typing import Any, Dict, List, Set
LOGGER = logging.getLogger(__name__)
-LOGGER.setLevel(logging.INFO)
class InMemoryObjectDatabase:
def __init__(self) -> None:
- self._database : Dict[str, Dict[str, Any]] = {}
+ self._database : Dict[str, Dict[str, Any]] = dict()
def _get_container(self, container_name : str) -> Dict[str, Any]:
return self._database.setdefault(container_name, {})
diff --git a/src/common/tests/InMemoryTimeSeriesDatabase.py b/src/common/tests/InMemoryTimeSeriesDatabase.py
index e0cd969f5add03e829b05220dd443b777f1ae339..1c4c86da88bfb3ca99ecd92e5baab7244bea414c 100644
--- a/src/common/tests/InMemoryTimeSeriesDatabase.py
+++ b/src/common/tests/InMemoryTimeSeriesDatabase.py
@@ -16,7 +16,6 @@ import logging, pandas
from typing import List, Optional
LOGGER = logging.getLogger(__name__)
-LOGGER.setLevel(logging.INFO)
class InMemoryTimeSeriesDatabase:
def __init__(self) -> None:
diff --git a/src/common/tests/MockServicerImpl_Context.py b/src/common/tests/MockServicerImpl_Context.py
index c7a3b48cc50426a9469a76a5f065461aa56fdf81..55f87b7b0c03a7ae563dc10bd5e4964a07317c21 100644
--- a/src/common/tests/MockServicerImpl_Context.py
+++ b/src/common/tests/MockServicerImpl_Context.py
@@ -13,7 +13,7 @@
# limitations under the License.
import grpc, json, logging
-from typing import Any, Dict, Iterator, List, Set, Tuple
+from typing import Any, Dict, Iterator, Set, Tuple
from common.Constants import DEFAULT_CONTEXT_NAME, DEFAULT_TOPOLOGY_NAME
from common.proto.context_pb2 import (
Connection, ConnectionEvent, ConnectionId, ConnectionIdList, ConnectionList,
@@ -25,213 +25,167 @@ from common.proto.context_pb2 import (
Slice, SliceEvent, SliceFilter, SliceId, SliceIdList, SliceList,
Topology, TopologyDetails, TopologyEvent, TopologyId, TopologyIdList, TopologyList)
from common.proto.context_pb2_grpc import ContextServiceServicer
-from common.tests.MockMessageBroker import (
+from common.tools.grpc.Tools import grpc_message_to_json, grpc_message_to_json_string
+from .InMemoryObjectDatabase import InMemoryObjectDatabase
+from .MockMessageBroker import (
TOPIC_CONNECTION, TOPIC_CONTEXT, TOPIC_DEVICE, TOPIC_LINK, TOPIC_SERVICE, TOPIC_SLICE, TOPIC_TOPOLOGY,
MockMessageBroker, notify_event)
-from common.tools.grpc.Tools import grpc_message_to_json, grpc_message_to_json_string
LOGGER = logging.getLogger(__name__)
-def get_container(database : Dict[str, Dict[str, Any]], container_name : str) -> Dict[str, Any]:
- return database.setdefault(container_name, {})
-
-def get_entries(database : Dict[str, Dict[str, Any]], container_name : str) -> List[Any]:
- container = get_container(database, container_name)
- return [container[entry_uuid] for entry_uuid in sorted(container.keys())]
-
-def has_entry(database : Dict[str, Dict[str, Any]], container_name : str, entry_uuid : str) -> Any:
- LOGGER.debug('[has_entry] BEFORE database={:s}'.format(str(database)))
- container = get_container(database, container_name)
- return entry_uuid in container
-
-def get_entry(
- context : grpc.ServicerContext, database : Dict[str, Dict[str, Any]], container_name : str, entry_uuid : str
-) -> Any:
- LOGGER.debug('[get_entry] BEFORE database={:s}'.format(str(database)))
- container = get_container(database, container_name)
- if entry_uuid not in container:
- context.abort(grpc.StatusCode.NOT_FOUND, str('{:s}({:s}) not found'.format(container_name, entry_uuid)))
- return container[entry_uuid]
-
-def set_entry(database : Dict[str, Dict[str, Any]], container_name : str, entry_uuid : str, entry : Any) -> Any:
- container = get_container(database, container_name)
- LOGGER.debug('[set_entry] BEFORE database={:s}'.format(str(database)))
- container[entry_uuid] = entry
- LOGGER.debug('[set_entry] AFTER database={:s}'.format(str(database)))
- return entry
-
-def del_entry(
- context : grpc.ServicerContext, database : Dict[str, Dict[str, Any]], container_name : str, entry_uuid : str
-) -> Any:
- container = get_container(database, container_name)
- if entry_uuid not in container:
- context.abort(grpc.StatusCode.NOT_FOUND, str('{:s}({:s}) not found'.format(container_name, entry_uuid)))
- del container[entry_uuid]
- return Empty()
-
-def select_entries(database : Dict[str, Dict[str, Any]], container_name : str, entry_uuids : Set[str]) -> List[Any]:
- if len(entry_uuids) == 0: return get_entries(database, container_name)
- container = get_container(database, container_name)
- return [
- container[entry_uuid]
- for entry_uuid in sorted(container.keys())
- if entry_uuid in entry_uuids
- ]
-
class MockServicerImpl_Context(ContextServiceServicer):
def __init__(self):
- LOGGER.info('[__init__] Creating Servicer...')
- self.database : Dict[str, Dict[str, Any]] = {}
+ LOGGER.debug('[__init__] Creating Servicer...')
+ self.obj_db = InMemoryObjectDatabase()
self.msg_broker = MockMessageBroker()
- LOGGER.info('[__init__] Servicer Created')
+ LOGGER.debug('[__init__] Servicer Created')
# ----- Common -----------------------------------------------------------------------------------------------------
- def _set(self, request, container_name, entry_uuid, entry_id_field_name, topic_name):
- exists = has_entry(self.database, container_name, entry_uuid)
- entry = set_entry(self.database, container_name, entry_uuid, request)
+ def _set(self, request, container_name, entry_uuid, entry_id_field_name, topic_name) -> Tuple[Any, Any]:
+ exists = self.obj_db.has_entry(container_name, entry_uuid)
+ entry = self.obj_db.set_entry(container_name, entry_uuid, request)
event_type = EventTypeEnum.EVENTTYPE_UPDATE if exists else EventTypeEnum.EVENTTYPE_CREATE
entry_id = getattr(entry, entry_id_field_name)
dict_entry_id = grpc_message_to_json(entry_id)
notify_event(self.msg_broker, topic_name, event_type, {entry_id_field_name: dict_entry_id})
- return entry_id
+ return entry_id, entry
- def _del(self, request, container_name, entry_uuid, entry_id_field_name, topic_name, grpc_context):
- empty = del_entry(grpc_context, self.database, container_name, entry_uuid)
+ def _del(self, request, container_name, entry_uuid, entry_id_field_name, topic_name, context) -> Empty:
+ self.obj_db.del_entry(container_name, entry_uuid, context)
event_type = EventTypeEnum.EVENTTYPE_REMOVE
dict_entry_id = grpc_message_to_json(request)
notify_event(self.msg_broker, topic_name, event_type, {entry_id_field_name: dict_entry_id})
- return empty
+ return Empty()
# ----- Context ----------------------------------------------------------------------------------------------------
def ListContextIds(self, request: Empty, context : grpc.ServicerContext) -> ContextIdList:
- LOGGER.info('[ListContextIds] request={:s}'.format(grpc_message_to_json_string(request)))
- reply = ContextIdList(context_ids=[context.context_id for context in get_entries(self.database, 'context')])
- LOGGER.info('[ListContextIds] reply={:s}'.format(grpc_message_to_json_string(reply)))
+ LOGGER.debug('[ListContextIds] request={:s}'.format(grpc_message_to_json_string(request)))
+ reply = ContextIdList(context_ids=[context.context_id for context in self.obj_db.get_entries('context')])
+ LOGGER.debug('[ListContextIds] reply={:s}'.format(grpc_message_to_json_string(reply)))
return reply
def ListContexts(self, request: Empty, context : grpc.ServicerContext) -> ContextList:
- LOGGER.info('[ListContexts] request={:s}'.format(grpc_message_to_json_string(request)))
- reply = ContextList(contexts=get_entries(self.database, 'context'))
- LOGGER.info('[ListContexts] reply={:s}'.format(grpc_message_to_json_string(reply)))
+ LOGGER.debug('[ListContexts] request={:s}'.format(grpc_message_to_json_string(request)))
+ reply = ContextList(contexts=self.obj_db.get_entries('context'))
+ LOGGER.debug('[ListContexts] reply={:s}'.format(grpc_message_to_json_string(reply)))
return reply
def GetContext(self, request: ContextId, context : grpc.ServicerContext) -> Context:
- LOGGER.info('[GetContext] request={:s}'.format(grpc_message_to_json_string(request)))
- reply = get_entry(context, self.database, 'context', request.context_uuid.uuid)
- LOGGER.info('[GetContext] reply={:s}'.format(grpc_message_to_json_string(reply)))
+ LOGGER.debug('[GetContext] request={:s}'.format(grpc_message_to_json_string(request)))
+ reply = self.obj_db.get_entry('context', request.context_uuid.uuid, context)
+ LOGGER.debug('[GetContext] reply={:s}'.format(grpc_message_to_json_string(reply)))
return reply
def SetContext(self, request: Context, context : grpc.ServicerContext) -> ContextId:
- LOGGER.info('[SetContext] request={:s}'.format(grpc_message_to_json_string(request)))
- reply = self._set(request, 'context', request.context_id.context_uuid.uuid, 'context_id', TOPIC_CONTEXT)
- LOGGER.info('[SetContext] reply={:s}'.format(grpc_message_to_json_string(reply)))
+ LOGGER.debug('[SetContext] request={:s}'.format(grpc_message_to_json_string(request)))
+ reply,_ = self._set(request, 'context', request.context_id.context_uuid.uuid, 'context_id', TOPIC_CONTEXT)
+ LOGGER.debug('[SetContext] reply={:s}'.format(grpc_message_to_json_string(reply)))
return reply
def RemoveContext(self, request: ContextId, context : grpc.ServicerContext) -> Empty:
- LOGGER.info('[RemoveContext] request={:s}'.format(grpc_message_to_json_string(request)))
+ LOGGER.debug('[RemoveContext] request={:s}'.format(grpc_message_to_json_string(request)))
reply = self._del(request, 'context', request.context_uuid.uuid, 'context_id', TOPIC_CONTEXT, context)
- LOGGER.info('[RemoveContext] reply={:s}'.format(grpc_message_to_json_string(reply)))
+ LOGGER.debug('[RemoveContext] reply={:s}'.format(grpc_message_to_json_string(reply)))
return reply
def GetContextEvents(self, request: Empty, context : grpc.ServicerContext) -> Iterator[ContextEvent]:
- LOGGER.info('[GetContextEvents] request={:s}'.format(grpc_message_to_json_string(request)))
+ LOGGER.debug('[GetContextEvents] request={:s}'.format(grpc_message_to_json_string(request)))
for message in self.msg_broker.consume({TOPIC_CONTEXT}): yield ContextEvent(**json.loads(message.content))
# ----- Topology ---------------------------------------------------------------------------------------------------
def ListTopologyIds(self, request: ContextId, context : grpc.ServicerContext) -> TopologyIdList:
- LOGGER.info('[ListTopologyIds] request={:s}'.format(grpc_message_to_json_string(request)))
- topologies = get_entries(self.database, 'topology[{:s}]'.format(str(request.context_uuid.uuid)))
+ LOGGER.debug('[ListTopologyIds] request={:s}'.format(grpc_message_to_json_string(request)))
+ topologies = self.obj_db.get_entries('topology[{:s}]'.format(str(request.context_uuid.uuid)))
reply = TopologyIdList(topology_ids=[topology.topology_id for topology in topologies])
- LOGGER.info('[ListTopologyIds] reply={:s}'.format(grpc_message_to_json_string(reply)))
+ LOGGER.debug('[ListTopologyIds] reply={:s}'.format(grpc_message_to_json_string(reply)))
return reply
def ListTopologies(self, request: ContextId, context : grpc.ServicerContext) -> TopologyList:
- LOGGER.info('[ListTopologies] request={:s}'.format(grpc_message_to_json_string(request)))
- topologies = get_entries(self.database, 'topology[{:s}]'.format(str(request.context_uuid.uuid)))
+ LOGGER.debug('[ListTopologies] request={:s}'.format(grpc_message_to_json_string(request)))
+ topologies = self.obj_db.get_entries('topology[{:s}]'.format(str(request.context_uuid.uuid)))
reply = TopologyList(topologies=[topology for topology in topologies])
- LOGGER.info('[ListTopologies] reply={:s}'.format(grpc_message_to_json_string(reply)))
+ LOGGER.debug('[ListTopologies] reply={:s}'.format(grpc_message_to_json_string(reply)))
return reply
def GetTopology(self, request: TopologyId, context : grpc.ServicerContext) -> Topology:
- LOGGER.info('[GetTopology] request={:s}'.format(grpc_message_to_json_string(request)))
+ LOGGER.debug('[GetTopology] 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)
- LOGGER.info('[GetTopology] reply={:s}'.format(grpc_message_to_json_string(reply)))
+ reply = self.obj_db.get_entry(container_name, request.topology_uuid.uuid, context)
+ LOGGER.debug('[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)))
+ LOGGER.debug('[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 = self.obj_db.get_entry(container_name, topology_uuid, context)
reply = TopologyDetails()
- reply.topology_id.CopyFrom(_reply.topology_id)
+ reply.topology_id.CopyFrom(_reply.topology_id) # pylint: disable=no-member
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)
+ for device in self.obj_db.get_entries('device'): reply.devices.append(device) # pylint: disable=no-member
+ for link in self.obj_db.get_entries('link' ): reply.links .append(link ) # pylint: disable=no-member
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)
+ device = self.obj_db.get_entry('device', device_id.device_uuid.uuid, context)
+ reply.devices.append(device) # pylint: disable=no-member
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)))
+ link = self.obj_db.get_entry('link', link_id.link_uuid.uuid, context)
+ reply.links.append(link) # pylint: disable=no-member
+ LOGGER.debug('[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)))
+ LOGGER.debug('[SetTopology] request={:s}'.format(grpc_message_to_json_string(request)))
container_name = 'topology[{:s}]'.format(str(request.topology_id.context_id.context_uuid.uuid))
topology_uuid = request.topology_id.topology_uuid.uuid
- reply = self._set(request, container_name, topology_uuid, 'topology_id', TOPIC_TOPOLOGY)
- LOGGER.info('[SetTopology] reply={:s}'.format(grpc_message_to_json_string(reply)))
+ reply,_ = self._set(request, container_name, topology_uuid, 'topology_id', TOPIC_TOPOLOGY)
+ LOGGER.debug('[SetTopology] reply={:s}'.format(grpc_message_to_json_string(reply)))
return reply
def RemoveTopology(self, request: TopologyId, context : grpc.ServicerContext) -> Empty:
- LOGGER.info('[RemoveTopology] request={:s}'.format(grpc_message_to_json_string(request)))
+ LOGGER.debug('[RemoveTopology] request={:s}'.format(grpc_message_to_json_string(request)))
container_name = 'topology[{:s}]'.format(str(request.context_id.context_uuid.uuid))
topology_uuid = request.topology_uuid.uuid
reply = self._del(request, container_name, topology_uuid, 'topology_id', TOPIC_TOPOLOGY, context)
- LOGGER.info('[RemoveTopology] reply={:s}'.format(grpc_message_to_json_string(reply)))
+ LOGGER.debug('[RemoveTopology] reply={:s}'.format(grpc_message_to_json_string(reply)))
return reply
def GetTopologyEvents(self, request: Empty, context : grpc.ServicerContext) -> Iterator[TopologyEvent]:
- LOGGER.info('[GetTopologyEvents] request={:s}'.format(grpc_message_to_json_string(request)))
+ LOGGER.debug('[GetTopologyEvents] request={:s}'.format(grpc_message_to_json_string(request)))
for message in self.msg_broker.consume({TOPIC_TOPOLOGY}): yield TopologyEvent(**json.loads(message.content))
# ----- Device -----------------------------------------------------------------------------------------------------
def ListDeviceIds(self, request: Empty, context : grpc.ServicerContext) -> DeviceIdList:
- LOGGER.info('[ListDeviceIds] request={:s}'.format(grpc_message_to_json_string(request)))
- reply = DeviceIdList(device_ids=[device.device_id for device in get_entries(self.database, 'device')])
- LOGGER.info('[ListDeviceIds] reply={:s}'.format(grpc_message_to_json_string(reply)))
+ LOGGER.debug('[ListDeviceIds] request={:s}'.format(grpc_message_to_json_string(request)))
+ reply = DeviceIdList(device_ids=[device.device_id for device in self.obj_db.get_entries('device')])
+ LOGGER.debug('[ListDeviceIds] reply={:s}'.format(grpc_message_to_json_string(reply)))
return reply
def ListDevices(self, request: Empty, context : grpc.ServicerContext) -> DeviceList:
- LOGGER.info('[ListDevices] request={:s}'.format(grpc_message_to_json_string(request)))
- reply = DeviceList(devices=get_entries(self.database, 'device'))
- LOGGER.info('[ListDevices] reply={:s}'.format(grpc_message_to_json_string(reply)))
+ LOGGER.debug('[ListDevices] request={:s}'.format(grpc_message_to_json_string(request)))
+ reply = DeviceList(devices=self.obj_db.get_entries('device'))
+ LOGGER.debug('[ListDevices] reply={:s}'.format(grpc_message_to_json_string(reply)))
return reply
def GetDevice(self, request: DeviceId, context : grpc.ServicerContext) -> Device:
- LOGGER.info('[GetDevice] request={:s}'.format(grpc_message_to_json_string(request)))
- reply = get_entry(context, self.database, 'device', request.device_uuid.uuid)
- LOGGER.info('[GetDevice] reply={:s}'.format(grpc_message_to_json_string(reply)))
+ LOGGER.debug('[GetDevice] request={:s}'.format(grpc_message_to_json_string(request)))
+ reply = self.obj_db.get_entry('device', request.device_uuid.uuid, context)
+ LOGGER.debug('[GetDevice] reply={:s}'.format(grpc_message_to_json_string(reply)))
return reply
def SetDevice(self, request: Context, context : grpc.ServicerContext) -> DeviceId:
- LOGGER.info('[SetDevice] request={:s}'.format(grpc_message_to_json_string(request)))
+ LOGGER.debug('[SetDevice] request={:s}'.format(grpc_message_to_json_string(request)))
device_uuid = request.device_id.device_uuid.uuid
- reply = self._set(request, 'device', device_uuid, 'device_id', TOPIC_DEVICE)
- device = get_entry(context, self.database, 'device', device_uuid)
+ reply, device = self._set(request, 'device', device_uuid, 'device_id', TOPIC_DEVICE)
context_topology_uuids : Set[Tuple[str, str]] = set()
context_topology_uuids.add((DEFAULT_CONTEXT_NAME, DEFAULT_TOPOLOGY_NAME))
@@ -244,20 +198,20 @@ class MockServicerImpl_Context(ContextServiceServicer):
for context_uuid,topology_uuid in context_topology_uuids:
container_name = 'topology[{:s}]'.format(str(context_uuid))
- topology = get_entry(context, self.database, container_name, topology_uuid)
- for device_id in topology.device_ids:
- if device_id.device_uuid.uuid == device_uuid: break
+ topology = self.obj_db.get_entry(container_name, topology_uuid, context)
+ for _device_id in topology.device_ids:
+ if _device_id.device_uuid.uuid == device_uuid: break
else:
# device not found, add it
topology.device_ids.add().device_uuid.uuid = device_uuid
- LOGGER.info('[SetDevice] reply={:s}'.format(grpc_message_to_json_string(reply)))
+ LOGGER.debug('[SetDevice] reply={:s}'.format(grpc_message_to_json_string(reply)))
return reply
def RemoveDevice(self, request: DeviceId, context : grpc.ServicerContext) -> Empty:
- LOGGER.info('[RemoveDevice] request={:s}'.format(grpc_message_to_json_string(request)))
+ LOGGER.debug('[RemoveDevice] request={:s}'.format(grpc_message_to_json_string(request)))
device_uuid = request.device_uuid.uuid
- device = get_entry(context, self.database, 'device', device_uuid)
+ device = self.obj_db.get_entry('device', device_uuid, context)
reply = self._del(request, 'device', device_uuid, 'device_id', TOPIC_DEVICE, context)
context_topology_uuids : Set[Tuple[str, str]] = set()
@@ -271,21 +225,21 @@ class MockServicerImpl_Context(ContextServiceServicer):
for context_uuid,topology_uuid in context_topology_uuids:
container_name = 'topology[{:s}]'.format(str(context_uuid))
- topology = get_entry(context, self.database, container_name, topology_uuid)
+ topology = self.obj_db.get_entry(container_name, topology_uuid, context)
for device_id in topology.device_ids:
if device_id.device_uuid.uuid == device_uuid:
topology.device_ids.remove(device_id)
break
- LOGGER.info('[RemoveDevice] reply={:s}'.format(grpc_message_to_json_string(reply)))
+ LOGGER.debug('[RemoveDevice] reply={:s}'.format(grpc_message_to_json_string(reply)))
return reply
def GetDeviceEvents(self, request: Empty, context : grpc.ServicerContext) -> Iterator[DeviceEvent]:
- LOGGER.info('[GetDeviceEvents] request={:s}'.format(grpc_message_to_json_string(request)))
+ LOGGER.debug('[GetDeviceEvents] request={:s}'.format(grpc_message_to_json_string(request)))
for message in self.msg_broker.consume({TOPIC_DEVICE}): yield DeviceEvent(**json.loads(message.content))
def SelectDevice(self, request : DeviceFilter, context : grpc.ServicerContext) -> DeviceList:
- LOGGER.info('[SelectDevice] request={:s}'.format(grpc_message_to_json_string(request)))
+ LOGGER.debug('[SelectDevice] request={:s}'.format(grpc_message_to_json_string(request)))
container_entry_uuids : Dict[str, Set[str]] = {}
container_name = 'device'
for device_id in request.device_ids.device_ids:
@@ -299,7 +253,7 @@ class MockServicerImpl_Context(ContextServiceServicer):
devices = list()
for container_name in sorted(container_entry_uuids.keys()):
entry_uuids = container_entry_uuids[container_name]
- for device in select_entries(self.database, container_name, entry_uuids):
+ for device in self.obj_db.select_entries(container_name, entry_uuids):
reply_device = Device()
reply_device.CopyFrom(device)
if exclude_endpoints: del reply_device.device_endpoints [:] # pylint: disable=no-member
@@ -308,35 +262,34 @@ class MockServicerImpl_Context(ContextServiceServicer):
devices.append(reply_device)
reply = DeviceList(devices=devices)
- LOGGER.info('[SelectDevice] reply={:s}'.format(grpc_message_to_json_string(reply)))
+ LOGGER.debug('[SelectDevice] reply={:s}'.format(grpc_message_to_json_string(reply)))
return reply
# ----- Link -------------------------------------------------------------------------------------------------------
def ListLinkIds(self, request: Empty, context : grpc.ServicerContext) -> LinkIdList:
- LOGGER.info('[ListLinkIds] request={:s}'.format(grpc_message_to_json_string(request)))
- reply = LinkIdList(link_ids=[link.link_id for link in get_entries(self.database, 'link')])
- LOGGER.info('[ListLinkIds] reply={:s}'.format(grpc_message_to_json_string(reply)))
+ LOGGER.debug('[ListLinkIds] request={:s}'.format(grpc_message_to_json_string(request)))
+ reply = LinkIdList(link_ids=[link.link_id for link in self.obj_db.get_entries('link')])
+ LOGGER.debug('[ListLinkIds] reply={:s}'.format(grpc_message_to_json_string(reply)))
return reply
def ListLinks(self, request: Empty, context : grpc.ServicerContext) -> LinkList:
- LOGGER.info('[ListLinks] request={:s}'.format(grpc_message_to_json_string(request)))
- reply = LinkList(links=get_entries(self.database, 'link'))
- LOGGER.info('[ListLinks] reply={:s}'.format(grpc_message_to_json_string(reply)))
+ LOGGER.debug('[ListLinks] request={:s}'.format(grpc_message_to_json_string(request)))
+ reply = LinkList(links=self.obj_db.get_entries('link'))
+ LOGGER.debug('[ListLinks] reply={:s}'.format(grpc_message_to_json_string(reply)))
return reply
def GetLink(self, request: LinkId, context : grpc.ServicerContext) -> Link:
- LOGGER.info('[GetLink] request={:s}'.format(grpc_message_to_json_string(request)))
- reply = get_entry(context, self.database, 'link', request.link_uuid.uuid)
- LOGGER.info('[GetLink] reply={:s}'.format(grpc_message_to_json_string(reply)))
+ LOGGER.debug('[GetLink] request={:s}'.format(grpc_message_to_json_string(request)))
+ reply = self.obj_db.get_entry('link', request.link_uuid.uuid, context)
+ LOGGER.debug('[GetLink] reply={:s}'.format(grpc_message_to_json_string(reply)))
return reply
def SetLink(self, request: Context, context : grpc.ServicerContext) -> LinkId:
- LOGGER.info('[SetLink] request={:s}'.format(grpc_message_to_json_string(request)))
+ LOGGER.debug('[SetLink] request={:s}'.format(grpc_message_to_json_string(request)))
link_uuid = request.link_id.link_uuid.uuid
- reply = self._set(request, 'link', link_uuid, 'link_id', TOPIC_LINK)
- link = get_entry(context, self.database, 'link', link_uuid)
+ reply, link = self._set(request, 'link', link_uuid, 'link_id', TOPIC_LINK)
context_topology_uuids : Set[Tuple[str, str]] = set()
context_topology_uuids.add((DEFAULT_CONTEXT_NAME, DEFAULT_TOPOLOGY_NAME))
@@ -349,20 +302,20 @@ class MockServicerImpl_Context(ContextServiceServicer):
for context_uuid,topology_uuid in context_topology_uuids:
container_name = 'topology[{:s}]'.format(str(context_uuid))
- topology = get_entry(context, self.database, container_name, topology_uuid)
- for link_id in topology.link_ids:
- if link_id.link_uuid.uuid == link_uuid: break
+ topology = self.obj_db.get_entry(container_name, topology_uuid, context)
+ for _link_id in topology.link_ids:
+ if _link_id.link_uuid.uuid == link_uuid: break
else:
# link not found, add it
topology.link_ids.add().link_uuid.uuid = link_uuid
- LOGGER.info('[SetLink] reply={:s}'.format(grpc_message_to_json_string(reply)))
+ LOGGER.debug('[SetLink] reply={:s}'.format(grpc_message_to_json_string(reply)))
return reply
def RemoveLink(self, request: LinkId, context : grpc.ServicerContext) -> Empty:
- LOGGER.info('[RemoveLink] request={:s}'.format(grpc_message_to_json_string(request)))
+ LOGGER.debug('[RemoveLink] request={:s}'.format(grpc_message_to_json_string(request)))
link_uuid = request.link_uuid.uuid
- link = get_entry(context, self.database, 'link', link_uuid)
+ link = self.obj_db.get_entry('link', link_uuid, context)
reply = self._del(request, 'link', link_uuid, 'link_id', TOPIC_LINK, context)
context_topology_uuids : Set[Tuple[str, str]] = set()
@@ -376,65 +329,65 @@ class MockServicerImpl_Context(ContextServiceServicer):
for context_uuid,topology_uuid in context_topology_uuids:
container_name = 'topology[{:s}]'.format(str(context_uuid))
- topology = get_entry(context, self.database, container_name, topology_uuid)
+ topology = self.obj_db.get_entry(container_name, topology_uuid, context)
for link_id in topology.link_ids:
if link_id.link_uuid.uuid == link_uuid:
topology.link_ids.remove(link_id)
break
- LOGGER.info('[RemoveLink] reply={:s}'.format(grpc_message_to_json_string(reply)))
+ LOGGER.debug('[RemoveLink] reply={:s}'.format(grpc_message_to_json_string(reply)))
return reply
def GetLinkEvents(self, request: Empty, context : grpc.ServicerContext) -> Iterator[LinkEvent]:
- LOGGER.info('[GetLinkEvents] request={:s}'.format(grpc_message_to_json_string(request)))
+ LOGGER.debug('[GetLinkEvents] request={:s}'.format(grpc_message_to_json_string(request)))
for message in self.msg_broker.consume({TOPIC_LINK}): yield LinkEvent(**json.loads(message.content))
# ----- Slice ------------------------------------------------------------------------------------------------------
def ListSliceIds(self, request: ContextId, context : grpc.ServicerContext) -> SliceIdList:
- LOGGER.info('[ListSliceIds] request={:s}'.format(grpc_message_to_json_string(request)))
- slices = get_entries(self.database, 'slice[{:s}]'.format(str(request.context_uuid.uuid)))
+ LOGGER.debug('[ListSliceIds] request={:s}'.format(grpc_message_to_json_string(request)))
+ slices = self.obj_db.get_entries('slice[{:s}]'.format(str(request.context_uuid.uuid)))
reply = SliceIdList(slice_ids=[slice.slice_id for slice in slices])
- LOGGER.info('[ListSliceIds] reply={:s}'.format(grpc_message_to_json_string(reply)))
+ LOGGER.debug('[ListSliceIds] reply={:s}'.format(grpc_message_to_json_string(reply)))
return reply
def ListSlices(self, request: ContextId, context : grpc.ServicerContext) -> SliceList:
- LOGGER.info('[ListSlices] request={:s}'.format(grpc_message_to_json_string(request)))
- slices = get_entries(self.database, 'slice[{:s}]'.format(str(request.context_uuid.uuid)))
+ LOGGER.debug('[ListSlices] request={:s}'.format(grpc_message_to_json_string(request)))
+ slices = self.obj_db.get_entries('slice[{:s}]'.format(str(request.context_uuid.uuid)))
reply = SliceList(slices=[slice for slice in slices])
- LOGGER.info('[ListSlices] reply={:s}'.format(grpc_message_to_json_string(reply)))
+ LOGGER.debug('[ListSlices] reply={:s}'.format(grpc_message_to_json_string(reply)))
return reply
def GetSlice(self, request: SliceId, context : grpc.ServicerContext) -> Slice:
- LOGGER.info('[GetSlice] request={:s}'.format(grpc_message_to_json_string(request)))
+ LOGGER.debug('[GetSlice] request={:s}'.format(grpc_message_to_json_string(request)))
container_name = 'slice[{:s}]'.format(str(request.context_id.context_uuid.uuid))
- reply = get_entry(context, self.database, container_name, request.slice_uuid.uuid)
- LOGGER.info('[GetSlice] reply={:s}'.format(grpc_message_to_json_string(reply)))
+ reply = self.obj_db.get_entry(container_name, request.slice_uuid.uuid, context)
+ LOGGER.debug('[GetSlice] reply={:s}'.format(grpc_message_to_json_string(reply)))
return reply
def SetSlice(self, request: Slice, context : grpc.ServicerContext) -> SliceId:
- LOGGER.info('[SetSlice] request={:s}'.format(grpc_message_to_json_string(request)))
+ LOGGER.debug('[SetSlice] request={:s}'.format(grpc_message_to_json_string(request)))
container_name = 'slice[{:s}]'.format(str(request.slice_id.context_id.context_uuid.uuid))
slice_uuid = request.slice_id.slice_uuid.uuid
- reply = self._set(request, container_name, slice_uuid, 'slice_id', TOPIC_SLICE)
- LOGGER.info('[SetSlice] reply={:s}'.format(grpc_message_to_json_string(reply)))
+ reply,_ = self._set(request, container_name, slice_uuid, 'slice_id', TOPIC_SLICE)
+ LOGGER.debug('[SetSlice] reply={:s}'.format(grpc_message_to_json_string(reply)))
return reply
def RemoveSlice(self, request: SliceId, context : grpc.ServicerContext) -> Empty:
- LOGGER.info('[RemoveSlice] request={:s}'.format(grpc_message_to_json_string(request)))
+ LOGGER.debug('[RemoveSlice] request={:s}'.format(grpc_message_to_json_string(request)))
container_name = 'slice[{:s}]'.format(str(request.context_id.context_uuid.uuid))
slice_uuid = request.slice_uuid.uuid
reply = self._del(request, container_name, slice_uuid, 'slice_id', TOPIC_SLICE, context)
- LOGGER.info('[RemoveSlice] reply={:s}'.format(grpc_message_to_json_string(reply)))
+ LOGGER.debug('[RemoveSlice] reply={:s}'.format(grpc_message_to_json_string(reply)))
return reply
def GetSliceEvents(self, request: Empty, context : grpc.ServicerContext) -> Iterator[SliceEvent]:
- LOGGER.info('[GetSliceEvents] request={:s}'.format(grpc_message_to_json_string(request)))
+ LOGGER.debug('[GetSliceEvents] request={:s}'.format(grpc_message_to_json_string(request)))
for message in self.msg_broker.consume({TOPIC_SLICE}): yield SliceEvent(**json.loads(message.content))
def SelectSlice(self, request : SliceFilter, context : grpc.ServicerContext) -> SliceList:
- LOGGER.info('[SelectSlice] request={:s}'.format(grpc_message_to_json_string(request)))
+ LOGGER.debug('[SelectSlice] request={:s}'.format(grpc_message_to_json_string(request)))
container_entry_uuids : Dict[str, Set[str]] = {}
for slice_id in request.slice_ids.slice_ids:
container_name = 'slice[{:s}]'.format(str(slice_id.context_id.context_uuid.uuid))
@@ -450,7 +403,7 @@ class MockServicerImpl_Context(ContextServiceServicer):
slices = list()
for container_name in sorted(container_entry_uuids.keys()):
entry_uuids = container_entry_uuids[container_name]
- for eslice in select_entries(self.database, container_name, entry_uuids):
+ for eslice in self.obj_db.select_entries(container_name, entry_uuids):
reply_slice = Slice()
reply_slice.CopyFrom(eslice)
if exclude_endpoint_ids: del reply_slice.service_endpoint_ids[:] # pylint: disable=no-member
@@ -461,55 +414,55 @@ class MockServicerImpl_Context(ContextServiceServicer):
slices.append(reply_slice)
reply = SliceList(slices=slices)
- LOGGER.info('[SelectSlice] reply={:s}'.format(grpc_message_to_json_string(reply)))
+ LOGGER.debug('[SelectSlice] reply={:s}'.format(grpc_message_to_json_string(reply)))
return reply
# ----- Service ----------------------------------------------------------------------------------------------------
def ListServiceIds(self, request: ContextId, context : grpc.ServicerContext) -> ServiceIdList:
- LOGGER.info('[ListServiceIds] request={:s}'.format(grpc_message_to_json_string(request)))
- services = get_entries(self.database, 'service[{:s}]'.format(str(request.context_uuid.uuid)))
+ LOGGER.debug('[ListServiceIds] request={:s}'.format(grpc_message_to_json_string(request)))
+ services = self.obj_db.get_entries('service[{:s}]'.format(str(request.context_uuid.uuid)))
reply = ServiceIdList(service_ids=[service.service_id for service in services])
- LOGGER.info('[ListServiceIds] reply={:s}'.format(grpc_message_to_json_string(reply)))
+ LOGGER.debug('[ListServiceIds] reply={:s}'.format(grpc_message_to_json_string(reply)))
return reply
def ListServices(self, request: ContextId, context : grpc.ServicerContext) -> ServiceList:
- LOGGER.info('[ListServices] request={:s}'.format(grpc_message_to_json_string(request)))
- services = get_entries(self.database, 'service[{:s}]'.format(str(request.context_uuid.uuid)))
+ LOGGER.debug('[ListServices] request={:s}'.format(grpc_message_to_json_string(request)))
+ services = self.obj_db.get_entries('service[{:s}]'.format(str(request.context_uuid.uuid)))
reply = ServiceList(services=[service for service in services])
- LOGGER.info('[ListServices] reply={:s}'.format(grpc_message_to_json_string(reply)))
+ LOGGER.debug('[ListServices] reply={:s}'.format(grpc_message_to_json_string(reply)))
return reply
def GetService(self, request: ServiceId, context : grpc.ServicerContext) -> Service:
- LOGGER.info('[GetService] request={:s}'.format(grpc_message_to_json_string(request)))
+ LOGGER.debug('[GetService] request={:s}'.format(grpc_message_to_json_string(request)))
container_name = 'service[{:s}]'.format(str(request.context_id.context_uuid.uuid))
- reply = get_entry(context, self.database, container_name, request.service_uuid.uuid)
- LOGGER.info('[GetService] reply={:s}'.format(grpc_message_to_json_string(reply)))
+ reply = self.obj_db.get_entry(container_name, request.service_uuid.uuid, context)
+ LOGGER.debug('[GetService] reply={:s}'.format(grpc_message_to_json_string(reply)))
return reply
def SetService(self, request: Service, context : grpc.ServicerContext) -> ServiceId:
- LOGGER.info('[SetService] request={:s}'.format(grpc_message_to_json_string(request)))
+ LOGGER.debug('[SetService] request={:s}'.format(grpc_message_to_json_string(request)))
container_name = 'service[{:s}]'.format(str(request.service_id.context_id.context_uuid.uuid))
service_uuid = request.service_id.service_uuid.uuid
- reply = self._set(request, container_name, service_uuid, 'service_id', TOPIC_SERVICE)
- LOGGER.info('[SetService] reply={:s}'.format(grpc_message_to_json_string(reply)))
+ reply,_ = self._set(request, container_name, service_uuid, 'service_id', TOPIC_SERVICE)
+ LOGGER.debug('[SetService] reply={:s}'.format(grpc_message_to_json_string(reply)))
return reply
def RemoveService(self, request: ServiceId, context : grpc.ServicerContext) -> Empty:
- LOGGER.info('[RemoveService] request={:s}'.format(grpc_message_to_json_string(request)))
+ LOGGER.debug('[RemoveService] request={:s}'.format(grpc_message_to_json_string(request)))
container_name = 'service[{:s}]'.format(str(request.context_id.context_uuid.uuid))
service_uuid = request.service_uuid.uuid
reply = self._del(request, container_name, service_uuid, 'service_id', TOPIC_SERVICE, context)
- LOGGER.info('[RemoveService] reply={:s}'.format(grpc_message_to_json_string(reply)))
+ LOGGER.debug('[RemoveService] reply={:s}'.format(grpc_message_to_json_string(reply)))
return reply
def GetServiceEvents(self, request: Empty, context : grpc.ServicerContext) -> Iterator[ServiceEvent]:
- LOGGER.info('[GetServiceEvents] request={:s}'.format(grpc_message_to_json_string(request)))
+ LOGGER.debug('[GetServiceEvents] request={:s}'.format(grpc_message_to_json_string(request)))
for message in self.msg_broker.consume({TOPIC_SERVICE}): yield ServiceEvent(**json.loads(message.content))
def SelectService(self, request : ServiceFilter, context : grpc.ServicerContext) -> ServiceList:
- LOGGER.info('[SelectService] request={:s}'.format(grpc_message_to_json_string(request)))
+ LOGGER.debug('[SelectService] request={:s}'.format(grpc_message_to_json_string(request)))
container_entry_uuids : Dict[str, Set[str]] = {}
for service_id in request.service_ids.service_ids:
container_name = 'service[{:s}]'.format(str(service_id.context_id.context_uuid.uuid))
@@ -523,7 +476,7 @@ class MockServicerImpl_Context(ContextServiceServicer):
services = list()
for container_name in sorted(container_entry_uuids.keys()):
entry_uuids = container_entry_uuids[container_name]
- for service in select_entries(self.database, container_name, entry_uuids):
+ for service in self.obj_db.select_entries(container_name, entry_uuids):
reply_service = Service()
reply_service.CopyFrom(service)
if exclude_endpoint_ids: del reply_service.service_endpoint_ids[:] # pylint: disable=no-member
@@ -532,54 +485,54 @@ class MockServicerImpl_Context(ContextServiceServicer):
services.append(reply_service)
reply = ServiceList(services=services)
- LOGGER.info('[SelectService] reply={:s}'.format(grpc_message_to_json_string(reply)))
+ LOGGER.debug('[SelectService] reply={:s}'.format(grpc_message_to_json_string(reply)))
return reply
# ----- Connection -------------------------------------------------------------------------------------------------
def ListConnectionIds(self, request: ServiceId, context : grpc.ServicerContext) -> ConnectionIdList:
- LOGGER.info('[ListConnectionIds] request={:s}'.format(grpc_message_to_json_string(request)))
+ LOGGER.debug('[ListConnectionIds] request={:s}'.format(grpc_message_to_json_string(request)))
container_name = 'service_connections[{:s}/{:s}]'.format(
str(request.context_id.context_uuid.uuid), str(request.service_uuid.uuid))
- reply = ConnectionIdList(connection_ids=[c.connection_id for c in get_entries(self.database, container_name)])
- LOGGER.info('[ListConnectionIds] reply={:s}'.format(grpc_message_to_json_string(reply)))
+ reply = ConnectionIdList(connection_ids=[c.connection_id for c in self.obj_db.get_entries(container_name)])
+ LOGGER.debug('[ListConnectionIds] reply={:s}'.format(grpc_message_to_json_string(reply)))
return reply
def ListConnections(self, request: ServiceId, context : grpc.ServicerContext) -> ConnectionList:
- LOGGER.info('[ListConnections] request={:s}'.format(grpc_message_to_json_string(request)))
+ LOGGER.debug('[ListConnections] request={:s}'.format(grpc_message_to_json_string(request)))
container_name = 'service_connections[{:s}/{:s}]'.format(
str(request.context_id.context_uuid.uuid), str(request.service_uuid.uuid))
- reply = ConnectionList(connections=get_entries(self.database, container_name))
- LOGGER.info('[ListConnections] reply={:s}'.format(grpc_message_to_json_string(reply)))
+ reply = ConnectionList(connections=self.obj_db.get_entries(container_name))
+ LOGGER.debug('[ListConnections] reply={:s}'.format(grpc_message_to_json_string(reply)))
return reply
def GetConnection(self, request: ConnectionId, context : grpc.ServicerContext) -> Connection:
- LOGGER.info('[GetConnection] request={:s}'.format(grpc_message_to_json_string(request)))
- reply = get_entry(context, self.database, 'connection', request.connection_uuid.uuid)
- LOGGER.info('[GetConnection] reply={:s}'.format(grpc_message_to_json_string(reply)))
+ LOGGER.debug('[GetConnection] request={:s}'.format(grpc_message_to_json_string(request)))
+ reply = self.obj_db.get_entry('connection', request.connection_uuid.uuid, context)
+ LOGGER.debug('[GetConnection] reply={:s}'.format(grpc_message_to_json_string(reply)))
return reply
def SetConnection(self, request: Connection, context : grpc.ServicerContext) -> ConnectionId:
- LOGGER.info('[SetConnection] request={:s}'.format(grpc_message_to_json_string(request)))
+ LOGGER.debug('[SetConnection] request={:s}'.format(grpc_message_to_json_string(request)))
container_name = 'service_connection[{:s}/{:s}]'.format(
str(request.service_id.context_id.context_uuid.uuid), str(request.service_id.service_uuid.uuid))
connection_uuid = request.connection_id.connection_uuid.uuid
- set_entry(self.database, container_name, connection_uuid, request)
- reply = self._set(request, 'connection', connection_uuid, 'connection_id', TOPIC_CONNECTION)
- LOGGER.info('[SetConnection] reply={:s}'.format(grpc_message_to_json_string(reply)))
+ self.obj_db.set_entry(container_name, connection_uuid, request)
+ reply,_ = self._set(request, 'connection', connection_uuid, 'connection_id', TOPIC_CONNECTION)
+ LOGGER.debug('[SetConnection] reply={:s}'.format(grpc_message_to_json_string(reply)))
return reply
def RemoveConnection(self, request: ConnectionId, context : grpc.ServicerContext) -> Empty:
- LOGGER.info('[RemoveConnection] request={:s}'.format(grpc_message_to_json_string(request)))
- connection = get_entry(context, self.database, 'connection', request.connection_uuid.uuid)
+ LOGGER.debug('[RemoveConnection] request={:s}'.format(grpc_message_to_json_string(request)))
+ connection = self.obj_db.get_entry('connection', request.connection_uuid.uuid, context)
container_name = 'service_connection[{:s}/{:s}]'.format(
str(connection.service_id.context_id.context_uuid.uuid), str(connection.service_id.service_uuid.uuid))
connection_uuid = request.connection_uuid.uuid
- del_entry(context, self.database, container_name, connection_uuid)
+ self.obj_db.del_entry(container_name, connection_uuid, context)
reply = self._del(request, 'connection', connection_uuid, 'connection_id', TOPIC_CONNECTION, context)
- LOGGER.info('[RemoveConnection] reply={:s}'.format(grpc_message_to_json_string(reply)))
+ LOGGER.debug('[RemoveConnection] reply={:s}'.format(grpc_message_to_json_string(reply)))
return reply
def GetConnectionEvents(self, request: Empty, context : grpc.ServicerContext) -> Iterator[ConnectionEvent]:
- LOGGER.info('[GetConnectionEvents] request={:s}'.format(grpc_message_to_json_string(request)))
+ LOGGER.debug('[GetConnectionEvents] request={:s}'.format(grpc_message_to_json_string(request)))
for message in self.msg_broker.consume({TOPIC_CONNECTION}): yield ConnectionEvent(**json.loads(message.content))
diff --git a/src/common/tests/MockServicerImpl_Monitoring.py b/src/common/tests/MockServicerImpl_Monitoring.py
index 419e488a1f9ffef4f1c6f3a3127b1049d2b25706..4aadb8e5e20575321df2003c69a5ab9fe2390af8 100644
--- a/src/common/tests/MockServicerImpl_Monitoring.py
+++ b/src/common/tests/MockServicerImpl_Monitoring.py
@@ -12,9 +12,9 @@
# See the License for the specific language governing permissions and
# limitations under the License.
-import enum, grpc, logging, pandas
+import enum, grpc, logging
from queue import Queue
-from typing import Optional
+from typing import Any, Optional
from common.proto.context_pb2 import Empty
from common.proto.monitoring_pb2 import Kpi, KpiDescriptor, KpiDescriptorList, KpiId, KpiQuery, RawKpiTable
from common.proto.monitoring_pb2_grpc import MonitoringServiceServicer
@@ -31,61 +31,61 @@ class MockServicerImpl_Monitoring(MonitoringServiceServicer):
def __init__(
self, queue_samples : Optional[Queue] = None
) -> None:
- LOGGER.info('[__init__] Creating Servicer...')
+ LOGGER.debug('[__init__] Creating Servicer...')
if queue_samples is None: queue_samples = Queue()
self.queue_samples = queue_samples
self.obj_db = InMemoryObjectDatabase()
self.ts_db = InMemoryTimeSeriesDatabase()
- LOGGER.info('[__init__] Servicer Created')
+ LOGGER.debug('[__init__] Servicer Created')
# ----- Common -----------------------------------------------------------------------------------------------------
- def _set(self, container_name, entry_uuid, entry_id_field_name, entry):
+ def _set(self, container_name, entry_uuid, entry_id_field_name, entry) -> Any:
entry = self.obj_db.set_entry(container_name, entry_uuid, entry)
return getattr(entry, entry_id_field_name)
- def _del(self, container_name, entry_uuid, grpc_context):
+ def _del(self, container_name, entry_uuid, grpc_context) -> Empty:
self.obj_db.del_entry(container_name, entry_uuid, grpc_context)
return Empty()
# ----- KPI Descriptor ---------------------------------------------------------------------------------------------
def GetKpiDescriptorList(self, request : Empty, context : grpc.ServicerContext) -> KpiDescriptorList:
- LOGGER.info('[GetKpiDescriptorList] request={:s}'.format(grpc_message_to_json_string(request)))
+ LOGGER.debug('[GetKpiDescriptorList] request={:s}'.format(grpc_message_to_json_string(request)))
kpi_descriptor_list = self.obj_db.get_entries(IMDB_ContainersEnum.KPI_DESCRIPTORS.value)
reply = KpiDescriptorList(kpi_descriptor_list=kpi_descriptor_list)
- LOGGER.info('[GetKpiDescriptorList] reply={:s}'.format(grpc_message_to_json_string(reply)))
+ LOGGER.debug('[GetKpiDescriptorList] reply={:s}'.format(grpc_message_to_json_string(reply)))
return reply
def GetKpiDescriptor(self, request : KpiId, context : grpc.ServicerContext) -> KpiDescriptor:
- LOGGER.info('[GetKpiDescriptor] request={:s}'.format(grpc_message_to_json_string(request)))
+ LOGGER.debug('[GetKpiDescriptor] request={:s}'.format(grpc_message_to_json_string(request)))
reply = self.obj_db.get_entry(IMDB_ContainersEnum.KPI_DESCRIPTORS.value, request.kpi_id.uuid, context)
- LOGGER.info('[GetKpiDescriptor] reply={:s}'.format(grpc_message_to_json_string(reply)))
+ LOGGER.debug('[GetKpiDescriptor] reply={:s}'.format(grpc_message_to_json_string(reply)))
return reply
def SetKpi(self, request : KpiDescriptor, context : grpc.ServicerContext) -> KpiId:
- LOGGER.info('[SetKpi] request={:s}'.format(grpc_message_to_json_string(request)))
+ LOGGER.debug('[SetKpi] request={:s}'.format(grpc_message_to_json_string(request)))
reply = self._set(IMDB_ContainersEnum.KPI_DESCRIPTORS.value, request.kpi_id.kpi_id.uuid, 'kpi_id', request)
- LOGGER.info('[SetKpi] reply={:s}'.format(grpc_message_to_json_string(reply)))
+ LOGGER.debug('[SetKpi] reply={:s}'.format(grpc_message_to_json_string(reply)))
return reply
def DeleteKpi(self, request : KpiId, context : grpc.ServicerContext) -> Empty:
- LOGGER.info('[DeleteKpi] request={:s}'.format(grpc_message_to_json_string(request)))
+ LOGGER.debug('[DeleteKpi] request={:s}'.format(grpc_message_to_json_string(request)))
reply = self._del(IMDB_ContainersEnum.KPI_DESCRIPTORS.value, request.kpi_id.kpi_id.uuid, context)
- LOGGER.info('[DeleteKpi] reply={:s}'.format(grpc_message_to_json_string(reply)))
+ LOGGER.debug('[DeleteKpi] reply={:s}'.format(grpc_message_to_json_string(reply)))
return reply
# ----- KPI Sample -------------------------------------------------------------------------------------------------
def IncludeKpi(self, request : Kpi, context : grpc.ServicerContext) -> Empty:
- LOGGER.info('[IncludeKpi] request={:s}'.format(grpc_message_to_json_string(request)))
+ LOGGER.debug('[IncludeKpi] request={:s}'.format(grpc_message_to_json_string(request)))
self.queue_samples.put(request)
reply = Empty()
- LOGGER.info('[IncludeKpi] reply={:s}'.format(grpc_message_to_json_string(reply)))
+ LOGGER.debug('[IncludeKpi] reply={:s}'.format(grpc_message_to_json_string(reply)))
return reply
def QueryKpiData(self, request : KpiQuery, context : grpc.ServicerContext) -> RawKpiTable:
- LOGGER.info('[QueryKpiData] request={:s}'.format(grpc_message_to_json_string(request)))
+ LOGGER.debug('[QueryKpiData] request={:s}'.format(grpc_message_to_json_string(request)))
# TODO: add filters for request.monitoring_window_s
# TODO: add filters for request.last_n_samples
kpi_uuids = [kpi_id.kpi_id.uuid for kpi_id in request.kpi_ids]
@@ -97,7 +97,7 @@ class MockServicerImpl_Monitoring(MonitoringServiceServicer):
if end_timestamp <= 0: end_timestamp = None
df_samples = self.ts_db.filter(kpi_uuids, start_timestamp=start_timestamp, end_timestamp=end_timestamp)
- LOGGER.info('[QueryKpiData] df_samples={:s}'.format(df_samples.to_string()))
+ #LOGGER.debug('[QueryKpiData] df_samples={:s}'.format(df_samples.to_string()))
reply = RawKpiTable()
kpi_uuid__to__raw_kpi_list = dict()
@@ -114,5 +114,5 @@ class MockServicerImpl_Monitoring(MonitoringServiceServicer):
raw_kpi.timestamp.timestamp = df_sample.timestamp.timestamp()
raw_kpi.kpi_value.floatVal = df_sample.value
- LOGGER.info('[QueryKpiData] reply={:s}'.format(grpc_message_to_json_string(reply)))
+ LOGGER.debug('[QueryKpiData] reply={:s}'.format(grpc_message_to_json_string(reply)))
return reply