Loading src/common/tests/MockServicerImpl_Context.py +85 −1 Original line number Diff line number Diff line Loading @@ -25,12 +25,18 @@ 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.proto.policy_pb2 import ( PolicyRule, PolicyRuleId, PolicyRuleIdList, PolicyRuleList, ) from common.tools.grpc.Tools import grpc_message_to_json, grpc_message_to_json_string from common.tools.object_factory.Device import json_device_id from common.tools.object_factory.Link import json_link_id from .InMemoryObjectDatabase import InMemoryObjectDatabase from .MockMessageBroker import ( TOPIC_CONNECTION, TOPIC_CONTEXT, TOPIC_DEVICE, TOPIC_LINK, TOPIC_SERVICE, TOPIC_SLICE, TOPIC_TOPOLOGY, TOPIC_CONNECTION, TOPIC_CONTEXT, TOPIC_DEVICE, TOPIC_LINK, TOPIC_SERVICE, TOPIC_SLICE, TOPIC_TOPOLOGY, TOPIC_POLICY, MockMessageBroker, notify_event) LOGGER = logging.getLogger(__name__) Loading Loading @@ -615,3 +621,81 @@ class MockServicerImpl_Context(ContextServiceServicer): 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 src/common/type_checkers/Assertions.py +15 −1 Original line number Diff line number Diff line Loading @@ -192,6 +192,15 @@ def validate_connection_id(message): assert 'connection_uuid' in message validate_uuid(message['connection_uuid']) def validate_slice_id(message, context_uuid = None): assert isinstance(message, dict) assert len(message.keys()) == 2 assert 'context_id' in message validate_context_id(message['context_id']) if context_uuid is not None: assert message['context_id']['context_uuid']['uuid'] == context_uuid assert 'slice_uuid' in message validate_uuid(message['slice_uuid']) # ----- Lists of Identifiers ------------------------------------------------------------------------------------------- Loading Loading @@ -242,7 +251,9 @@ def validate_connection_ids(message): def validate_context(message): assert isinstance(message, dict) assert len(message.keys()) == 3 assert len(message.keys()) == 5 assert 'name' in message assert isinstance(message['name'], str) assert 'context_id' in message validate_context_id(message['context_id']) context_uuid = message['context_id']['context_uuid']['uuid'] Loading @@ -252,6 +263,9 @@ def validate_context(message): assert 'topology_ids' in message assert isinstance(message['topology_ids'], list) for topology_id in message['topology_ids']: validate_topology_id(topology_id, context_uuid=context_uuid) assert 'slice_ids' in message assert isinstance(message['slice_ids'], list) for slice_id in message['slice_ids']: validate_slice_id(slice_id, context_uuid=context_uuid) def validate_service_state(message): assert isinstance(message, dict) Loading src/nbi/tests/test_debug_api.py +164 −162 Original line number Diff line number Diff line Loading @@ -13,6 +13,7 @@ # limitations under the License. import logging, os, pytest, time, urllib import pprint from common.Constants import DEFAULT_CONTEXT_NAME, DEFAULT_TOPOLOGY_NAME, ServiceNameEnum from common.proto.context_pb2 import Connection, Context, Device, Link, Service, Slice, Topology from common.proto.policy_pb2 import PolicyRuleIdList, PolicyRuleId, PolicyRuleList, PolicyRule Loading @@ -25,17 +26,35 @@ from common.type_checkers.Assertions import ( validate_contexts, validate_device, validate_device_ids, validate_devices, validate_link, validate_link_ids, validate_links, validate_service, validate_service_ids, validate_services, validate_topologies, validate_topology, validate_topology_ids) from context.client.ContextClient import ContextClient from nbi.tests.PrepareTestScenario import do_rest_get_request from nbi.tests.PrepareTestScenario import do_rest_get_request, context_client, nbi_service_rest from ..service.rest_server.RestServer import RestServer from .MockService_Dependencies import MockService_Dependencies from .Objects import ( CONNECTION_R1_R3, CONNECTION_R1_R3_ID, CONNECTION_R1_R3_UUID, CONTEXT, CONTEXT_ID, DEVICE_R1, DEVICE_R1_ID, DEVICE_R1_UUID, DEVICE_R2, DEVICE_R2_ID, DEVICE_R2_UUID, DEVICE_R3, DEVICE_R3_ID, DEVICE_R3_UUID, LINK_R1_R2, LINK_R1_R2_ID, LINK_R1_R2_UUID, SERVICE_R1_R2, SERVICE_R1_R2_ID, SERVICE_R1_R2_UUID, SERVICE_R1_R3, SERVICE_R1_R3_ID, SERVICE_R1_R3_UUID, SERVICE_R2_R3, SERVICE_R2_R3_ID, SERVICE_R2_R3_UUID, SLICE_R1_R3, TOPOLOGY, TOPOLOGY_ID, POLICY_RULE, POLICY_RULE_ID, POLICY_RULE_UUID from context.tests.Objects import ( CONNECTION_R1_R3, CONTEXT, DEVICE_R1, DEVICE_R2, DEVICE_R3, LINK_R1_R2, LINK_R1_R3, LINK_R2_R3, SERVICE_R1_R2, SERVICE_R1_R3, SERVICE_R2_R3, SLICE_R1_R3, TOPOLOGY ) from ..service.rest_server.nbi_plugins.debug_api import URL_PREFIX # from context.client.ContextClient import ContextClient # from ..service.rest_server.nbi_plugins.debug_api import RESOURCES # connection_r1_r3_uuid = '' # policyrule_uuid = '' # slice_r1_r3_uuid = '' # service_r1_r2_uuid = '' # service_r1_r3_uuid = '' # link_r1_r2_uuid = '' # device_r1_uuid = '' def pytest_namespace(): return { 'connection_r1_r3_uuid' : '', 'policyrule_uuid' : '', 'slice_r1_r3_uuid' : '', 'service_r1_r2_uuid' : '', 'service_r1_r3_uuid' : '', 'link_r1_r2_uuid' : '', 'device_r1_uuid' : '', } @pytest.fixture(scope='session') def mock_service(): Loading @@ -61,160 +80,143 @@ os.environ[get_env_var_name(ServiceNameEnum.CONTEXT, ENVVAR_SUFIX_SERVICE_HOST os.environ[get_env_var_name(ServiceNameEnum.CONTEXT, ENVVAR_SUFIX_SERVICE_PORT_GRPC)] = str(GRPC_PORT) os.environ[get_env_var_name(ServiceNameEnum.CONTEXT, ENVVAR_SUFIX_SERVICE_PORT_HTTP)] = str(HTTP_PORT) @pytest.fixture(scope='session') def context_service_grpc(): _service = ContextService(context_s_mb[0], context_s_mb[1]) _service.start() yield _service _service.stop() @pytest.fixture(scope='session') def context_service_rest(): database = context_db_mb[0] _rest_server = RestServer() for endpoint_name, resource_class, resource_url in RESOURCES: _rest_server.add_resource(resource_class, resource_url, endpoint=endpoint_name, resource_class_args=(database,)) _rest_server.start() time.sleep(1) # bring time for the server to start yield _rest_server _rest_server.shutdown() _rest_server.join() @pytest.fixture(scope='session') def context_client_grpc(context_service_grpc : ContextService): # pylint: disable=redefined-outer-name _client = ContextClient() yield _client _client.close() def test_populate_database(): client = ContextClient(host=LOCAL_HOST, port=GRPC_PORT) client.SetContext(Context(**CONTEXT)) client.SetTopology(Topology(**TOPOLOGY)) client.SetDevice(Device(**DEVICE_R1)) client.SetDevice(Device(**DEVICE_R2)) client.SetDevice(Device(**DEVICE_R3)) client.SetLink(Link(**LINK_R1_R2)) client.SetLink(Link(**LINK_R1_R3)) client.SetLink(Link(**LINK_R2_R3)) client.SetService(Service(**SERVICE_R1_R2)) client.SetService(Service(**SERVICE_R1_R3)) client.SetService(Service(**SERVICE_R2_R3)) client.SetSlice(Slice(**SLICE_R1_R3)) client.SetConnection(Connection(**CONNECTION_R1_R3)) def test_rest_get_context_ids(context_service_rest : RestServer): # pylint: disable=redefined-outer-name reply = do_rest_get_request('/context_ids') def test_populate_database(context_client): set_context_response = context_client.SetContext(Context(**CONTEXT)) pytest.connection_r1_r3_uuid = set_context_response.context_uuid.uuid context_client.SetTopology(Topology(**TOPOLOGY)) set_device_r1_response = context_client.SetDevice(Device(**DEVICE_R1)) pytest.device_r1_uuid = set_device_r1_response.device_uuid.uuid context_client.SetDevice(Device(**DEVICE_R2)) context_client.SetDevice(Device(**DEVICE_R3)) set_link_r1_r2_response = context_client.SetLink(Link(**LINK_R1_R2)) pytest.link_r1_r2_uuid = set_link_r1_r2_response.link_uuid.uuid context_client.SetLink(Link(**LINK_R1_R3)) context_client.SetLink(Link(**LINK_R2_R3)) set_service_r1_r2_response = context_client.SetService(Service(**SERVICE_R1_R2)) pytest.service_r1_r2_uuid = set_service_r1_r2_response.service_uuid.uuid set_service_r1_r3_response = context_client.SetService(Service(**SERVICE_R1_R3)) pytest.service_r1_r3_uuid = set_service_r1_r3_response.service_uuid.uuid context_client.SetService(Service(**SERVICE_R2_R3)) set_slice_response = context_client.SetSlice(Slice(**SLICE_R1_R3)) pytest.slice_r1_r3_uuid = set_slice_response.slice_uuid.uuid context_client.SetConnection(Connection(**CONNECTION_R1_R3)) def test_rest_get_context_ids(nbi_service_rest: RestServer): # pylint: disable=redefined-outer-name reply = do_rest_get_request(URL_PREFIX + '/context_ids') validate_context_ids(reply) def test_rest_get_contexts(context_service_rest : RestServer): # pylint: disable=redefined-outer-name reply = do_rest_get_request('/contexts') def test_rest_get_contexts(nbi_service_rest : RestServer): # pylint: disable=redefined-outer-name reply = do_rest_get_request(URL_PREFIX + '/contexts') # LOGGER.warning(reply['contexts'][0]['service_ids'][0]) # LOGGER.warning(reply['contexts'][0]['service_ids'][0].keys()) pprint.pprint(reply) validate_contexts(reply) def test_rest_get_context(context_service_rest : RestServer): # pylint: disable=redefined-outer-name context_uuid = urllib.parse.quote(DEFAULT_CONTEXT_NAME) reply = do_rest_get_request('/context/{:s}'.format(context_uuid)) validate_context(reply) def test_rest_get_topology_ids(context_service_rest : RestServer): # pylint: disable=redefined-outer-name context_uuid = urllib.parse.quote(DEFAULT_CONTEXT_NAME) reply = do_rest_get_request('/context/{:s}/topology_ids'.format(context_uuid)) validate_topology_ids(reply) def test_rest_get_topologies(context_service_rest : RestServer): # pylint: disable=redefined-outer-name context_uuid = urllib.parse.quote(DEFAULT_CONTEXT_NAME) reply = do_rest_get_request('/context/{:s}/topologies'.format(context_uuid)) validate_topologies(reply) def test_rest_get_topology(context_service_rest : RestServer): # pylint: disable=redefined-outer-name context_uuid = urllib.parse.quote(DEFAULT_CONTEXT_NAME) topology_uuid = urllib.parse.quote(DEFAULT_TOPOLOGY_NAME) reply = do_rest_get_request('/context/{:s}/topology/{:s}'.format(context_uuid, topology_uuid)) validate_topology(reply, num_devices=3, num_links=3) def test_rest_get_service_ids(context_service_rest : RestServer): # pylint: disable=redefined-outer-name context_uuid = urllib.parse.quote(DEFAULT_CONTEXT_NAME) reply = do_rest_get_request('/context/{:s}/service_ids'.format(context_uuid)) validate_service_ids(reply) def test_rest_get_services(context_service_rest : RestServer): # pylint: disable=redefined-outer-name context_uuid = urllib.parse.quote(DEFAULT_CONTEXT_NAME) reply = do_rest_get_request('/context/{:s}/services'.format(context_uuid)) validate_services(reply) def test_rest_get_service(context_service_rest : RestServer): # pylint: disable=redefined-outer-name context_uuid = urllib.parse.quote(DEFAULT_CONTEXT_NAME) service_uuid = urllib.parse.quote(SERVICE_R1_R2_UUID, safe='') reply = do_rest_get_request('/context/{:s}/service/{:s}'.format(context_uuid, service_uuid)) validate_service(reply) def test_rest_get_slice_ids(context_service_rest : RestServer): # pylint: disable=redefined-outer-name context_uuid = urllib.parse.quote(DEFAULT_CONTEXT_NAME) reply = do_rest_get_request('/context/{:s}/slice_ids'.format(context_uuid)) #validate_slice_ids(reply) def test_rest_get_slices(context_service_rest : RestServer): # pylint: disable=redefined-outer-name context_uuid = urllib.parse.quote(DEFAULT_CONTEXT_NAME) reply = do_rest_get_request('/context/{:s}/slices'.format(context_uuid)) #validate_slices(reply) def test_rest_get_slice(context_service_rest : RestServer): # pylint: disable=redefined-outer-name context_uuid = urllib.parse.quote(DEFAULT_CONTEXT_NAME) slice_uuid = urllib.parse.quote(SLICE_R1_R3_UUID, safe='') reply = do_rest_get_request('/context/{:s}/slice/{:s}'.format(context_uuid, slice_uuid)) #validate_slice(reply) def test_rest_get_device_ids(context_service_rest : RestServer): # pylint: disable=redefined-outer-name reply = do_rest_get_request('/device_ids') validate_device_ids(reply) def test_rest_get_devices(context_service_rest : RestServer): # pylint: disable=redefined-outer-name reply = do_rest_get_request('/devices') validate_devices(reply) def test_rest_get_device(context_service_rest : RestServer): # pylint: disable=redefined-outer-name device_uuid = urllib.parse.quote(DEVICE_R1_UUID, safe='') reply = do_rest_get_request('/device/{:s}'.format(device_uuid)) validate_device(reply) def test_rest_get_link_ids(context_service_rest : RestServer): # pylint: disable=redefined-outer-name reply = do_rest_get_request('/link_ids') validate_link_ids(reply) def test_rest_get_links(context_service_rest : RestServer): # pylint: disable=redefined-outer-name reply = do_rest_get_request('/links') validate_links(reply) def test_rest_get_link(context_service_rest : RestServer): # pylint: disable=redefined-outer-name link_uuid = urllib.parse.quote(LINK_R1_R2_UUID, safe='') reply = do_rest_get_request('/link/{:s}'.format(link_uuid)) validate_link(reply) def test_rest_get_connection_ids(context_service_rest : RestServer): # pylint: disable=redefined-outer-name context_uuid = urllib.parse.quote(DEFAULT_CONTEXT_NAME) service_uuid = urllib.parse.quote(SERVICE_R1_R3_UUID, safe='') reply = do_rest_get_request('/context/{:s}/service/{:s}/connection_ids'.format(context_uuid, service_uuid)) validate_connection_ids(reply) def test_rest_get_connections(context_service_rest : RestServer): # pylint: disable=redefined-outer-name context_uuid = urllib.parse.quote(DEFAULT_CONTEXT_NAME) service_uuid = urllib.parse.quote(SERVICE_R1_R3_UUID, safe='') reply = do_rest_get_request('/context/{:s}/service/{:s}/connections'.format(context_uuid, service_uuid)) validate_connections(reply) def test_rest_get_connection(context_service_rest : RestServer): # pylint: disable=redefined-outer-name connection_uuid = urllib.parse.quote(CONNECTION_R1_R3_UUID, safe='') reply = do_rest_get_request('/connection/{:s}'.format(connection_uuid)) validate_connection(reply) def test_rest_get_policyrule_ids(context_service_rest : RestServer): # pylint: disable=redefined-outer-name reply = do_rest_get_request('/policyrule_ids') #validate_policyrule_ids(reply) def test_rest_get_policyrules(context_service_rest : RestServer): # pylint: disable=redefined-outer-name reply = do_rest_get_request('/policyrules') #validate_policyrules(reply) def test_rest_get_policyrule(context_service_rest : RestServer): # pylint: disable=redefined-outer-name policyrule_uuid = urllib.parse.quote(POLICYRULE_UUID, safe='') reply = do_rest_get_request('/policyrule/{:s}'.format(policyrule_uuid)) #validate_policyrule(reply) # def test_rest_get_context(nbi_service_rest : RestServer): # pylint: disable=redefined-outer-name # context_uuid = urllib.parse.quote(DEFAULT_CONTEXT_NAME) # reply = do_rest_get_request(URL_PREFIX + '/context/{:s}'.format(context_uuid)) # validate_context(reply) # def test_rest_get_topology_ids(nbi_service_rest : RestServer): # pylint: disable=redefined-outer-name # context_uuid = urllib.parse.quote(DEFAULT_CONTEXT_NAME) # reply = do_rest_get_request(URL_PREFIX + '/context/{:s}/topology_ids'.format(context_uuid)) # validate_topology_ids(reply) # def test_rest_get_topologies(nbi_service_rest : RestServer): # pylint: disable=redefined-outer-name # context_uuid = urllib.parse.quote(DEFAULT_CONTEXT_NAME) # reply = do_rest_get_request(URL_PREFIX + '/context/{:s}/topologies'.format(context_uuid)) # validate_topologies(reply) # def test_rest_get_topology(nbi_service_rest : RestServer): # pylint: disable=redefined-outer-name # context_uuid = urllib.parse.quote(DEFAULT_CONTEXT_NAME) # topology_uuid = urllib.parse.quote(DEFAULT_TOPOLOGY_NAME) # reply = do_rest_get_request(URL_PREFIX + '/context/{:s}/topology/{:s}'.format(context_uuid, topology_uuid)) # validate_topology(reply, num_devices=3, num_links=3) # def test_rest_get_service_ids(nbi_service_rest : RestServer): # pylint: disable=redefined-outer-name # context_uuid = urllib.parse.quote(DEFAULT_CONTEXT_NAME) # reply = do_rest_get_request(URL_PREFIX + '/context/{:s}/service_ids'.format(context_uuid)) # validate_service_ids(reply) # def test_rest_get_services(nbi_service_rest : RestServer): # pylint: disable=redefined-outer-name # context_uuid = urllib.parse.quote(DEFAULT_CONTEXT_NAME) # reply = do_rest_get_request(URL_PREFIX + '/context/{:s}/services'.format(context_uuid)) # validate_services(reply) # def test_rest_get_service(nbi_service_rest : RestServer): # pylint: disable=redefined-outer-name # context_uuid = urllib.parse.quote(DEFAULT_CONTEXT_NAME) # service_uuid = urllib.parse.quote(service_r1_r2_uuid, safe='') # reply = do_rest_get_request(URL_PREFIX + '/context/{:s}/service/{:s}'.format(context_uuid, service_uuid)) # validate_service(reply) # def test_rest_get_slice_ids(nbi_service_rest : RestServer): # pylint: disable=redefined-outer-name # context_uuid = urllib.parse.quote(DEFAULT_CONTEXT_NAME) # reply = do_rest_get_request(URL_PREFIX + '/context/{:s}/slice_ids'.format(context_uuid)) # #validate_slice_ids(reply) # def test_rest_get_slices(nbi_service_rest : RestServer): # pylint: disable=redefined-outer-name # context_uuid = urllib.parse.quote(DEFAULT_CONTEXT_NAME) # reply = do_rest_get_request(URL_PREFIX + '/context/{:s}/slices'.format(context_uuid)) # #validate_slices(reply) # def test_rest_get_slice(nbi_service_rest : RestServer): # pylint: disable=redefined-outer-name # context_uuid = urllib.parse.quote(DEFAULT_CONTEXT_NAME) # slice_uuid = urllib.parse.quote(slice_r1_r3_uuid, safe='') # reply = do_rest_get_request(URL_PREFIX + '/context/{:s}/slice/{:s}'.format(context_uuid, slice_uuid)) # #validate_slice(reply) # def test_rest_get_device_ids(nbi_service_rest : RestServer): # pylint: disable=redefined-outer-name # reply = do_rest_get_request(URL_PREFIX + '/device_ids') # validate_device_ids(reply) # def test_rest_get_devices(nbi_service_rest : RestServer): # pylint: disable=redefined-outer-name # reply = do_rest_get_request(URL_PREFIX + '/devices') # validate_devices(reply) # def test_rest_get_device(nbi_service_rest : RestServer): # pylint: disable=redefined-outer-name # device_uuid = urllib.parse.quote(device_r1_uuid, safe='') # reply = do_rest_get_request(URL_PREFIX + '/device/{:s}'.format(device_uuid)) # validate_device(reply) # def test_rest_get_link_ids(nbi_service_rest : RestServer): # pylint: disable=redefined-outer-name # reply = do_rest_get_request(URL_PREFIX + '/link_ids') # validate_link_ids(reply) # def test_rest_get_links(nbi_service_rest : RestServer): # pylint: disable=redefined-outer-name # reply = do_rest_get_request(URL_PREFIX + '/links') # validate_links(reply) # def test_rest_get_link(nbi_service_rest : RestServer): # pylint: disable=redefined-outer-name # link_uuid = urllib.parse.quote(link_r1_r2_uuid, safe='') # reply = do_rest_get_request(URL_PREFIX + '/link/{:s}'.format(link_uuid)) # validate_link(reply) # def test_rest_get_connection_ids(nbi_service_rest : RestServer): # pylint: disable=redefined-outer-name # context_uuid = urllib.parse.quote(DEFAULT_CONTEXT_NAME) # service_uuid = urllib.parse.quote(service_r1_r3_uuid, safe='') # reply = do_rest_get_request(URL_PREFIX + '/context/{:s}/service/{:s}/connection_ids'.format(context_uuid, service_uuid)) # validate_connection_ids(reply) # def test_rest_get_connections(nbi_service_rest : RestServer): # pylint: disable=redefined-outer-name # context_uuid = urllib.parse.quote(DEFAULT_CONTEXT_NAME) # service_uuid = urllib.parse.quote(service_r1_r3_uuid, safe='') # reply = do_rest_get_request(URL_PREFIX + '/context/{:s}/service/{:s}/connections'.format(context_uuid, service_uuid)) # validate_connections(reply) # def test_rest_get_connection(nbi_service_rest : RestServer): # pylint: disable=redefined-outer-name # connection_uuid = urllib.parse.quote(connection_r1_r3_uuid, safe='') # reply = do_rest_get_request(URL_PREFIX + '/connection/{:s}'.format(connection_uuid)) # validate_connection(reply) # def test_rest_get_policyrule_ids(nbi_service_rest : RestServer): # pylint: disable=redefined-outer-name # reply = do_rest_get_request(URL_PREFIX + '/policyrule_ids') # #validate_policyrule_ids(reply) # def test_rest_get_policyrules(nbi_service_rest : RestServer): # pylint: disable=redefined-outer-name # reply = do_rest_get_request(URL_PREFIX + '/policyrules') # #validate_policyrules(reply) # def test_rest_get_policyrule(nbi_service_rest : RestServer): # pylint: disable=redefined-outer-name # policyrule_uuid_quoted = urllib.parse.quote(policyrule_uuid, safe='') # reply = do_rest_get_request(URL_PREFIX + '/policyrule/{:s}'.format(policyrule_uuid_quoted)) # #validate_policyrule(reply) Loading
src/common/tests/MockServicerImpl_Context.py +85 −1 Original line number Diff line number Diff line Loading @@ -25,12 +25,18 @@ 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.proto.policy_pb2 import ( PolicyRule, PolicyRuleId, PolicyRuleIdList, PolicyRuleList, ) from common.tools.grpc.Tools import grpc_message_to_json, grpc_message_to_json_string from common.tools.object_factory.Device import json_device_id from common.tools.object_factory.Link import json_link_id from .InMemoryObjectDatabase import InMemoryObjectDatabase from .MockMessageBroker import ( TOPIC_CONNECTION, TOPIC_CONTEXT, TOPIC_DEVICE, TOPIC_LINK, TOPIC_SERVICE, TOPIC_SLICE, TOPIC_TOPOLOGY, TOPIC_CONNECTION, TOPIC_CONTEXT, TOPIC_DEVICE, TOPIC_LINK, TOPIC_SERVICE, TOPIC_SLICE, TOPIC_TOPOLOGY, TOPIC_POLICY, MockMessageBroker, notify_event) LOGGER = logging.getLogger(__name__) Loading Loading @@ -615,3 +621,81 @@ class MockServicerImpl_Context(ContextServiceServicer): 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
src/common/type_checkers/Assertions.py +15 −1 Original line number Diff line number Diff line Loading @@ -192,6 +192,15 @@ def validate_connection_id(message): assert 'connection_uuid' in message validate_uuid(message['connection_uuid']) def validate_slice_id(message, context_uuid = None): assert isinstance(message, dict) assert len(message.keys()) == 2 assert 'context_id' in message validate_context_id(message['context_id']) if context_uuid is not None: assert message['context_id']['context_uuid']['uuid'] == context_uuid assert 'slice_uuid' in message validate_uuid(message['slice_uuid']) # ----- Lists of Identifiers ------------------------------------------------------------------------------------------- Loading Loading @@ -242,7 +251,9 @@ def validate_connection_ids(message): def validate_context(message): assert isinstance(message, dict) assert len(message.keys()) == 3 assert len(message.keys()) == 5 assert 'name' in message assert isinstance(message['name'], str) assert 'context_id' in message validate_context_id(message['context_id']) context_uuid = message['context_id']['context_uuid']['uuid'] Loading @@ -252,6 +263,9 @@ def validate_context(message): assert 'topology_ids' in message assert isinstance(message['topology_ids'], list) for topology_id in message['topology_ids']: validate_topology_id(topology_id, context_uuid=context_uuid) assert 'slice_ids' in message assert isinstance(message['slice_ids'], list) for slice_id in message['slice_ids']: validate_slice_id(slice_id, context_uuid=context_uuid) def validate_service_state(message): assert isinstance(message, dict) Loading
src/nbi/tests/test_debug_api.py +164 −162 Original line number Diff line number Diff line Loading @@ -13,6 +13,7 @@ # limitations under the License. import logging, os, pytest, time, urllib import pprint from common.Constants import DEFAULT_CONTEXT_NAME, DEFAULT_TOPOLOGY_NAME, ServiceNameEnum from common.proto.context_pb2 import Connection, Context, Device, Link, Service, Slice, Topology from common.proto.policy_pb2 import PolicyRuleIdList, PolicyRuleId, PolicyRuleList, PolicyRule Loading @@ -25,17 +26,35 @@ from common.type_checkers.Assertions import ( validate_contexts, validate_device, validate_device_ids, validate_devices, validate_link, validate_link_ids, validate_links, validate_service, validate_service_ids, validate_services, validate_topologies, validate_topology, validate_topology_ids) from context.client.ContextClient import ContextClient from nbi.tests.PrepareTestScenario import do_rest_get_request from nbi.tests.PrepareTestScenario import do_rest_get_request, context_client, nbi_service_rest from ..service.rest_server.RestServer import RestServer from .MockService_Dependencies import MockService_Dependencies from .Objects import ( CONNECTION_R1_R3, CONNECTION_R1_R3_ID, CONNECTION_R1_R3_UUID, CONTEXT, CONTEXT_ID, DEVICE_R1, DEVICE_R1_ID, DEVICE_R1_UUID, DEVICE_R2, DEVICE_R2_ID, DEVICE_R2_UUID, DEVICE_R3, DEVICE_R3_ID, DEVICE_R3_UUID, LINK_R1_R2, LINK_R1_R2_ID, LINK_R1_R2_UUID, SERVICE_R1_R2, SERVICE_R1_R2_ID, SERVICE_R1_R2_UUID, SERVICE_R1_R3, SERVICE_R1_R3_ID, SERVICE_R1_R3_UUID, SERVICE_R2_R3, SERVICE_R2_R3_ID, SERVICE_R2_R3_UUID, SLICE_R1_R3, TOPOLOGY, TOPOLOGY_ID, POLICY_RULE, POLICY_RULE_ID, POLICY_RULE_UUID from context.tests.Objects import ( CONNECTION_R1_R3, CONTEXT, DEVICE_R1, DEVICE_R2, DEVICE_R3, LINK_R1_R2, LINK_R1_R3, LINK_R2_R3, SERVICE_R1_R2, SERVICE_R1_R3, SERVICE_R2_R3, SLICE_R1_R3, TOPOLOGY ) from ..service.rest_server.nbi_plugins.debug_api import URL_PREFIX # from context.client.ContextClient import ContextClient # from ..service.rest_server.nbi_plugins.debug_api import RESOURCES # connection_r1_r3_uuid = '' # policyrule_uuid = '' # slice_r1_r3_uuid = '' # service_r1_r2_uuid = '' # service_r1_r3_uuid = '' # link_r1_r2_uuid = '' # device_r1_uuid = '' def pytest_namespace(): return { 'connection_r1_r3_uuid' : '', 'policyrule_uuid' : '', 'slice_r1_r3_uuid' : '', 'service_r1_r2_uuid' : '', 'service_r1_r3_uuid' : '', 'link_r1_r2_uuid' : '', 'device_r1_uuid' : '', } @pytest.fixture(scope='session') def mock_service(): Loading @@ -61,160 +80,143 @@ os.environ[get_env_var_name(ServiceNameEnum.CONTEXT, ENVVAR_SUFIX_SERVICE_HOST os.environ[get_env_var_name(ServiceNameEnum.CONTEXT, ENVVAR_SUFIX_SERVICE_PORT_GRPC)] = str(GRPC_PORT) os.environ[get_env_var_name(ServiceNameEnum.CONTEXT, ENVVAR_SUFIX_SERVICE_PORT_HTTP)] = str(HTTP_PORT) @pytest.fixture(scope='session') def context_service_grpc(): _service = ContextService(context_s_mb[0], context_s_mb[1]) _service.start() yield _service _service.stop() @pytest.fixture(scope='session') def context_service_rest(): database = context_db_mb[0] _rest_server = RestServer() for endpoint_name, resource_class, resource_url in RESOURCES: _rest_server.add_resource(resource_class, resource_url, endpoint=endpoint_name, resource_class_args=(database,)) _rest_server.start() time.sleep(1) # bring time for the server to start yield _rest_server _rest_server.shutdown() _rest_server.join() @pytest.fixture(scope='session') def context_client_grpc(context_service_grpc : ContextService): # pylint: disable=redefined-outer-name _client = ContextClient() yield _client _client.close() def test_populate_database(): client = ContextClient(host=LOCAL_HOST, port=GRPC_PORT) client.SetContext(Context(**CONTEXT)) client.SetTopology(Topology(**TOPOLOGY)) client.SetDevice(Device(**DEVICE_R1)) client.SetDevice(Device(**DEVICE_R2)) client.SetDevice(Device(**DEVICE_R3)) client.SetLink(Link(**LINK_R1_R2)) client.SetLink(Link(**LINK_R1_R3)) client.SetLink(Link(**LINK_R2_R3)) client.SetService(Service(**SERVICE_R1_R2)) client.SetService(Service(**SERVICE_R1_R3)) client.SetService(Service(**SERVICE_R2_R3)) client.SetSlice(Slice(**SLICE_R1_R3)) client.SetConnection(Connection(**CONNECTION_R1_R3)) def test_rest_get_context_ids(context_service_rest : RestServer): # pylint: disable=redefined-outer-name reply = do_rest_get_request('/context_ids') def test_populate_database(context_client): set_context_response = context_client.SetContext(Context(**CONTEXT)) pytest.connection_r1_r3_uuid = set_context_response.context_uuid.uuid context_client.SetTopology(Topology(**TOPOLOGY)) set_device_r1_response = context_client.SetDevice(Device(**DEVICE_R1)) pytest.device_r1_uuid = set_device_r1_response.device_uuid.uuid context_client.SetDevice(Device(**DEVICE_R2)) context_client.SetDevice(Device(**DEVICE_R3)) set_link_r1_r2_response = context_client.SetLink(Link(**LINK_R1_R2)) pytest.link_r1_r2_uuid = set_link_r1_r2_response.link_uuid.uuid context_client.SetLink(Link(**LINK_R1_R3)) context_client.SetLink(Link(**LINK_R2_R3)) set_service_r1_r2_response = context_client.SetService(Service(**SERVICE_R1_R2)) pytest.service_r1_r2_uuid = set_service_r1_r2_response.service_uuid.uuid set_service_r1_r3_response = context_client.SetService(Service(**SERVICE_R1_R3)) pytest.service_r1_r3_uuid = set_service_r1_r3_response.service_uuid.uuid context_client.SetService(Service(**SERVICE_R2_R3)) set_slice_response = context_client.SetSlice(Slice(**SLICE_R1_R3)) pytest.slice_r1_r3_uuid = set_slice_response.slice_uuid.uuid context_client.SetConnection(Connection(**CONNECTION_R1_R3)) def test_rest_get_context_ids(nbi_service_rest: RestServer): # pylint: disable=redefined-outer-name reply = do_rest_get_request(URL_PREFIX + '/context_ids') validate_context_ids(reply) def test_rest_get_contexts(context_service_rest : RestServer): # pylint: disable=redefined-outer-name reply = do_rest_get_request('/contexts') def test_rest_get_contexts(nbi_service_rest : RestServer): # pylint: disable=redefined-outer-name reply = do_rest_get_request(URL_PREFIX + '/contexts') # LOGGER.warning(reply['contexts'][0]['service_ids'][0]) # LOGGER.warning(reply['contexts'][0]['service_ids'][0].keys()) pprint.pprint(reply) validate_contexts(reply) def test_rest_get_context(context_service_rest : RestServer): # pylint: disable=redefined-outer-name context_uuid = urllib.parse.quote(DEFAULT_CONTEXT_NAME) reply = do_rest_get_request('/context/{:s}'.format(context_uuid)) validate_context(reply) def test_rest_get_topology_ids(context_service_rest : RestServer): # pylint: disable=redefined-outer-name context_uuid = urllib.parse.quote(DEFAULT_CONTEXT_NAME) reply = do_rest_get_request('/context/{:s}/topology_ids'.format(context_uuid)) validate_topology_ids(reply) def test_rest_get_topologies(context_service_rest : RestServer): # pylint: disable=redefined-outer-name context_uuid = urllib.parse.quote(DEFAULT_CONTEXT_NAME) reply = do_rest_get_request('/context/{:s}/topologies'.format(context_uuid)) validate_topologies(reply) def test_rest_get_topology(context_service_rest : RestServer): # pylint: disable=redefined-outer-name context_uuid = urllib.parse.quote(DEFAULT_CONTEXT_NAME) topology_uuid = urllib.parse.quote(DEFAULT_TOPOLOGY_NAME) reply = do_rest_get_request('/context/{:s}/topology/{:s}'.format(context_uuid, topology_uuid)) validate_topology(reply, num_devices=3, num_links=3) def test_rest_get_service_ids(context_service_rest : RestServer): # pylint: disable=redefined-outer-name context_uuid = urllib.parse.quote(DEFAULT_CONTEXT_NAME) reply = do_rest_get_request('/context/{:s}/service_ids'.format(context_uuid)) validate_service_ids(reply) def test_rest_get_services(context_service_rest : RestServer): # pylint: disable=redefined-outer-name context_uuid = urllib.parse.quote(DEFAULT_CONTEXT_NAME) reply = do_rest_get_request('/context/{:s}/services'.format(context_uuid)) validate_services(reply) def test_rest_get_service(context_service_rest : RestServer): # pylint: disable=redefined-outer-name context_uuid = urllib.parse.quote(DEFAULT_CONTEXT_NAME) service_uuid = urllib.parse.quote(SERVICE_R1_R2_UUID, safe='') reply = do_rest_get_request('/context/{:s}/service/{:s}'.format(context_uuid, service_uuid)) validate_service(reply) def test_rest_get_slice_ids(context_service_rest : RestServer): # pylint: disable=redefined-outer-name context_uuid = urllib.parse.quote(DEFAULT_CONTEXT_NAME) reply = do_rest_get_request('/context/{:s}/slice_ids'.format(context_uuid)) #validate_slice_ids(reply) def test_rest_get_slices(context_service_rest : RestServer): # pylint: disable=redefined-outer-name context_uuid = urllib.parse.quote(DEFAULT_CONTEXT_NAME) reply = do_rest_get_request('/context/{:s}/slices'.format(context_uuid)) #validate_slices(reply) def test_rest_get_slice(context_service_rest : RestServer): # pylint: disable=redefined-outer-name context_uuid = urllib.parse.quote(DEFAULT_CONTEXT_NAME) slice_uuid = urllib.parse.quote(SLICE_R1_R3_UUID, safe='') reply = do_rest_get_request('/context/{:s}/slice/{:s}'.format(context_uuid, slice_uuid)) #validate_slice(reply) def test_rest_get_device_ids(context_service_rest : RestServer): # pylint: disable=redefined-outer-name reply = do_rest_get_request('/device_ids') validate_device_ids(reply) def test_rest_get_devices(context_service_rest : RestServer): # pylint: disable=redefined-outer-name reply = do_rest_get_request('/devices') validate_devices(reply) def test_rest_get_device(context_service_rest : RestServer): # pylint: disable=redefined-outer-name device_uuid = urllib.parse.quote(DEVICE_R1_UUID, safe='') reply = do_rest_get_request('/device/{:s}'.format(device_uuid)) validate_device(reply) def test_rest_get_link_ids(context_service_rest : RestServer): # pylint: disable=redefined-outer-name reply = do_rest_get_request('/link_ids') validate_link_ids(reply) def test_rest_get_links(context_service_rest : RestServer): # pylint: disable=redefined-outer-name reply = do_rest_get_request('/links') validate_links(reply) def test_rest_get_link(context_service_rest : RestServer): # pylint: disable=redefined-outer-name link_uuid = urllib.parse.quote(LINK_R1_R2_UUID, safe='') reply = do_rest_get_request('/link/{:s}'.format(link_uuid)) validate_link(reply) def test_rest_get_connection_ids(context_service_rest : RestServer): # pylint: disable=redefined-outer-name context_uuid = urllib.parse.quote(DEFAULT_CONTEXT_NAME) service_uuid = urllib.parse.quote(SERVICE_R1_R3_UUID, safe='') reply = do_rest_get_request('/context/{:s}/service/{:s}/connection_ids'.format(context_uuid, service_uuid)) validate_connection_ids(reply) def test_rest_get_connections(context_service_rest : RestServer): # pylint: disable=redefined-outer-name context_uuid = urllib.parse.quote(DEFAULT_CONTEXT_NAME) service_uuid = urllib.parse.quote(SERVICE_R1_R3_UUID, safe='') reply = do_rest_get_request('/context/{:s}/service/{:s}/connections'.format(context_uuid, service_uuid)) validate_connections(reply) def test_rest_get_connection(context_service_rest : RestServer): # pylint: disable=redefined-outer-name connection_uuid = urllib.parse.quote(CONNECTION_R1_R3_UUID, safe='') reply = do_rest_get_request('/connection/{:s}'.format(connection_uuid)) validate_connection(reply) def test_rest_get_policyrule_ids(context_service_rest : RestServer): # pylint: disable=redefined-outer-name reply = do_rest_get_request('/policyrule_ids') #validate_policyrule_ids(reply) def test_rest_get_policyrules(context_service_rest : RestServer): # pylint: disable=redefined-outer-name reply = do_rest_get_request('/policyrules') #validate_policyrules(reply) def test_rest_get_policyrule(context_service_rest : RestServer): # pylint: disable=redefined-outer-name policyrule_uuid = urllib.parse.quote(POLICYRULE_UUID, safe='') reply = do_rest_get_request('/policyrule/{:s}'.format(policyrule_uuid)) #validate_policyrule(reply) # def test_rest_get_context(nbi_service_rest : RestServer): # pylint: disable=redefined-outer-name # context_uuid = urllib.parse.quote(DEFAULT_CONTEXT_NAME) # reply = do_rest_get_request(URL_PREFIX + '/context/{:s}'.format(context_uuid)) # validate_context(reply) # def test_rest_get_topology_ids(nbi_service_rest : RestServer): # pylint: disable=redefined-outer-name # context_uuid = urllib.parse.quote(DEFAULT_CONTEXT_NAME) # reply = do_rest_get_request(URL_PREFIX + '/context/{:s}/topology_ids'.format(context_uuid)) # validate_topology_ids(reply) # def test_rest_get_topologies(nbi_service_rest : RestServer): # pylint: disable=redefined-outer-name # context_uuid = urllib.parse.quote(DEFAULT_CONTEXT_NAME) # reply = do_rest_get_request(URL_PREFIX + '/context/{:s}/topologies'.format(context_uuid)) # validate_topologies(reply) # def test_rest_get_topology(nbi_service_rest : RestServer): # pylint: disable=redefined-outer-name # context_uuid = urllib.parse.quote(DEFAULT_CONTEXT_NAME) # topology_uuid = urllib.parse.quote(DEFAULT_TOPOLOGY_NAME) # reply = do_rest_get_request(URL_PREFIX + '/context/{:s}/topology/{:s}'.format(context_uuid, topology_uuid)) # validate_topology(reply, num_devices=3, num_links=3) # def test_rest_get_service_ids(nbi_service_rest : RestServer): # pylint: disable=redefined-outer-name # context_uuid = urllib.parse.quote(DEFAULT_CONTEXT_NAME) # reply = do_rest_get_request(URL_PREFIX + '/context/{:s}/service_ids'.format(context_uuid)) # validate_service_ids(reply) # def test_rest_get_services(nbi_service_rest : RestServer): # pylint: disable=redefined-outer-name # context_uuid = urllib.parse.quote(DEFAULT_CONTEXT_NAME) # reply = do_rest_get_request(URL_PREFIX + '/context/{:s}/services'.format(context_uuid)) # validate_services(reply) # def test_rest_get_service(nbi_service_rest : RestServer): # pylint: disable=redefined-outer-name # context_uuid = urllib.parse.quote(DEFAULT_CONTEXT_NAME) # service_uuid = urllib.parse.quote(service_r1_r2_uuid, safe='') # reply = do_rest_get_request(URL_PREFIX + '/context/{:s}/service/{:s}'.format(context_uuid, service_uuid)) # validate_service(reply) # def test_rest_get_slice_ids(nbi_service_rest : RestServer): # pylint: disable=redefined-outer-name # context_uuid = urllib.parse.quote(DEFAULT_CONTEXT_NAME) # reply = do_rest_get_request(URL_PREFIX + '/context/{:s}/slice_ids'.format(context_uuid)) # #validate_slice_ids(reply) # def test_rest_get_slices(nbi_service_rest : RestServer): # pylint: disable=redefined-outer-name # context_uuid = urllib.parse.quote(DEFAULT_CONTEXT_NAME) # reply = do_rest_get_request(URL_PREFIX + '/context/{:s}/slices'.format(context_uuid)) # #validate_slices(reply) # def test_rest_get_slice(nbi_service_rest : RestServer): # pylint: disable=redefined-outer-name # context_uuid = urllib.parse.quote(DEFAULT_CONTEXT_NAME) # slice_uuid = urllib.parse.quote(slice_r1_r3_uuid, safe='') # reply = do_rest_get_request(URL_PREFIX + '/context/{:s}/slice/{:s}'.format(context_uuid, slice_uuid)) # #validate_slice(reply) # def test_rest_get_device_ids(nbi_service_rest : RestServer): # pylint: disable=redefined-outer-name # reply = do_rest_get_request(URL_PREFIX + '/device_ids') # validate_device_ids(reply) # def test_rest_get_devices(nbi_service_rest : RestServer): # pylint: disable=redefined-outer-name # reply = do_rest_get_request(URL_PREFIX + '/devices') # validate_devices(reply) # def test_rest_get_device(nbi_service_rest : RestServer): # pylint: disable=redefined-outer-name # device_uuid = urllib.parse.quote(device_r1_uuid, safe='') # reply = do_rest_get_request(URL_PREFIX + '/device/{:s}'.format(device_uuid)) # validate_device(reply) # def test_rest_get_link_ids(nbi_service_rest : RestServer): # pylint: disable=redefined-outer-name # reply = do_rest_get_request(URL_PREFIX + '/link_ids') # validate_link_ids(reply) # def test_rest_get_links(nbi_service_rest : RestServer): # pylint: disable=redefined-outer-name # reply = do_rest_get_request(URL_PREFIX + '/links') # validate_links(reply) # def test_rest_get_link(nbi_service_rest : RestServer): # pylint: disable=redefined-outer-name # link_uuid = urllib.parse.quote(link_r1_r2_uuid, safe='') # reply = do_rest_get_request(URL_PREFIX + '/link/{:s}'.format(link_uuid)) # validate_link(reply) # def test_rest_get_connection_ids(nbi_service_rest : RestServer): # pylint: disable=redefined-outer-name # context_uuid = urllib.parse.quote(DEFAULT_CONTEXT_NAME) # service_uuid = urllib.parse.quote(service_r1_r3_uuid, safe='') # reply = do_rest_get_request(URL_PREFIX + '/context/{:s}/service/{:s}/connection_ids'.format(context_uuid, service_uuid)) # validate_connection_ids(reply) # def test_rest_get_connections(nbi_service_rest : RestServer): # pylint: disable=redefined-outer-name # context_uuid = urllib.parse.quote(DEFAULT_CONTEXT_NAME) # service_uuid = urllib.parse.quote(service_r1_r3_uuid, safe='') # reply = do_rest_get_request(URL_PREFIX + '/context/{:s}/service/{:s}/connections'.format(context_uuid, service_uuid)) # validate_connections(reply) # def test_rest_get_connection(nbi_service_rest : RestServer): # pylint: disable=redefined-outer-name # connection_uuid = urllib.parse.quote(connection_r1_r3_uuid, safe='') # reply = do_rest_get_request(URL_PREFIX + '/connection/{:s}'.format(connection_uuid)) # validate_connection(reply) # def test_rest_get_policyrule_ids(nbi_service_rest : RestServer): # pylint: disable=redefined-outer-name # reply = do_rest_get_request(URL_PREFIX + '/policyrule_ids') # #validate_policyrule_ids(reply) # def test_rest_get_policyrules(nbi_service_rest : RestServer): # pylint: disable=redefined-outer-name # reply = do_rest_get_request(URL_PREFIX + '/policyrules') # #validate_policyrules(reply) # def test_rest_get_policyrule(nbi_service_rest : RestServer): # pylint: disable=redefined-outer-name # policyrule_uuid_quoted = urllib.parse.quote(policyrule_uuid, safe='') # reply = do_rest_get_request(URL_PREFIX + '/policyrule/{:s}'.format(policyrule_uuid_quoted)) # #validate_policyrule(reply)