# Copyright 2022-2023 ETSI TeraFlowSDN - TFS OSG (https://tfs.etsi.org/) # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # 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 from common.Settings import ( ENVVAR_SUFIX_SERVICE_HOST, ENVVAR_SUFIX_SERVICE_PORT_GRPC, ENVVAR_SUFIX_SERVICE_PORT_HTTP, get_env_var_name, get_service_port_grpc, get_service_port_http ) from common.type_checkers.Assertions import ( validate_connection, validate_connection_ids, validate_connections, validate_context, validate_context_ids, 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 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 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(): _service = MockService_Dependencies(MOCKSERVICE_PORT) _service.configure_env_vars() _service.start() yield _service _service.stop() LOGGER = logging.getLogger(__name__) LOGGER.setLevel(logging.DEBUG) LOCAL_HOST = '127.0.0.1' GRPC_PORT = 10000 + int(get_service_port_grpc(ServiceNameEnum.CONTEXT)) # avoid privileged ports HTTP_PORT = 10000 + int(get_service_port_http(ServiceNameEnum.CONTEXT)) # avoid privileged ports MOCKSERVICE_PORT = 10000 DEVICE_SERVICE_PORT = MOCKSERVICE_PORT + get_service_port_grpc(ServiceNameEnum.DEVICE) # avoid privileged ports os.environ[get_env_var_name(ServiceNameEnum.CONTEXT, ENVVAR_SUFIX_SERVICE_HOST )] = str(LOCAL_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) 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(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(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)