diff --git a/src/common/tests/MockServicerImpl_Context.py b/src/common/tests/MockServicerImpl_Context.py
index 6e749e17c156a41965667238dbd98189c71f9acb..b50b044a12548e11888468e7acbfad9280491022 100644
--- a/src/common/tests/MockServicerImpl_Context.py
+++ b/src/common/tests/MockServicerImpl_Context.py
@@ -68,58 +68,58 @@ class MockServicerImpl_Context(ContextServiceServicer):
# ----- Context ----------------------------------------------------------------------------------------------------
- def ListContextIds(self, request: Empty, context : grpc.ServicerContext) -> ContextIdList:
+ def ListContextIds(self, request : Empty, context : grpc.ServicerContext) -> ContextIdList:
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:
+ def ListContexts(self, request : Empty, context : grpc.ServicerContext) -> ContextList:
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:
+ def GetContext(self, request : ContextId, context : grpc.ServicerContext) -> Context:
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:
+ def SetContext(self, request : Context, context : grpc.ServicerContext) -> ContextId:
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:
+ def RemoveContext(self, request : ContextId, context : grpc.ServicerContext) -> Empty:
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.debug('[RemoveContext] reply={:s}'.format(grpc_message_to_json_string(reply)))
return reply
- def GetContextEvents(self, request: Empty, context : grpc.ServicerContext) -> Iterator[ContextEvent]:
+ def GetContextEvents(self, request : Empty, context : grpc.ServicerContext) -> Iterator[ContextEvent]:
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:
+ def ListTopologyIds(self, request : ContextId, context : grpc.ServicerContext) -> TopologyIdList:
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.debug('[ListTopologyIds] reply={:s}'.format(grpc_message_to_json_string(reply)))
return reply
- def ListTopologies(self, request: ContextId, context : grpc.ServicerContext) -> TopologyList:
+ def ListTopologies(self, request : ContextId, context : grpc.ServicerContext) -> TopologyList:
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.debug('[ListTopologies] reply={:s}'.format(grpc_message_to_json_string(reply)))
return reply
- def GetTopology(self, request: TopologyId, context : grpc.ServicerContext) -> Topology:
+ def GetTopology(self, request : TopologyId, context : grpc.ServicerContext) -> Topology:
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 = self.obj_db.get_entry(container_name, request.topology_uuid.uuid, context)
@@ -149,7 +149,7 @@ class MockServicerImpl_Context(ContextServiceServicer):
LOGGER.debug('[GetTopologyDetails] reply={:s}'.format(grpc_message_to_json_string(reply)))
return reply
- def SetTopology(self, request: Topology, context : grpc.ServicerContext) -> TopologyId:
+ def SetTopology(self, request : Topology, context : grpc.ServicerContext) -> TopologyId:
LOGGER.debug('[SetTopology] request={:s}'.format(grpc_message_to_json_string(request)))
context_uuid = str(request.topology_id.context_id.context_uuid.uuid)
container_name = 'topology[{:s}]'.format(context_uuid)
@@ -192,7 +192,7 @@ class MockServicerImpl_Context(ContextServiceServicer):
LOGGER.debug('[SetTopology] reply={:s}'.format(grpc_message_to_json_string(reply)))
return reply
- def RemoveTopology(self, request: TopologyId, context : grpc.ServicerContext) -> Empty:
+ def RemoveTopology(self, request : TopologyId, context : grpc.ServicerContext) -> Empty:
LOGGER.debug('[RemoveTopology] request={:s}'.format(grpc_message_to_json_string(request)))
context_uuid = str(request.context_id.context_uuid.uuid)
container_name = 'topology[{:s}]'.format(context_uuid)
@@ -208,32 +208,32 @@ class MockServicerImpl_Context(ContextServiceServicer):
LOGGER.debug('[RemoveTopology] reply={:s}'.format(grpc_message_to_json_string(reply)))
return reply
- def GetTopologyEvents(self, request: Empty, context : grpc.ServicerContext) -> Iterator[TopologyEvent]:
+ def GetTopologyEvents(self, request : Empty, context : grpc.ServicerContext) -> Iterator[TopologyEvent]:
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:
+ def ListDeviceIds(self, request : Empty, context : grpc.ServicerContext) -> DeviceIdList:
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:
+ def ListDevices(self, request : Empty, context : grpc.ServicerContext) -> DeviceList:
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:
+ def GetDevice(self, request : DeviceId, context : grpc.ServicerContext) -> Device:
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:
+ def SetDevice(self, request : Context, context : grpc.ServicerContext) -> DeviceId:
LOGGER.debug('[SetDevice] request={:s}'.format(grpc_message_to_json_string(request)))
device_uuid = request.device_id.device_uuid.uuid
reply, device = self._set(request, 'device', device_uuid, 'device_id', TOPIC_DEVICE)
@@ -259,7 +259,7 @@ class MockServicerImpl_Context(ContextServiceServicer):
LOGGER.debug('[SetDevice] reply={:s}'.format(grpc_message_to_json_string(reply)))
return reply
- def RemoveDevice(self, request: DeviceId, context : grpc.ServicerContext) -> Empty:
+ def RemoveDevice(self, request : DeviceId, context : grpc.ServicerContext) -> Empty:
LOGGER.debug('[RemoveDevice] request={:s}'.format(grpc_message_to_json_string(request)))
device_uuid = request.device_uuid.uuid
device = self.obj_db.get_entry('device', device_uuid, context)
@@ -285,7 +285,7 @@ class MockServicerImpl_Context(ContextServiceServicer):
LOGGER.debug('[RemoveDevice] reply={:s}'.format(grpc_message_to_json_string(reply)))
return reply
- def GetDeviceEvents(self, request: Empty, context : grpc.ServicerContext) -> Iterator[DeviceEvent]:
+ def GetDeviceEvents(self, request : Empty, context : grpc.ServicerContext) -> Iterator[DeviceEvent]:
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))
@@ -319,25 +319,25 @@ class MockServicerImpl_Context(ContextServiceServicer):
# ----- Link -------------------------------------------------------------------------------------------------------
- def ListLinkIds(self, request: Empty, context : grpc.ServicerContext) -> LinkIdList:
+ def ListLinkIds(self, request : Empty, context : grpc.ServicerContext) -> LinkIdList:
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:
+ def ListLinks(self, request : Empty, context : grpc.ServicerContext) -> LinkList:
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:
+ def GetLink(self, request : LinkId, context : grpc.ServicerContext) -> Link:
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:
+ def SetLink(self, request : Context, context : grpc.ServicerContext) -> LinkId:
LOGGER.debug('[SetLink] request={:s}'.format(grpc_message_to_json_string(request)))
link_uuid = request.link_id.link_uuid.uuid
reply, link = self._set(request, 'link', link_uuid, 'link_id', TOPIC_LINK)
@@ -363,7 +363,7 @@ class MockServicerImpl_Context(ContextServiceServicer):
LOGGER.debug('[SetLink] reply={:s}'.format(grpc_message_to_json_string(reply)))
return reply
- def RemoveLink(self, request: LinkId, context : grpc.ServicerContext) -> Empty:
+ def RemoveLink(self, request : LinkId, context : grpc.ServicerContext) -> Empty:
LOGGER.debug('[RemoveLink] request={:s}'.format(grpc_message_to_json_string(request)))
link_uuid = request.link_uuid.uuid
link = self.obj_db.get_entry('link', link_uuid, context)
@@ -389,35 +389,35 @@ class MockServicerImpl_Context(ContextServiceServicer):
LOGGER.debug('[RemoveLink] reply={:s}'.format(grpc_message_to_json_string(reply)))
return reply
- def GetLinkEvents(self, request: Empty, context : grpc.ServicerContext) -> Iterator[LinkEvent]:
+ def GetLinkEvents(self, request : Empty, context : grpc.ServicerContext) -> Iterator[LinkEvent]:
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:
+ def ListSliceIds(self, request : ContextId, context : grpc.ServicerContext) -> SliceIdList:
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.debug('[ListSliceIds] reply={:s}'.format(grpc_message_to_json_string(reply)))
return reply
- def ListSlices(self, request: ContextId, context : grpc.ServicerContext) -> SliceList:
+ def ListSlices(self, request : ContextId, context : grpc.ServicerContext) -> SliceList:
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.debug('[ListSlices] reply={:s}'.format(grpc_message_to_json_string(reply)))
return reply
- def GetSlice(self, request: SliceId, context : grpc.ServicerContext) -> Slice:
+ def GetSlice(self, request : SliceId, context : grpc.ServicerContext) -> Slice:
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 = 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:
+ def SetSlice(self, request : Slice, context : grpc.ServicerContext) -> SliceId:
LOGGER.debug('[SetSlice] request={:s}'.format(grpc_message_to_json_string(request)))
context_uuid = str(request.slice_id.context_id.context_uuid.uuid)
container_name = 'slice[{:s}]'.format(context_uuid)
@@ -434,7 +434,7 @@ class MockServicerImpl_Context(ContextServiceServicer):
LOGGER.debug('[SetSlice] reply={:s}'.format(grpc_message_to_json_string(reply)))
return reply
- def RemoveSlice(self, request: SliceId, context : grpc.ServicerContext) -> Empty:
+ def RemoveSlice(self, request : SliceId, context : grpc.ServicerContext) -> Empty:
LOGGER.debug('[RemoveSlice] request={:s}'.format(grpc_message_to_json_string(request)))
context_uuid = str(request.context_id.context_uuid.uuid)
container_name = 'slice[{:s}]'.format(context_uuid)
@@ -450,7 +450,7 @@ class MockServicerImpl_Context(ContextServiceServicer):
LOGGER.debug('[RemoveSlice] reply={:s}'.format(grpc_message_to_json_string(reply)))
return reply
- def GetSliceEvents(self, request: Empty, context : grpc.ServicerContext) -> Iterator[SliceEvent]:
+ def GetSliceEvents(self, request : Empty, context : grpc.ServicerContext) -> Iterator[SliceEvent]:
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))
@@ -488,28 +488,28 @@ class MockServicerImpl_Context(ContextServiceServicer):
# ----- Service ----------------------------------------------------------------------------------------------------
- def ListServiceIds(self, request: ContextId, context : grpc.ServicerContext) -> ServiceIdList:
+ def ListServiceIds(self, request : ContextId, context : grpc.ServicerContext) -> ServiceIdList:
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.debug('[ListServiceIds] reply={:s}'.format(grpc_message_to_json_string(reply)))
return reply
- def ListServices(self, request: ContextId, context : grpc.ServicerContext) -> ServiceList:
+ def ListServices(self, request : ContextId, context : grpc.ServicerContext) -> ServiceList:
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.debug('[ListServices] reply={:s}'.format(grpc_message_to_json_string(reply)))
return reply
- def GetService(self, request: ServiceId, context : grpc.ServicerContext) -> Service:
+ def GetService(self, request : ServiceId, context : grpc.ServicerContext) -> Service:
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 = 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:
+ def SetService(self, request : Service, context : grpc.ServicerContext) -> ServiceId:
LOGGER.debug('[SetService] request={:s}'.format(grpc_message_to_json_string(request)))
context_uuid = str(request.service_id.context_id.context_uuid.uuid)
container_name = 'service[{:s}]'.format(context_uuid)
@@ -526,7 +526,7 @@ class MockServicerImpl_Context(ContextServiceServicer):
LOGGER.debug('[SetService] reply={:s}'.format(grpc_message_to_json_string(reply)))
return reply
- def RemoveService(self, request: ServiceId, context : grpc.ServicerContext) -> Empty:
+ def RemoveService(self, request : ServiceId, context : grpc.ServicerContext) -> Empty:
LOGGER.debug('[RemoveService] request={:s}'.format(grpc_message_to_json_string(request)))
context_uuid = str(request.context_id.context_uuid.uuid)
container_name = 'service[{:s}]'.format(context_uuid)
@@ -542,7 +542,7 @@ class MockServicerImpl_Context(ContextServiceServicer):
LOGGER.debug('[RemoveService] reply={:s}'.format(grpc_message_to_json_string(reply)))
return reply
- def GetServiceEvents(self, request: Empty, context : grpc.ServicerContext) -> Iterator[ServiceEvent]:
+ def GetServiceEvents(self, request : Empty, context : grpc.ServicerContext) -> Iterator[ServiceEvent]:
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))
@@ -575,7 +575,7 @@ class MockServicerImpl_Context(ContextServiceServicer):
# ----- Connection -------------------------------------------------------------------------------------------------
- def ListConnectionIds(self, request: ServiceId, context : grpc.ServicerContext) -> ConnectionIdList:
+ def ListConnectionIds(self, request : ServiceId, context : grpc.ServicerContext) -> ConnectionIdList:
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))
@@ -583,7 +583,7 @@ class MockServicerImpl_Context(ContextServiceServicer):
LOGGER.debug('[ListConnectionIds] reply={:s}'.format(grpc_message_to_json_string(reply)))
return reply
- def ListConnections(self, request: ServiceId, context : grpc.ServicerContext) -> ConnectionList:
+ def ListConnections(self, request : ServiceId, context : grpc.ServicerContext) -> ConnectionList:
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))
@@ -591,13 +591,13 @@ class MockServicerImpl_Context(ContextServiceServicer):
LOGGER.debug('[ListConnections] reply={:s}'.format(grpc_message_to_json_string(reply)))
return reply
- def GetConnection(self, request: ConnectionId, context : grpc.ServicerContext) -> Connection:
+ def GetConnection(self, request : ConnectionId, context : grpc.ServicerContext) -> Connection:
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:
+ def SetConnection(self, request : Connection, context : grpc.ServicerContext) -> ConnectionId:
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))
@@ -607,7 +607,7 @@ class MockServicerImpl_Context(ContextServiceServicer):
LOGGER.debug('[SetConnection] reply={:s}'.format(grpc_message_to_json_string(reply)))
return reply
- def RemoveConnection(self, request: ConnectionId, context : grpc.ServicerContext) -> Empty:
+ def RemoveConnection(self, request : ConnectionId, context : grpc.ServicerContext) -> Empty:
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(
@@ -618,84 +618,45 @@ class MockServicerImpl_Context(ContextServiceServicer):
LOGGER.debug('[RemoveConnection] reply={:s}'.format(grpc_message_to_json_string(reply)))
return reply
- def GetConnectionEvents(self, request: Empty, context : grpc.ServicerContext) -> Iterator[ConnectionEvent]:
+ def GetConnectionEvents(self, request : Empty, context : grpc.ServicerContext) -> Iterator[ConnectionEvent]:
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))
- def ListPolicyRuleIds(self, request: Empty, context: grpc.ServicerContext):
- LOGGER.debug(
- "[ListPolicyRuleIds] request={:s}".format(
- grpc_message_to_json_string(request)
- )
- )
- reply = PolicyRuleIdList(
- policyRuleIdList=[
- getattr(
- policy_rule, policy_rule.WhichOneof("policy_rule")
- ).policyRuleBasic.policyRuleId
- for policy_rule in self.obj_db.get_entries("policy")
- ]
- )
- LOGGER.debug(
- "[ListPolicyRuleIds] reply={:s}".format(grpc_message_to_json_string(reply))
- )
- return reply
-
- def ListPolicyRules(self, request: Empty, context: grpc.ServicerContext):
- LOGGER.debug(
- "[ListPolicyRules] request={:s}".format(
- grpc_message_to_json_string(request)
- )
- )
- reply = PolicyRuleList(policyRules=self.obj_db.get_entries("policy"))
- LOGGER.debug(
- "[ListPolicyRules] reply={:s}".format(grpc_message_to_json_string(reply))
- )
- return reply
-
- def GetPolicyRule(self, request: PolicyRuleId, context: grpc.ServicerContext):
- LOGGER.debug(
- "[GetPolicyRule] request={:s}".format(grpc_message_to_json_string(request))
- )
- reply = self.obj_db.get_entry("policy_rule", request.uuid.uuid, context)
- LOGGER.debug(
- "[GetPolicyRule] reply={:s}".format(grpc_message_to_json_string(reply))
- )
- return reply
-
- def SetPolicyRule(self, request: PolicyRule, context: grpc.ServicerContext):
- LOGGER.debug(
- "[SetPolicyRule] request={:s}".format(grpc_message_to_json_string(request))
- )
- policy_type = request.WhichOneof("policy_rule")
- reply, _ = self._set(
- request,
- "policy",
- getattr(request, policy_type).policyRuleBasic.policyRuleId.uuid.uuid,
- f"{policy_type}.policyRuleBasic.policyRuleId",
- TOPIC_POLICY,
- )
- LOGGER.debug(
- "[SetPolicyRule] reply={:s}".format(grpc_message_to_json_string(reply))
- )
- return reply
-
- def RemovePolicyRule(self, request: PolicyRuleId, context: grpc.ServicerContext):
- LOGGER.debug(
- "[RemovePolicyRule] request={:s}".format(
- grpc_message_to_json_string(request)
- )
- )
- policy_type = request.WhichOneof("policy_rule")
- reply = self._del(
- request,
- "policy",
- getattr(request, policy_type).policyRuleBasic.policyRuleId.uuid.uuid,
- f"{policy_type}.policyRuleBasic.policyRuleId",
- TOPIC_CONTEXT,
- context,
- )
- LOGGER.debug(
- "[RemovePolicyRule] reply={:s}".format(grpc_message_to_json_string(reply))
- )
- return reply
\ No newline at end of file
+ def ListPolicyRuleIds(self, request : Empty, context : grpc.ServicerContext):
+ LOGGER.debug('[ListPolicyRuleIds] request={:s}'.format(grpc_message_to_json_string(request)))
+ reply = PolicyRuleIdList(policyRuleIdList=[
+ getattr(policy_rule, policy_rule.WhichOneof('policy_rule')).policyRuleBasic.policyRuleId
+ for policy_rule in self.obj_db.get_entries('policy')
+ ])
+ LOGGER.debug('[ListPolicyRuleIds] reply={:s}'.format(grpc_message_to_json_string(reply)))
+ return reply
+
+ def ListPolicyRules(self, request : Empty, context : grpc.ServicerContext):
+ LOGGER.debug('[ListPolicyRules] request={:s}'.format(grpc_message_to_json_string(request)))
+ reply = PolicyRuleList(policyRules=self.obj_db.get_entries('policy'))
+ LOGGER.debug('[ListPolicyRules] reply={:s}'.format(grpc_message_to_json_string(reply)))
+ return reply
+
+ def GetPolicyRule(self, request : PolicyRuleId, context : grpc.ServicerContext):
+ LOGGER.debug('[GetPolicyRule] request={:s}'.format(grpc_message_to_json_string(request)))
+ reply = self.obj_db.get_entry('policy_rule', request.uuid.uuid, context)
+ LOGGER.debug('[GetPolicyRule] reply={:s}'.format(grpc_message_to_json_string(reply)))
+ return reply
+
+ def SetPolicyRule(self, request : PolicyRule, context : grpc.ServicerContext):
+ LOGGER.debug('[SetPolicyRule] request={:s}'.format(grpc_message_to_json_string(request)))
+ policy_type = request.WhichOneof('policy_rule')
+ policy_uuid = getattr(request, policy_type).policyRuleBasic.policyRuleId.uuid.uuid
+ rule_id_field = '{:s}.policyRuleBasic.policyRuleId'.format(policy_type)
+ reply, _ = self._set(request, 'policy', policy_uuid, rule_id_field, TOPIC_POLICY)
+ LOGGER.debug('[SetPolicyRule] reply={:s}'.format(grpc_message_to_json_string(reply)))
+ return reply
+
+ def RemovePolicyRule(self, request : PolicyRuleId, context : grpc.ServicerContext):
+ LOGGER.debug('[RemovePolicyRule] request={:s}'.format(grpc_message_to_json_string(request)))
+ policy_type = request.WhichOneof('policy_rule')
+ policy_uuid = getattr(request, policy_type).policyRuleBasic.policyRuleId.uuid.uuid
+ rule_id_field = '{:s}.policyRuleBasic.policyRuleId'.format(policy_type)
+ reply = self._del(request, 'policy', policy_uuid, rule_id_field, TOPIC_CONTEXT, context)
+ LOGGER.debug('[RemovePolicyRule] reply={:s}'.format(grpc_message_to_json_string(reply)))
+ return reply