# Copyright 2022-2025 ETSI SDG TeraFlowSDN (TFS) (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 copy, deepdiff, json, logging, operator, os, pytest, time from flask import Flask, jsonify, make_response from flask_restful import Resource from common.proto.context_pb2 import ConfigActionEnum, Device, DeviceId from common.tools.descriptor.Tools import format_custom_config_rules from common.tools.grpc.Tools import grpc_message_to_json_string from common.tools.object_factory.Device import ( json_device_connect_rules, json_device_id, json_device_ietf_actn_disabled ) from common.tools.service.GenericRestServer import GenericRestServer from context.client.ContextClient import ContextClient from device.client.DeviceClient import DeviceClient from device.service.DeviceService import DeviceService from device.service.driver_api._Driver import _Driver from tests.tools.mock_ietf_actn_sdn_ctrl.ResourceEthServices import EthService, EthServices from tests.tools.mock_ietf_actn_sdn_ctrl.ResourceOsuTunnels import OsuTunnel, OsuTunnels os.environ['DEVICE_EMULATED_ONLY'] = 'TRUE' from .PrepareTestScenario import ( # pylint: disable=unused-import # be careful, order of symbols is important here! mock_service, device_service, context_client, device_client, monitoring_client, test_prepare_environment ) DEVICE_UUID = 'DEVICE-IETF-ACTN' DEVICE_ADDRESS = '127.0.0.1' DEVICE_PORT = 8080 DEVICE_USERNAME = 'admin' DEVICE_PASSWORD = 'admin' DEVICE_SCHEME = 'http' DEVICE_BASE_URL = '/restconf/v2/data' DEVICE_TIMEOUT = 120 DEVICE_VERIFY = False DEVICE_ID = json_device_id(DEVICE_UUID) DEVICE = json_device_ietf_actn_disabled(DEVICE_UUID) DEVICE_CONNECT_RULES = json_device_connect_rules(DEVICE_ADDRESS, DEVICE_PORT, { 'scheme' : DEVICE_SCHEME, 'username': DEVICE_USERNAME, 'password': DEVICE_PASSWORD, 'base_url': DEVICE_BASE_URL, 'timeout' : DEVICE_TIMEOUT, 'verify' : DEVICE_VERIFY, }) DATA_FILE_CONFIG_RULES = 'device/tests/data/ietf_actn/config_rules.json' DATA_FILE_DECONFIG_RULES = 'device/tests/data/ietf_actn/deconfig_rules.json' DATA_FILE_EXPECTED_OSU_TUNNELS = 'device/tests/data/ietf_actn/expected_osu_tunnels.json' DATA_FILE_EXPECTED_ETHT_SERVICES = 'device/tests/data/ietf_actn/expected_etht_services.json' LOGGER = logging.getLogger(__name__) LOGGER.setLevel(logging.DEBUG) @pytest.fixture(scope='session') def ietf_actn_sdn_ctrl( device_service : DeviceService, # pylint: disable=redefined-outer-name ) -> Flask: _rest_server = GenericRestServer(DEVICE_PORT, DEVICE_BASE_URL, bind_address=DEVICE_ADDRESS) _rest_server.app.config['DEBUG' ] = True _rest_server.app.config['ENV' ] = 'development' _rest_server.app.config['SERVER_NAME'] = '{:s}:{:d}'.format(DEVICE_ADDRESS, DEVICE_PORT) _rest_server.app.config['TESTING' ] = True class Root(Resource): def get(self): return make_response(jsonify({}), 200) add_rsrc = _rest_server.add_resource add_rsrc(Root, '/') add_rsrc(OsuTunnels, '/ietf-te:te/tunnels') add_rsrc(OsuTunnel, '/ietf-te:te/tunnels/tunnel="<string:osu_tunnel_name>"') add_rsrc(EthServices, '/ietf-eth-tran-service:etht-svc') add_rsrc(EthService, '/ietf-eth-tran-service:etht-svc/etht-svc-instances="<string:etht_service_name>"') _rest_server.start() time.sleep(1) # bring time for the server to start yield _rest_server _rest_server.shutdown() _rest_server.join() def test_device_ietf_actn_add( device_client : DeviceClient, # pylint: disable=redefined-outer-name device_service : DeviceService, # pylint: disable=redefined-outer-name ietf_actn_sdn_ctrl : GenericRestServer, # pylint: disable=redefined-outer-name ) -> None: DEVICE_WITH_CONNECT_RULES = copy.deepcopy(DEVICE) DEVICE_WITH_CONNECT_RULES['device_config']['config_rules'].extend(DEVICE_CONNECT_RULES) device_client.AddDevice(Device(**DEVICE_WITH_CONNECT_RULES)) driver_instance_cache = device_service.device_servicer.driver_instance_cache driver: _Driver = driver_instance_cache.get(DEVICE_UUID) assert driver is not None def test_device_ietf_actn_get( context_client : ContextClient, # pylint: disable=redefined-outer-name device_client : DeviceClient, # pylint: disable=redefined-outer-name ietf_actn_sdn_ctrl : GenericRestServer, # pylint: disable=redefined-outer-name ) -> None: initial_config = device_client.GetInitialConfig(DeviceId(**DEVICE_ID)) LOGGER.info('initial_config = {:s}'.format(grpc_message_to_json_string(initial_config))) device_data = context_client.GetDevice(DeviceId(**DEVICE_ID)) LOGGER.info('device_data = {:s}'.format(grpc_message_to_json_string(device_data))) def test_device_ietf_actn_configure( device_client : DeviceClient, # pylint: disable=redefined-outer-name device_service : DeviceService, # pylint: disable=redefined-outer-name ietf_actn_sdn_ctrl : GenericRestServer, # pylint: disable=redefined-outer-name ) -> None: ietf_actn_client = ietf_actn_sdn_ctrl.app.test_client() driver_instance_cache = device_service.device_servicer.driver_instance_cache driver : _Driver = driver_instance_cache.get(DEVICE_UUID) assert driver is not None retrieved_osu_tunnels = ietf_actn_client.get('/restconf/v2/data/ietf-te:te/tunnels') retrieved_osu_tunnels = retrieved_osu_tunnels.json LOGGER.info('osu_tunnels = {:s}'.format(str(retrieved_osu_tunnels))) expected_osu_tunnels = {'ietf-te:tunnel': []} osu_tunnels_diff = deepdiff.DeepDiff(expected_osu_tunnels, retrieved_osu_tunnels) if len(osu_tunnels_diff) > 0: LOGGER.error('PRE OSU TUNNELS - Differences:\n{:s}'.format(str(osu_tunnels_diff.pretty()))) assert len(osu_tunnels_diff) == 0 retrieved_etht_services = ietf_actn_client.get('/restconf/v2/data/ietf-eth-tran-service:etht-svc') retrieved_etht_services = retrieved_etht_services.json LOGGER.info('etht_services = {:s}'.format(str(retrieved_etht_services))) expected_etht_services = {'ietf-eth-tran-service:etht-svc': {'etht-svc-instances': []}} etht_services_diff = deepdiff.DeepDiff(expected_etht_services, retrieved_etht_services) if len(etht_services_diff) > 0: LOGGER.error('PRE ETHT SERVICES - Differences:\n{:s}'.format(str(etht_services_diff.pretty()))) assert len(etht_services_diff) == 0 retrieved_driver_config_rules = sorted(driver.GetConfig(), key=operator.itemgetter(0)) LOGGER.info('driver_config = {:s}'.format(str(retrieved_driver_config_rules))) assert isinstance(retrieved_driver_config_rules, list) retrieved_driver_config_rules = [ (resource_key, resource_value) for resource_key, resource_value in retrieved_driver_config_rules if resource_key != '/endpoints/endpoint[mgmt]' ] if len(retrieved_driver_config_rules) > 0: LOGGER.error('PRE DRIVER CONFIG RULES - Differences:\n{:s}'.format(str(retrieved_driver_config_rules))) assert len(retrieved_driver_config_rules) == 0 DEVICE_WITH_CONFIG_RULES = copy.deepcopy(DEVICE) with open(DATA_FILE_CONFIG_RULES, 'r', encoding='UTF-8') as f: config_rules = format_custom_config_rules(json.load(f)) DEVICE_WITH_CONFIG_RULES['device_config']['config_rules'].extend(config_rules) device_client.ConfigureDevice(Device(**DEVICE_WITH_CONFIG_RULES)) retrieved_osu_tunnels = ietf_actn_client.get('/restconf/v2/data/ietf-te:te/tunnels') retrieved_osu_tunnels = retrieved_osu_tunnels.json LOGGER.info('osu_tunnels = {:s}'.format(str(retrieved_osu_tunnels))) with open(DATA_FILE_EXPECTED_OSU_TUNNELS, 'r', encoding='UTF-8') as f: expected_osu_tunnels = json.load(f) osu_tunnels_diff = deepdiff.DeepDiff(expected_osu_tunnels, retrieved_osu_tunnels) if len(osu_tunnels_diff) > 0: LOGGER.error('POST OSU TUNNELS - Differences:\n{:s}'.format(str(osu_tunnels_diff.pretty()))) assert len(osu_tunnels_diff) == 0 retrieved_etht_services = ietf_actn_client.get('/restconf/v2/data/ietf-eth-tran-service:etht-svc') retrieved_etht_services = retrieved_etht_services.json LOGGER.info('etht_services = {:s}'.format(str(retrieved_etht_services))) with open(DATA_FILE_EXPECTED_ETHT_SERVICES, 'r', encoding='UTF-8') as f: expected_etht_services = json.load(f) etht_services_diff = deepdiff.DeepDiff(expected_etht_services, retrieved_etht_services) if len(etht_services_diff) > 0: LOGGER.error('POST ETHT SERVICES - Differences:\n{:s}'.format(str(etht_services_diff.pretty()))) assert len(etht_services_diff) == 0 retrieved_driver_config_rules = sorted(driver.GetConfig(), key=operator.itemgetter(0)) LOGGER.info('driver_config = {:s}'.format(str(retrieved_driver_config_rules))) retrieved_driver_config_rules = [ {'action': 1, 'custom': {'resource_key': resource_key, 'resource_value': resource_value}} for resource_key, resource_value in retrieved_driver_config_rules if resource_key != '/endpoints/endpoint[mgmt]' ] with open(DATA_FILE_CONFIG_RULES, 'r', encoding='UTF-8') as f: expected_driver_config_rules = sorted(json.load(f), key=lambda cr: cr['custom']['resource_key']) driver_config_rules_diff = deepdiff.DeepDiff(expected_driver_config_rules, retrieved_driver_config_rules) if len(driver_config_rules_diff) > 0: LOGGER.error('POST DRIVER CONFIG RULES - Differences:\n{:s}'.format(str(driver_config_rules_diff.pretty()))) assert len(driver_config_rules_diff) == 0 def test_device_ietf_actn_deconfigure( device_client : DeviceClient, # pylint: disable=redefined-outer-name device_service : DeviceService, # pylint: disable=redefined-outer-name ietf_actn_sdn_ctrl : GenericRestServer, # pylint: disable=redefined-outer-name ) -> None: ietf_actn_client = ietf_actn_sdn_ctrl.app.test_client() driver_instance_cache = device_service.device_servicer.driver_instance_cache driver : _Driver = driver_instance_cache.get(DEVICE_UUID) assert driver is not None retrieved_osu_tunnels = ietf_actn_client.get('/restconf/v2/data/ietf-te:te/tunnels') retrieved_osu_tunnels = retrieved_osu_tunnels.json LOGGER.info('osu_tunnels = {:s}'.format(str(retrieved_osu_tunnels))) with open(DATA_FILE_EXPECTED_OSU_TUNNELS, 'r', encoding='UTF-8') as f: expected_osu_tunnels = json.load(f) osu_tunnels_diff = deepdiff.DeepDiff(expected_osu_tunnels, retrieved_osu_tunnels) if len(osu_tunnels_diff) > 0: LOGGER.error('PRE OSU TUNNELS - Differences:\n{:s}'.format(str(osu_tunnels_diff.pretty()))) assert len(osu_tunnels_diff) == 0 retrieved_etht_services = ietf_actn_client.get('/restconf/v2/data/ietf-eth-tran-service:etht-svc') retrieved_etht_services = retrieved_etht_services.json LOGGER.info('etht_services = {:s}'.format(str(retrieved_etht_services))) with open(DATA_FILE_EXPECTED_ETHT_SERVICES, 'r', encoding='UTF-8') as f: expected_etht_services = json.load(f) etht_services_diff = deepdiff.DeepDiff(expected_etht_services, retrieved_etht_services) if len(etht_services_diff) > 0: LOGGER.error('PRE ETHT SERVICES - Differences:\n{:s}'.format(str(etht_services_diff.pretty()))) assert len(etht_services_diff) == 0 retrieved_driver_config_rules = sorted(driver.GetConfig(), key=operator.itemgetter(0)) LOGGER.info('driver_config = {:s}'.format(str(retrieved_driver_config_rules))) retrieved_driver_config_rules = [ {'action': 1, 'custom': {'resource_key': resource_key, 'resource_value': resource_value}} for resource_key, resource_value in retrieved_driver_config_rules if resource_key != '/endpoints/endpoint[mgmt]' ] with open(DATA_FILE_CONFIG_RULES, 'r', encoding='UTF-8') as f: expected_driver_config_rules = sorted(json.load(f), key=lambda cr: cr['custom']['resource_key']) driver_config_rules_diff = deepdiff.DeepDiff(expected_driver_config_rules, retrieved_driver_config_rules) if len(driver_config_rules_diff) > 0: LOGGER.error('PRE DRIVER CONFIG RULES - Differences:\n{:s}'.format(str(driver_config_rules_diff.pretty()))) assert len(driver_config_rules_diff) == 0 DEVICE_WITH_DECONFIG_RULES = copy.deepcopy(DEVICE) with open(DATA_FILE_DECONFIG_RULES, 'r', encoding='UTF-8') as f: deconfig_rules = format_custom_config_rules(json.load(f)) DEVICE_WITH_DECONFIG_RULES['device_config']['config_rules'].extend(deconfig_rules) device_client.ConfigureDevice(Device(**DEVICE_WITH_DECONFIG_RULES)) retrieved_osu_tunnels = ietf_actn_client.get('/restconf/v2/data/ietf-te:te/tunnels') retrieved_osu_tunnels = retrieved_osu_tunnels.json LOGGER.info('osu_tunnels = {:s}'.format(str(retrieved_osu_tunnels))) expected_osu_tunnels = {'ietf-te:tunnel': []} osu_tunnels_diff = deepdiff.DeepDiff(expected_osu_tunnels, retrieved_osu_tunnels) if len(osu_tunnels_diff) > 0: LOGGER.error('POST OSU TUNNELS - Differences:\n{:s}'.format(str(osu_tunnels_diff.pretty()))) assert len(osu_tunnels_diff) == 0 retrieved_etht_services = ietf_actn_client.get('/restconf/v2/data/ietf-eth-tran-service:etht-svc') retrieved_etht_services = retrieved_etht_services.json LOGGER.info('etht_services = {:s}'.format(str(retrieved_etht_services))) expected_etht_services = {'ietf-eth-tran-service:etht-svc': {'etht-svc-instances': []}} etht_services_diff = deepdiff.DeepDiff(expected_etht_services, retrieved_etht_services) if len(etht_services_diff) > 0: LOGGER.error('POST ETHT SERVICES - Differences:\n{:s}'.format(str(etht_services_diff.pretty()))) assert len(etht_services_diff) == 0 retrieved_driver_config_rules = sorted(driver.GetConfig(), key=operator.itemgetter(0)) LOGGER.info('retrieved_driver_config_rules = {:s}'.format(str(retrieved_driver_config_rules))) assert isinstance(retrieved_driver_config_rules, list) retrieved_driver_config_rules = [ (resource_key, resource_value) for resource_key, resource_value in retrieved_driver_config_rules if resource_key != '/endpoints/endpoint[mgmt]' ] if len(retrieved_driver_config_rules) > 0: LOGGER.error('POST DRIVER CONFIG RULES - Differences:\n{:s}'.format(str(retrieved_driver_config_rules))) assert len(retrieved_driver_config_rules) == 0 def test_device_ietf_actn_delete( device_client : DeviceClient, # pylint: disable=redefined-outer-name device_service : DeviceService, # pylint: disable=redefined-outer-name ietf_actn_sdn_ctrl : GenericRestServer, # pylint: disable=redefined-outer-name ) -> None: device_client.DeleteDevice(DeviceId(**DEVICE_ID)) driver_instance_cache = device_service.device_servicer.driver_instance_cache driver : _Driver = driver_instance_cache.get(DEVICE_UUID, {}) assert driver is None