Common - Tests - Mock Context: (9cc2fb42) · Commits · TFS / controller

src/common/tests/MockServicerImpl_Context.py

+80 −119

Original line number	Diff line number	Diff line
		@@ -623,79 +623,40 @@ class MockServicerImpl_Context(ContextServiceServicer):
		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))
		)
		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))
		)
		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))
		)
		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))
		)
		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")
		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))
		)
		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