Loading src/tests/ofc22/deploy_specs.sh +1 −1 Original line number Diff line number Diff line Loading @@ -7,7 +7,7 @@ export TFS_REGISTRY_IMAGE="http://localhost:32000/tfs/" # interdomain slice pathcomp dlt # dbscanserving opticalattackmitigator opticalattackdetector # l3_attackmitigator l3_centralizedattackdetector l3_distributedattackdetector export TFS_COMPONENTS="context device automation monitoring pathcomp service slice compute webui" export TFS_COMPONENTS="context device monitoring pathcomp service slice compute webui" # automation # Set the tag you want to use for your images. export TFS_IMAGE_TAG="dev" Loading src/tests/ofc22/descriptors_emulated.json +50 −37 Original line number Diff line number Diff line Loading @@ -9,70 +9,83 @@ "topologies": [ { "topology_id": {"topology_uuid": {"uuid": "admin"}, "context_id": {"context_uuid": {"uuid": "admin"}}}, "device_ids": [], "link_ids": [] "device_ids": [ {"device_uuid": {"uuid": "R1-EMU"}}, {"device_uuid": {"uuid": "R2-EMU"}}, {"device_uuid": {"uuid": "R3-EMU"}}, {"device_uuid": {"uuid": "R4-EMU"}}, {"device_uuid": {"uuid": "O1-OLS"}} ], "link_ids": [ {"link_uuid": {"uuid": "R1-EMU/13/0/0==O1-OLS/aade6001-f00b-5e2f-a357-6a0a9d3de870"}}, {"link_uuid": {"uuid": "R2-EMU/13/0/0==O1-OLS/eb287d83-f05e-53ec-ab5a-adf6bd2b5418"}}, {"link_uuid": {"uuid": "R3-EMU/13/0/0==O1-OLS/0ef74f99-1acc-57bd-ab9d-4b958b06c513"}}, {"link_uuid": {"uuid": "R4-EMU/13/0/0==O1-OLS/50296d99-58cc-5ce7-82f5-fc8ee4eec2ec"}} ] } ], "devices": [ { "device_id": {"device_uuid": {"uuid": "R1-EMU"}}, "device_type": "emu-packet-router", "device_id": {"device_uuid": {"uuid": "R1-EMU"}}, "device_type": "emu-packet-router", "device_operational_status": 1, "device_drivers": [0], "device_endpoints": [], "device_config": {"config_rules": [ {"action": 1, "custom": {"resource_key": "_connect/address", "resource_value": "127.0.0.1"}}, {"action": 1, "custom": {"resource_key": "_connect/port", "resource_value": "0"}}, {"action": 1, "custom": {"resource_key": "_connect/settings", "resource_value": "{\"endpoints\": [{\"sample_types\": [], \"type\": \"optical\", \"uuid\": \"13/0/0\"}, {\"sample_types\": [101, 102, 201, 202], \"type\": \"copper\", \"uuid\": \"13/1/2\"}]}"}} ]}, "device_operational_status": 1, "device_drivers": [0], "device_endpoints": [] {"action": 1, "custom": {"resource_key": "_connect/settings", "resource_value": {"endpoints": [ {"uuid": "13/0/0", "type": "optical", "sample_types": []}, {"uuid": "13/1/2", "type": "copper", "sample_types": [101, 102, 201, 202]} ]}}} ]} }, { "device_id": {"device_uuid": {"uuid": "R2-EMU"}}, "device_type": "emu-packet-router", "device_id": {"device_uuid": {"uuid": "R2-EMU"}}, "device_type": "emu-packet-router", "device_operational_status": 1, "device_drivers": [0], "device_endpoints": [], "device_config": {"config_rules": [ {"action": 1, "custom": {"resource_key": "_connect/address", "resource_value": "127.0.0.1"}}, {"action": 1, "custom": {"resource_key": "_connect/port", "resource_value": "0"}}, {"action": 1, "custom": {"resource_key": "_connect/settings", "resource_value": "{\"endpoints\": [{\"sample_types\": [], \"type\": \"optical\", \"uuid\": \"13/0/0\"}, {\"sample_types\": [101, 102, 201, 202], \"type\": \"copper\", \"uuid\": \"13/1/2\"}]}"}} ]}, "device_operational_status": 1, "device_drivers": [0], "device_endpoints": [] {"action": 1, "custom": {"resource_key": "_connect/settings", "resource_value": {"endpoints": [ {"uuid": "13/0/0", "type": "optical", "sample_types": []}, {"uuid": "13/1/2", "type": "copper", "sample_types": [101, 102, 201, 202]} ]}}} ]} }, { "device_id": {"device_uuid": {"uuid": "R3-EMU"}}, "device_type": "emu-packet-router", "device_id": {"device_uuid": {"uuid": "R3-EMU"}}, "device_type": "emu-packet-router", "device_operational_status": 1, "device_drivers": [0], "device_endpoints": [], "device_config": {"config_rules": [ {"action": 1, "custom": {"resource_key": "_connect/address", "resource_value": "127.0.0.1"}}, {"action": 1, "custom": {"resource_key": "_connect/port", "resource_value": "0"}}, {"action": 1, "custom": {"resource_key": "_connect/settings", "resource_value": "{\"endpoints\": [{\"sample_types\": [], \"type\": \"optical\", \"uuid\": \"13/0/0\"}, {\"sample_types\": [101, 102, 201, 202], \"type\": \"copper\", \"uuid\": \"13/1/2\"}]}"}} ]}, "device_operational_status": 1, "device_drivers": [0], "device_endpoints": [] {"action": 1, "custom": {"resource_key": "_connect/settings", "resource_value": {"endpoints": [ {"uuid": "13/0/0", "type": "optical", "sample_types": []}, {"uuid": "13/1/2", "type": "copper", "sample_types": [101, 102, 201, 202]} ]}}} ]} }, { "device_id": {"device_uuid": {"uuid": "R4-EMU"}}, "device_type": "emu-packet-router", "device_id": {"device_uuid": {"uuid": "R4-EMU"}}, "device_type": "emu-packet-router", "device_operational_status": 1, "device_drivers": [0], "device_endpoints": [], "device_config": {"config_rules": [ {"action": 1, "custom": {"resource_key": "_connect/address", "resource_value": "127.0.0.1"}}, {"action": 1, "custom": {"resource_key": "_connect/port", "resource_value": "0"}}, {"action": 1, "custom": {"resource_key": "_connect/settings", "resource_value": "{\"endpoints\": [{\"sample_types\": [], \"type\": \"optical\", \"uuid\": \"13/0/0\"}, {\"sample_types\": [101, 102, 201, 202], \"type\": \"copper\", \"uuid\": \"13/1/2\"}]}"}} ]}, "device_operational_status": 1, "device_drivers": [0], "device_endpoints": [] {"action": 1, "custom": {"resource_key": "_connect/settings", "resource_value": {"endpoints": [ {"uuid": "13/0/0", "type": "optical", "sample_types": []}, {"uuid": "13/1/2", "type": "copper", "sample_types": [101, 102, 201, 202]} ]}}} ]} }, { "device_id": {"device_uuid": {"uuid": "O1-OLS"}}, "device_type": "emu-open-line-system", "device_id": {"device_uuid": {"uuid": "O1-OLS"}}, "device_type": "emu-open-line-system", "device_operational_status": 1, "device_drivers": [0], "device_endpoints": [], "device_config": {"config_rules": [ {"action": 1, "custom": {"resource_key": "_connect/address", "resource_value": "127.0.0.1"}}, {"action": 1, "custom": {"resource_key": "_connect/port", "resource_value": "0"}}, {"action": 1, "custom": {"resource_key": "_connect/settings", "resource_value": "{\"endpoints\": [{\"sample_types\": [], \"type\": \"optical\", \"uuid\": \"aade6001-f00b-5e2f-a357-6a0a9d3de870\"}, {\"sample_types\": [], \"type\": \"optical\", \"uuid\": \"eb287d83-f05e-53ec-ab5a-adf6bd2b5418\"}, {\"sample_types\": [], \"type\": \"optical\", \"uuid\": \"0ef74f99-1acc-57bd-ab9d-4b958b06c513\"}, {\"sample_types\": [], \"type\": \"optical\", \"uuid\": \"50296d99-58cc-5ce7-82f5-fc8ee4eec2ec\"}]}"}} ]}, "device_operational_status": 1, "device_drivers": [0], "device_endpoints": [] {"action": 1, "custom": {"resource_key": "_connect/settings", "resource_value": {"endpoints": [ {"uuid": "aade6001-f00b-5e2f-a357-6a0a9d3de870", "type": "optical", "sample_types": []}, {"uuid": "eb287d83-f05e-53ec-ab5a-adf6bd2b5418", "type": "optical", "sample_types": []}, {"uuid": "0ef74f99-1acc-57bd-ab9d-4b958b06c513", "type": "optical", "sample_types": []}, {"uuid": "50296d99-58cc-5ce7-82f5-fc8ee4eec2ec", "type": "optical", "sample_types": []} ]}}} ]} } ], "links": [ Loading src/tests/ofc22/tests/Fixtures.py +38 −0 Original line number Diff line number Diff line Loading @@ -17,6 +17,44 @@ from common.Settings import get_setting from compute.tests.mock_osm.MockOSM import MockOSM from .Objects import WIM_MAPPING, WIM_PASSWORD, WIM_USERNAME # Copyright 2021-2023 H2020 TeraFlow (https://www.teraflow-h2020.eu/) # # 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 pytest from common.Settings import get_setting from compute.tests.mock_osm.MockOSM import MockOSM from context.client.ContextClient import ContextClient from device.client.DeviceClient import DeviceClient from monitoring.client.MonitoringClient import MonitoringClient @pytest.fixture(scope='session') def context_client(): _client = ContextClient() yield _client _client.close() @pytest.fixture(scope='session') def device_client(): _client = DeviceClient() yield _client _client.close() @pytest.fixture(scope='session') def monitoring_client(): _client = MonitoringClient() yield _client _client.close() @pytest.fixture(scope='session') def osm_wim(): Loading src/tests/ofc22/tests/Objects.py +33 −106 Original line number Diff line number Diff line Loading @@ -19,33 +19,40 @@ from typing import Dict, List, Tuple from common.Constants import DEFAULT_CONTEXT_UUID, DEFAULT_TOPOLOGY_UUID from common.tools.object_factory.Context import json_context, json_context_id from common.tools.object_factory.Device import ( json_device_connect_rules, json_device_emulated_connect_rules, json_device_emulated_packet_router_disabled, json_device_emulated_tapi_disabled, json_device_id, json_device_packetrouter_disabled, json_device_tapi_disabled) from common.tools.object_factory.EndPoint import json_endpoint, json_endpoint_id from common.tools.object_factory.Link import json_link, json_link_id from common.tools.object_factory.Topology import json_topology, json_topology_id from common.proto.kpi_sample_types_pb2 import KpiSampleType import os, uuid from common.Constants import DEFAULT_CONTEXT_UUID, DEFAULT_TOPOLOGY_UUID from common.tools.object_factory.Context import json_context, json_context_id from common.tools.object_factory.Device import ( json_device_emulated_connect_rules, json_device_emulated_datacenter_disabled, json_device_emulated_packet_router_disabled, json_device_emulated_tapi_disabled, json_device_id) from common.tools.object_factory.EndPoint import json_endpoints json_device_connect_rules, json_device_emulated_connect_rules, json_device_emulated_datacenter_disabled, json_device_emulated_packet_router_disabled, json_device_emulated_tapi_disabled, json_device_id, json_device_packetrouter_disabled, json_device_tapi_disabled) from common.tools.object_factory.EndPoint import json_endpoint, json_endpoint_id, json_endpoints from common.tools.object_factory.Link import get_link_uuid, json_link, json_link_id from common.tools.object_factory.Topology import json_topology, json_topology_id from common.tools.object_factory.Service import get_service_uuid, json_service_l3nm_planned from common.tools.object_factory.Topology import json_topology, json_topology_id from common.proto.kpi_sample_types_pb2 import KpiSampleType # ----- Device Credentials and Settings -------------------------------------------------------------------------------- # if true, Device component is present and will infeer the endpoints from connect-rules # if false, Device component is not present and device objects must contain preconfigured endpoints ADD_CONNECT_RULES_TO_DEVICES = os.environ.get('ADD_CONNECT_RULES_TO_DEVICES', 'True') ADD_CONNECT_RULES_TO_DEVICES = ADD_CONNECT_RULES_TO_DEVICES.upper() in {'T', 'TRUE', '1', 'Y', 'YES'} try: from .Credentials import DEVICE_R1_ADDRESS, DEVICE_R1_PORT, DEVICE_R1_USERNAME, DEVICE_R1_PASSWORD from .Credentials import DEVICE_R3_ADDRESS, DEVICE_R3_PORT, DEVICE_R3_USERNAME, DEVICE_R3_PASSWORD from .Credentials import DEVICE_O1_ADDRESS, DEVICE_O1_PORT USE_REAL_DEVICES = True # Use real devices except ImportError: USE_REAL_DEVICES = False # Use emulated devices DEVICE_R1_ADDRESS, DEVICE_R1_PORT, DEVICE_R1_USERNAME, DEVICE_R1_PASSWORD = '0.0.0.0', 830, 'admin', 'admin' DEVICE_R3_ADDRESS, DEVICE_R3_PORT, DEVICE_R3_USERNAME, DEVICE_R3_PASSWORD = '0.0.0.0', 830, 'admin', 'admin' DEVICE_O1_ADDRESS, DEVICE_O1_PORT = '0.0.0.0', 4900 #USE_REAL_DEVICES = False # Uncomment to force to use emulated devices if not USE_REAL_DEVICES: json_device_packetrouter_disabled = json_device_emulated_packet_router_disabled json_device_tapi_disabled = json_device_emulated_tapi_disabled def compose_router(device_uuid, endpoint_uuids, topology_id=None): device_id = json_device_id(device_uuid) Loading Loading @@ -74,12 +81,6 @@ def compose_datacenter(device_uuid, endpoint_uuids, topology_id=None): device = json_device_emulated_datacenter_disabled(device_uuid, config_rules=config_rules, endpoints=j_endpoints) return device_id, endpoints, device def compose_link(endpoint_a, endpoint_z): link_uuid = get_link_uuid(endpoint_a['endpoint_id'], endpoint_z['endpoint_id']) link_id = json_link_id(link_uuid) link = json_link(link_uuid, [endpoint_a['endpoint_id'], endpoint_z['endpoint_id']]) return link_id, link def compose_service(endpoint_a, endpoint_z, constraints=[]): service_uuid = get_service_uuid(endpoint_a['endpoint_id'], endpoint_z['endpoint_id']) endpoint_ids = [endpoint_a['endpoint_id'], endpoint_z['endpoint_id']] Loading @@ -91,42 +92,10 @@ CONTEXT_ID = json_context_id(DEFAULT_CONTEXT_UUID) CONTEXT = json_context(DEFAULT_CONTEXT_UUID) # ----- Topology ------------------------------------------------------------------------------------------------------- TOPOLOGY_ID = json_topology_id(DEFAULT_TOPOLOGY_UUID, context_id=CONTEXT_ID) TOPOLOGY = json_topology(DEFAULT_TOPOLOGY_UUID, context_id=CONTEXT_ID) # ----- Domains -------------------------------------------------------------------------------------------------------- # Overall network topology TOPO_ADMIN_UUID = DEFAULT_TOPOLOGY_UUID TOPO_ADMIN_ID = json_topology_id(TOPO_ADMIN_UUID, context_id=CONTEXT_ID) TOPO_ADMIN = json_topology(TOPO_ADMIN_UUID, context_id=CONTEXT_ID) # DataCenter #1 Network TOPO_DC1_UUID = 'DC1' TOPO_DC1_ID = json_topology_id(TOPO_DC1_UUID, context_id=CONTEXT_ID) TOPO_DC1 = json_topology(TOPO_DC1_UUID, context_id=CONTEXT_ID) # DataCenter #2 Network TOPO_DC2_UUID = 'DC2' TOPO_DC2_ID = json_topology_id(TOPO_DC2_UUID, context_id=CONTEXT_ID) TOPO_DC2 = json_topology(TOPO_DC2_UUID, context_id=CONTEXT_ID) # CellSite #1 Network TOPO_CS1_UUID = 'CS1' TOPO_CS1_ID = json_topology_id(TOPO_CS1_UUID, context_id=CONTEXT_ID) TOPO_CS1 = json_topology(TOPO_CS1_UUID, context_id=CONTEXT_ID) # CellSite #2 Network TOPO_CS2_UUID = 'CS2' TOPO_CS2_ID = json_topology_id(TOPO_CS2_UUID, context_id=CONTEXT_ID) TOPO_CS2 = json_topology(TOPO_CS2_UUID, context_id=CONTEXT_ID) # Transport Network Network TOPO_TN_UUID = 'TN' TOPO_TN_ID = json_topology_id(TOPO_TN_UUID, context_id=CONTEXT_ID) TOPO_TN = json_topology(TOPO_TN_UUID, context_id=CONTEXT_ID) # ----- Monitoring Samples --------------------------------------------------------------------------------------------- PACKET_PORT_SAMPLE_TYPES = [ KpiSampleType.KPISAMPLETYPE_PACKETS_TRANSMITTED, Loading @@ -135,52 +104,7 @@ PACKET_PORT_SAMPLE_TYPES = [ KpiSampleType.KPISAMPLETYPE_BYTES_RECEIVED, ] # ----- Device Credentials and Settings -------------------------------------------------------------------------------- try: from .Credentials import DEVICE_R1_ADDRESS, DEVICE_R1_PORT, DEVICE_R1_USERNAME, DEVICE_R1_PASSWORD from .Credentials import DEVICE_R3_ADDRESS, DEVICE_R3_PORT, DEVICE_R3_USERNAME, DEVICE_R3_PASSWORD from .Credentials import DEVICE_O1_ADDRESS, DEVICE_O1_PORT USE_REAL_DEVICES = True # Use real devices except ImportError: USE_REAL_DEVICES = False # Use emulated devices DEVICE_R1_ADDRESS = '0.0.0.0' DEVICE_R1_PORT = 830 DEVICE_R1_USERNAME = 'admin' DEVICE_R1_PASSWORD = 'admin' DEVICE_R3_ADDRESS = '0.0.0.0' DEVICE_R3_PORT = 830 DEVICE_R3_USERNAME = 'admin' DEVICE_R3_PASSWORD = 'admin' DEVICE_O1_ADDRESS = '0.0.0.0' DEVICE_O1_PORT = 4900 #USE_REAL_DEVICES = False # Uncomment to force to use emulated devices def json_endpoint_ids(device_id : Dict, endpoint_descriptors : List[Tuple[str, str, List[int]]]): return [ json_endpoint_id(device_id, ep_uuid, topology_id=None) for ep_uuid, _, _ in endpoint_descriptors ] def json_endpoints(device_id : Dict, endpoint_descriptors : List[Tuple[str, str, List[int]]]): return [ json_endpoint(device_id, ep_uuid, ep_type, topology_id=None, kpi_sample_types=ep_sample_types) for ep_uuid, ep_type, ep_sample_types in endpoint_descriptors ] def get_link_uuid(a_device_id : Dict, a_endpoint_id : Dict, z_device_id : Dict, z_endpoint_id : Dict) -> str: return '{:s}/{:s}=={:s}/{:s}'.format( a_device_id['device_uuid']['uuid'], a_endpoint_id['endpoint_uuid']['uuid'], z_device_id['device_uuid']['uuid'], z_endpoint_id['endpoint_uuid']['uuid']) # ----- Devices -------------------------------------------------------------------------------------------------------- if not USE_REAL_DEVICES: json_device_packetrouter_disabled = json_device_emulated_packet_router_disabled json_device_tapi_disabled = json_device_emulated_tapi_disabled DEVICE_R1_UUID = 'R1-EMU' DEVICE_R1_TIMEOUT = 120 Loading @@ -191,11 +115,19 @@ DEVICE_R1_ENDPOINT_IDS = json_endpoint_ids(DEVICE_R1_ID, DEVICE_R1_ENDPOINT_DEF DEVICE_R1 = json_device_packetrouter_disabled(DEVICE_R1_UUID) ENDPOINT_ID_R1_13_0_0 = DEVICE_R1_ENDPOINT_IDS[0] ENDPOINT_ID_R1_13_1_2 = DEVICE_R1_ENDPOINT_IDS[1] DEVICE_R1_CONNECT_RULES = json_device_connect_rules(DEVICE_R1_ADDRESS, DEVICE_R1_PORT, { DEVICE_R1_CONNECT_RULES = json_device_connect_rules(DEVICE_R1_ADDRESS, DEVICE_R1_PORT, ) if USE_REAL_DEVICES else json_device_emulated_connect_rules(DEVICE_R1_ENDPOINT_DEFS) def json_device_connect_rules(address : str, port : int, settings : Dict = {}): return [ json_config_rule_set('_connect/address', address), json_config_rule_set('_connect/port', port), json_config_rule_set('_connect/settings', { 'username': DEVICE_R1_USERNAME, 'password': DEVICE_R1_PASSWORD, 'timeout' : DEVICE_R1_TIMEOUT, }) if USE_REAL_DEVICES else json_device_emulated_connect_rules(DEVICE_R1_ENDPOINT_DEFS) }), ] DEVICE_R2_UUID = 'R2-EMU' Loading Loading @@ -258,11 +190,6 @@ DEVICE_O1_CONNECT_RULES = json_device_connect_rules(DEVICE_O1_ADDRESS, DEVICE_O1 # ----- Devices -------------------------------------------------------------------------------------------------------- # DataCenters DEV_DC1GW_ID, DEV_DC1GW_EPS, DEV_DC1GW = compose_datacenter('DC1-GW', ['eth1', 'eth2', 'int']) DEV_DC2GW_ID, DEV_DC2GW_EPS, DEV_DC2GW = compose_datacenter('DC2-GW', ['eth1', 'eth2', 'int']) # CellSites DEV_CS1GW1_ID, DEV_CS1GW1_EPS, DEV_CS1GW1 = compose_router('CS1-GW1', ['10/1', '1/1', '1/2']) DEV_CS1GW2_ID, DEV_CS1GW2_EPS, DEV_CS1GW2 = compose_router('CS1-GW2', ['10/1', '1/1', '1/2']) DEV_CS2GW1_ID, DEV_CS2GW1_EPS, DEV_CS2GW1 = compose_router('CS2-GW1', ['10/1', '1/1', '1/2']) Loading src/tests/ofc22/tests/test_functional_bootstrap.py +23 −37 Original line number Diff line number Diff line Loading @@ -14,27 +14,25 @@ # See the License for the specific language governing permissions and # limitations under the License. import copy, logging, pytest from common.Settings import get_setting import logging from common.proto.context_pb2 import ContextId, Empty from common.proto.monitoring_pb2 import KpiDescriptorList from common.tests.EventTools import EVENT_CREATE, EVENT_UPDATE, check_events from common.tools.object_factory.Context import json_context_id from common.tools.object_factory.Device import json_device_id from common.tools.object_factory.Link import json_link_id from common.tools.object_factory.Topology import json_topology_id from common.tests.LoadScenario import load_scenario_from_descriptor from context.client.ContextClient import ContextClient from monitoring.client.MonitoringClient import MonitoringClient from context.client.EventsCollector import EventsCollector from common.proto.context_pb2 import Context, ContextId, Device, Empty, Link, Topology from device.client.DeviceClient import DeviceClient from monitoring.client.MonitoringClient import MonitoringClient from tests.Fixtures import context_client, device_client, monitoring_client # pylint: disable=unused-import from .Objects import CONTEXT_ID, CONTEXTS, DEVICES, LINKS, TOPOLOGIES from tests.Fixtures import context_client, device_client, monitoring_client LOGGER = logging.getLogger(__name__) LOGGER.setLevel(logging.DEBUG) DESCRIPTOR_FILE = 'ofc22/descriptors_emulated.json' def test_scenario_empty(context_client : ContextClient): # pylint: disable=redefined-outer-name def test_scenario_empty( context_client : ContextClient, # pylint: disable=redefined-outer-name device_client : DeviceClient, # pylint: disable=redefined-outer-name ) -> None: # ----- List entities - Ensure database is empty ------------------------------------------------------------------- response = context_client.ListContexts(Empty()) assert len(response.contexts) == 0 Loading @@ -46,38 +44,26 @@ def test_scenario_empty(context_client : ContextClient): # pylint: disable=rede assert len(response.links) == 0 def test_prepare_environment(context_client : ContextClient): # pylint: disable=redefined-outer-name for context in CONTEXTS : context_client.SetContext (Context (**context )) for topology in TOPOLOGIES: context_client.SetTopology(Topology(**topology)) for device, connect_rules in DEVICES: device_with_connect_rules = copy.deepcopy(device) device_with_connect_rules['device_config']['config_rules'].extend(connect_rules) device_client.AddDevice(Device(**device_with_connect_rules)) for link in LINKS : context_client.SetLink (Link (**link )) # ----- Load Scenario ---------------------------------------------------------------------------------------------- load_scenario_from_descriptor(DESCRIPTOR_FILE, context_client, device_client, None, None) def test_scenario_ready(context_client : ContextClient): # pylint: disable=redefined-outer-name # ----- List entities - Ensure scenario is ready ------------------------------------------------------------------- response = context_client.ListContexts(Empty()) assert len(response.contexts) == len(CONTEXTS) #response = context_client.ListContexts(Empty()) #assert len(response.contexts) == len(CONTEXTS) response = context_client.ListTopologies(ContextId(**CONTEXT_ID)) assert len(response.topologies) == len(TOPOLOGIES) #response = context_client.ListTopologies(ContextId(**CONTEXT_ID)) #assert len(response.topologies) == len(TOPOLOGIES) response = context_client.ListDevices(Empty()) assert len(response.devices) == len(DEVICES) #response = context_client.ListDevices(Empty()) #assert len(response.devices) == len(DEVICES) response = context_client.ListLinks(Empty()) assert len(response.links) == len(LINKS) #response = context_client.ListLinks(Empty()) #assert len(response.links) == len(LINKS) #response = context_client.ListServices(ContextId(**CONTEXT_ID)) #assert len(response.services) == 0 response = context_client.ListServices(ContextId(**CONTEXT_ID)) assert len(response.services) == 0 def test_scenario_kpis_created(monitoring_client: MonitoringClient): """ Loading Loading
src/tests/ofc22/deploy_specs.sh +1 −1 Original line number Diff line number Diff line Loading @@ -7,7 +7,7 @@ export TFS_REGISTRY_IMAGE="http://localhost:32000/tfs/" # interdomain slice pathcomp dlt # dbscanserving opticalattackmitigator opticalattackdetector # l3_attackmitigator l3_centralizedattackdetector l3_distributedattackdetector export TFS_COMPONENTS="context device automation monitoring pathcomp service slice compute webui" export TFS_COMPONENTS="context device monitoring pathcomp service slice compute webui" # automation # Set the tag you want to use for your images. export TFS_IMAGE_TAG="dev" Loading
src/tests/ofc22/descriptors_emulated.json +50 −37 Original line number Diff line number Diff line Loading @@ -9,70 +9,83 @@ "topologies": [ { "topology_id": {"topology_uuid": {"uuid": "admin"}, "context_id": {"context_uuid": {"uuid": "admin"}}}, "device_ids": [], "link_ids": [] "device_ids": [ {"device_uuid": {"uuid": "R1-EMU"}}, {"device_uuid": {"uuid": "R2-EMU"}}, {"device_uuid": {"uuid": "R3-EMU"}}, {"device_uuid": {"uuid": "R4-EMU"}}, {"device_uuid": {"uuid": "O1-OLS"}} ], "link_ids": [ {"link_uuid": {"uuid": "R1-EMU/13/0/0==O1-OLS/aade6001-f00b-5e2f-a357-6a0a9d3de870"}}, {"link_uuid": {"uuid": "R2-EMU/13/0/0==O1-OLS/eb287d83-f05e-53ec-ab5a-adf6bd2b5418"}}, {"link_uuid": {"uuid": "R3-EMU/13/0/0==O1-OLS/0ef74f99-1acc-57bd-ab9d-4b958b06c513"}}, {"link_uuid": {"uuid": "R4-EMU/13/0/0==O1-OLS/50296d99-58cc-5ce7-82f5-fc8ee4eec2ec"}} ] } ], "devices": [ { "device_id": {"device_uuid": {"uuid": "R1-EMU"}}, "device_type": "emu-packet-router", "device_id": {"device_uuid": {"uuid": "R1-EMU"}}, "device_type": "emu-packet-router", "device_operational_status": 1, "device_drivers": [0], "device_endpoints": [], "device_config": {"config_rules": [ {"action": 1, "custom": {"resource_key": "_connect/address", "resource_value": "127.0.0.1"}}, {"action": 1, "custom": {"resource_key": "_connect/port", "resource_value": "0"}}, {"action": 1, "custom": {"resource_key": "_connect/settings", "resource_value": "{\"endpoints\": [{\"sample_types\": [], \"type\": \"optical\", \"uuid\": \"13/0/0\"}, {\"sample_types\": [101, 102, 201, 202], \"type\": \"copper\", \"uuid\": \"13/1/2\"}]}"}} ]}, "device_operational_status": 1, "device_drivers": [0], "device_endpoints": [] {"action": 1, "custom": {"resource_key": "_connect/settings", "resource_value": {"endpoints": [ {"uuid": "13/0/0", "type": "optical", "sample_types": []}, {"uuid": "13/1/2", "type": "copper", "sample_types": [101, 102, 201, 202]} ]}}} ]} }, { "device_id": {"device_uuid": {"uuid": "R2-EMU"}}, "device_type": "emu-packet-router", "device_id": {"device_uuid": {"uuid": "R2-EMU"}}, "device_type": "emu-packet-router", "device_operational_status": 1, "device_drivers": [0], "device_endpoints": [], "device_config": {"config_rules": [ {"action": 1, "custom": {"resource_key": "_connect/address", "resource_value": "127.0.0.1"}}, {"action": 1, "custom": {"resource_key": "_connect/port", "resource_value": "0"}}, {"action": 1, "custom": {"resource_key": "_connect/settings", "resource_value": "{\"endpoints\": [{\"sample_types\": [], \"type\": \"optical\", \"uuid\": \"13/0/0\"}, {\"sample_types\": [101, 102, 201, 202], \"type\": \"copper\", \"uuid\": \"13/1/2\"}]}"}} ]}, "device_operational_status": 1, "device_drivers": [0], "device_endpoints": [] {"action": 1, "custom": {"resource_key": "_connect/settings", "resource_value": {"endpoints": [ {"uuid": "13/0/0", "type": "optical", "sample_types": []}, {"uuid": "13/1/2", "type": "copper", "sample_types": [101, 102, 201, 202]} ]}}} ]} }, { "device_id": {"device_uuid": {"uuid": "R3-EMU"}}, "device_type": "emu-packet-router", "device_id": {"device_uuid": {"uuid": "R3-EMU"}}, "device_type": "emu-packet-router", "device_operational_status": 1, "device_drivers": [0], "device_endpoints": [], "device_config": {"config_rules": [ {"action": 1, "custom": {"resource_key": "_connect/address", "resource_value": "127.0.0.1"}}, {"action": 1, "custom": {"resource_key": "_connect/port", "resource_value": "0"}}, {"action": 1, "custom": {"resource_key": "_connect/settings", "resource_value": "{\"endpoints\": [{\"sample_types\": [], \"type\": \"optical\", \"uuid\": \"13/0/0\"}, {\"sample_types\": [101, 102, 201, 202], \"type\": \"copper\", \"uuid\": \"13/1/2\"}]}"}} ]}, "device_operational_status": 1, "device_drivers": [0], "device_endpoints": [] {"action": 1, "custom": {"resource_key": "_connect/settings", "resource_value": {"endpoints": [ {"uuid": "13/0/0", "type": "optical", "sample_types": []}, {"uuid": "13/1/2", "type": "copper", "sample_types": [101, 102, 201, 202]} ]}}} ]} }, { "device_id": {"device_uuid": {"uuid": "R4-EMU"}}, "device_type": "emu-packet-router", "device_id": {"device_uuid": {"uuid": "R4-EMU"}}, "device_type": "emu-packet-router", "device_operational_status": 1, "device_drivers": [0], "device_endpoints": [], "device_config": {"config_rules": [ {"action": 1, "custom": {"resource_key": "_connect/address", "resource_value": "127.0.0.1"}}, {"action": 1, "custom": {"resource_key": "_connect/port", "resource_value": "0"}}, {"action": 1, "custom": {"resource_key": "_connect/settings", "resource_value": "{\"endpoints\": [{\"sample_types\": [], \"type\": \"optical\", \"uuid\": \"13/0/0\"}, {\"sample_types\": [101, 102, 201, 202], \"type\": \"copper\", \"uuid\": \"13/1/2\"}]}"}} ]}, "device_operational_status": 1, "device_drivers": [0], "device_endpoints": [] {"action": 1, "custom": {"resource_key": "_connect/settings", "resource_value": {"endpoints": [ {"uuid": "13/0/0", "type": "optical", "sample_types": []}, {"uuid": "13/1/2", "type": "copper", "sample_types": [101, 102, 201, 202]} ]}}} ]} }, { "device_id": {"device_uuid": {"uuid": "O1-OLS"}}, "device_type": "emu-open-line-system", "device_id": {"device_uuid": {"uuid": "O1-OLS"}}, "device_type": "emu-open-line-system", "device_operational_status": 1, "device_drivers": [0], "device_endpoints": [], "device_config": {"config_rules": [ {"action": 1, "custom": {"resource_key": "_connect/address", "resource_value": "127.0.0.1"}}, {"action": 1, "custom": {"resource_key": "_connect/port", "resource_value": "0"}}, {"action": 1, "custom": {"resource_key": "_connect/settings", "resource_value": "{\"endpoints\": [{\"sample_types\": [], \"type\": \"optical\", \"uuid\": \"aade6001-f00b-5e2f-a357-6a0a9d3de870\"}, {\"sample_types\": [], \"type\": \"optical\", \"uuid\": \"eb287d83-f05e-53ec-ab5a-adf6bd2b5418\"}, {\"sample_types\": [], \"type\": \"optical\", \"uuid\": \"0ef74f99-1acc-57bd-ab9d-4b958b06c513\"}, {\"sample_types\": [], \"type\": \"optical\", \"uuid\": \"50296d99-58cc-5ce7-82f5-fc8ee4eec2ec\"}]}"}} ]}, "device_operational_status": 1, "device_drivers": [0], "device_endpoints": [] {"action": 1, "custom": {"resource_key": "_connect/settings", "resource_value": {"endpoints": [ {"uuid": "aade6001-f00b-5e2f-a357-6a0a9d3de870", "type": "optical", "sample_types": []}, {"uuid": "eb287d83-f05e-53ec-ab5a-adf6bd2b5418", "type": "optical", "sample_types": []}, {"uuid": "0ef74f99-1acc-57bd-ab9d-4b958b06c513", "type": "optical", "sample_types": []}, {"uuid": "50296d99-58cc-5ce7-82f5-fc8ee4eec2ec", "type": "optical", "sample_types": []} ]}}} ]} } ], "links": [ Loading
src/tests/ofc22/tests/Fixtures.py +38 −0 Original line number Diff line number Diff line Loading @@ -17,6 +17,44 @@ from common.Settings import get_setting from compute.tests.mock_osm.MockOSM import MockOSM from .Objects import WIM_MAPPING, WIM_PASSWORD, WIM_USERNAME # Copyright 2021-2023 H2020 TeraFlow (https://www.teraflow-h2020.eu/) # # 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 pytest from common.Settings import get_setting from compute.tests.mock_osm.MockOSM import MockOSM from context.client.ContextClient import ContextClient from device.client.DeviceClient import DeviceClient from monitoring.client.MonitoringClient import MonitoringClient @pytest.fixture(scope='session') def context_client(): _client = ContextClient() yield _client _client.close() @pytest.fixture(scope='session') def device_client(): _client = DeviceClient() yield _client _client.close() @pytest.fixture(scope='session') def monitoring_client(): _client = MonitoringClient() yield _client _client.close() @pytest.fixture(scope='session') def osm_wim(): Loading
src/tests/ofc22/tests/Objects.py +33 −106 Original line number Diff line number Diff line Loading @@ -19,33 +19,40 @@ from typing import Dict, List, Tuple from common.Constants import DEFAULT_CONTEXT_UUID, DEFAULT_TOPOLOGY_UUID from common.tools.object_factory.Context import json_context, json_context_id from common.tools.object_factory.Device import ( json_device_connect_rules, json_device_emulated_connect_rules, json_device_emulated_packet_router_disabled, json_device_emulated_tapi_disabled, json_device_id, json_device_packetrouter_disabled, json_device_tapi_disabled) from common.tools.object_factory.EndPoint import json_endpoint, json_endpoint_id from common.tools.object_factory.Link import json_link, json_link_id from common.tools.object_factory.Topology import json_topology, json_topology_id from common.proto.kpi_sample_types_pb2 import KpiSampleType import os, uuid from common.Constants import DEFAULT_CONTEXT_UUID, DEFAULT_TOPOLOGY_UUID from common.tools.object_factory.Context import json_context, json_context_id from common.tools.object_factory.Device import ( json_device_emulated_connect_rules, json_device_emulated_datacenter_disabled, json_device_emulated_packet_router_disabled, json_device_emulated_tapi_disabled, json_device_id) from common.tools.object_factory.EndPoint import json_endpoints json_device_connect_rules, json_device_emulated_connect_rules, json_device_emulated_datacenter_disabled, json_device_emulated_packet_router_disabled, json_device_emulated_tapi_disabled, json_device_id, json_device_packetrouter_disabled, json_device_tapi_disabled) from common.tools.object_factory.EndPoint import json_endpoint, json_endpoint_id, json_endpoints from common.tools.object_factory.Link import get_link_uuid, json_link, json_link_id from common.tools.object_factory.Topology import json_topology, json_topology_id from common.tools.object_factory.Service import get_service_uuid, json_service_l3nm_planned from common.tools.object_factory.Topology import json_topology, json_topology_id from common.proto.kpi_sample_types_pb2 import KpiSampleType # ----- Device Credentials and Settings -------------------------------------------------------------------------------- # if true, Device component is present and will infeer the endpoints from connect-rules # if false, Device component is not present and device objects must contain preconfigured endpoints ADD_CONNECT_RULES_TO_DEVICES = os.environ.get('ADD_CONNECT_RULES_TO_DEVICES', 'True') ADD_CONNECT_RULES_TO_DEVICES = ADD_CONNECT_RULES_TO_DEVICES.upper() in {'T', 'TRUE', '1', 'Y', 'YES'} try: from .Credentials import DEVICE_R1_ADDRESS, DEVICE_R1_PORT, DEVICE_R1_USERNAME, DEVICE_R1_PASSWORD from .Credentials import DEVICE_R3_ADDRESS, DEVICE_R3_PORT, DEVICE_R3_USERNAME, DEVICE_R3_PASSWORD from .Credentials import DEVICE_O1_ADDRESS, DEVICE_O1_PORT USE_REAL_DEVICES = True # Use real devices except ImportError: USE_REAL_DEVICES = False # Use emulated devices DEVICE_R1_ADDRESS, DEVICE_R1_PORT, DEVICE_R1_USERNAME, DEVICE_R1_PASSWORD = '0.0.0.0', 830, 'admin', 'admin' DEVICE_R3_ADDRESS, DEVICE_R3_PORT, DEVICE_R3_USERNAME, DEVICE_R3_PASSWORD = '0.0.0.0', 830, 'admin', 'admin' DEVICE_O1_ADDRESS, DEVICE_O1_PORT = '0.0.0.0', 4900 #USE_REAL_DEVICES = False # Uncomment to force to use emulated devices if not USE_REAL_DEVICES: json_device_packetrouter_disabled = json_device_emulated_packet_router_disabled json_device_tapi_disabled = json_device_emulated_tapi_disabled def compose_router(device_uuid, endpoint_uuids, topology_id=None): device_id = json_device_id(device_uuid) Loading Loading @@ -74,12 +81,6 @@ def compose_datacenter(device_uuid, endpoint_uuids, topology_id=None): device = json_device_emulated_datacenter_disabled(device_uuid, config_rules=config_rules, endpoints=j_endpoints) return device_id, endpoints, device def compose_link(endpoint_a, endpoint_z): link_uuid = get_link_uuid(endpoint_a['endpoint_id'], endpoint_z['endpoint_id']) link_id = json_link_id(link_uuid) link = json_link(link_uuid, [endpoint_a['endpoint_id'], endpoint_z['endpoint_id']]) return link_id, link def compose_service(endpoint_a, endpoint_z, constraints=[]): service_uuid = get_service_uuid(endpoint_a['endpoint_id'], endpoint_z['endpoint_id']) endpoint_ids = [endpoint_a['endpoint_id'], endpoint_z['endpoint_id']] Loading @@ -91,42 +92,10 @@ CONTEXT_ID = json_context_id(DEFAULT_CONTEXT_UUID) CONTEXT = json_context(DEFAULT_CONTEXT_UUID) # ----- Topology ------------------------------------------------------------------------------------------------------- TOPOLOGY_ID = json_topology_id(DEFAULT_TOPOLOGY_UUID, context_id=CONTEXT_ID) TOPOLOGY = json_topology(DEFAULT_TOPOLOGY_UUID, context_id=CONTEXT_ID) # ----- Domains -------------------------------------------------------------------------------------------------------- # Overall network topology TOPO_ADMIN_UUID = DEFAULT_TOPOLOGY_UUID TOPO_ADMIN_ID = json_topology_id(TOPO_ADMIN_UUID, context_id=CONTEXT_ID) TOPO_ADMIN = json_topology(TOPO_ADMIN_UUID, context_id=CONTEXT_ID) # DataCenter #1 Network TOPO_DC1_UUID = 'DC1' TOPO_DC1_ID = json_topology_id(TOPO_DC1_UUID, context_id=CONTEXT_ID) TOPO_DC1 = json_topology(TOPO_DC1_UUID, context_id=CONTEXT_ID) # DataCenter #2 Network TOPO_DC2_UUID = 'DC2' TOPO_DC2_ID = json_topology_id(TOPO_DC2_UUID, context_id=CONTEXT_ID) TOPO_DC2 = json_topology(TOPO_DC2_UUID, context_id=CONTEXT_ID) # CellSite #1 Network TOPO_CS1_UUID = 'CS1' TOPO_CS1_ID = json_topology_id(TOPO_CS1_UUID, context_id=CONTEXT_ID) TOPO_CS1 = json_topology(TOPO_CS1_UUID, context_id=CONTEXT_ID) # CellSite #2 Network TOPO_CS2_UUID = 'CS2' TOPO_CS2_ID = json_topology_id(TOPO_CS2_UUID, context_id=CONTEXT_ID) TOPO_CS2 = json_topology(TOPO_CS2_UUID, context_id=CONTEXT_ID) # Transport Network Network TOPO_TN_UUID = 'TN' TOPO_TN_ID = json_topology_id(TOPO_TN_UUID, context_id=CONTEXT_ID) TOPO_TN = json_topology(TOPO_TN_UUID, context_id=CONTEXT_ID) # ----- Monitoring Samples --------------------------------------------------------------------------------------------- PACKET_PORT_SAMPLE_TYPES = [ KpiSampleType.KPISAMPLETYPE_PACKETS_TRANSMITTED, Loading @@ -135,52 +104,7 @@ PACKET_PORT_SAMPLE_TYPES = [ KpiSampleType.KPISAMPLETYPE_BYTES_RECEIVED, ] # ----- Device Credentials and Settings -------------------------------------------------------------------------------- try: from .Credentials import DEVICE_R1_ADDRESS, DEVICE_R1_PORT, DEVICE_R1_USERNAME, DEVICE_R1_PASSWORD from .Credentials import DEVICE_R3_ADDRESS, DEVICE_R3_PORT, DEVICE_R3_USERNAME, DEVICE_R3_PASSWORD from .Credentials import DEVICE_O1_ADDRESS, DEVICE_O1_PORT USE_REAL_DEVICES = True # Use real devices except ImportError: USE_REAL_DEVICES = False # Use emulated devices DEVICE_R1_ADDRESS = '0.0.0.0' DEVICE_R1_PORT = 830 DEVICE_R1_USERNAME = 'admin' DEVICE_R1_PASSWORD = 'admin' DEVICE_R3_ADDRESS = '0.0.0.0' DEVICE_R3_PORT = 830 DEVICE_R3_USERNAME = 'admin' DEVICE_R3_PASSWORD = 'admin' DEVICE_O1_ADDRESS = '0.0.0.0' DEVICE_O1_PORT = 4900 #USE_REAL_DEVICES = False # Uncomment to force to use emulated devices def json_endpoint_ids(device_id : Dict, endpoint_descriptors : List[Tuple[str, str, List[int]]]): return [ json_endpoint_id(device_id, ep_uuid, topology_id=None) for ep_uuid, _, _ in endpoint_descriptors ] def json_endpoints(device_id : Dict, endpoint_descriptors : List[Tuple[str, str, List[int]]]): return [ json_endpoint(device_id, ep_uuid, ep_type, topology_id=None, kpi_sample_types=ep_sample_types) for ep_uuid, ep_type, ep_sample_types in endpoint_descriptors ] def get_link_uuid(a_device_id : Dict, a_endpoint_id : Dict, z_device_id : Dict, z_endpoint_id : Dict) -> str: return '{:s}/{:s}=={:s}/{:s}'.format( a_device_id['device_uuid']['uuid'], a_endpoint_id['endpoint_uuid']['uuid'], z_device_id['device_uuid']['uuid'], z_endpoint_id['endpoint_uuid']['uuid']) # ----- Devices -------------------------------------------------------------------------------------------------------- if not USE_REAL_DEVICES: json_device_packetrouter_disabled = json_device_emulated_packet_router_disabled json_device_tapi_disabled = json_device_emulated_tapi_disabled DEVICE_R1_UUID = 'R1-EMU' DEVICE_R1_TIMEOUT = 120 Loading @@ -191,11 +115,19 @@ DEVICE_R1_ENDPOINT_IDS = json_endpoint_ids(DEVICE_R1_ID, DEVICE_R1_ENDPOINT_DEF DEVICE_R1 = json_device_packetrouter_disabled(DEVICE_R1_UUID) ENDPOINT_ID_R1_13_0_0 = DEVICE_R1_ENDPOINT_IDS[0] ENDPOINT_ID_R1_13_1_2 = DEVICE_R1_ENDPOINT_IDS[1] DEVICE_R1_CONNECT_RULES = json_device_connect_rules(DEVICE_R1_ADDRESS, DEVICE_R1_PORT, { DEVICE_R1_CONNECT_RULES = json_device_connect_rules(DEVICE_R1_ADDRESS, DEVICE_R1_PORT, ) if USE_REAL_DEVICES else json_device_emulated_connect_rules(DEVICE_R1_ENDPOINT_DEFS) def json_device_connect_rules(address : str, port : int, settings : Dict = {}): return [ json_config_rule_set('_connect/address', address), json_config_rule_set('_connect/port', port), json_config_rule_set('_connect/settings', { 'username': DEVICE_R1_USERNAME, 'password': DEVICE_R1_PASSWORD, 'timeout' : DEVICE_R1_TIMEOUT, }) if USE_REAL_DEVICES else json_device_emulated_connect_rules(DEVICE_R1_ENDPOINT_DEFS) }), ] DEVICE_R2_UUID = 'R2-EMU' Loading Loading @@ -258,11 +190,6 @@ DEVICE_O1_CONNECT_RULES = json_device_connect_rules(DEVICE_O1_ADDRESS, DEVICE_O1 # ----- Devices -------------------------------------------------------------------------------------------------------- # DataCenters DEV_DC1GW_ID, DEV_DC1GW_EPS, DEV_DC1GW = compose_datacenter('DC1-GW', ['eth1', 'eth2', 'int']) DEV_DC2GW_ID, DEV_DC2GW_EPS, DEV_DC2GW = compose_datacenter('DC2-GW', ['eth1', 'eth2', 'int']) # CellSites DEV_CS1GW1_ID, DEV_CS1GW1_EPS, DEV_CS1GW1 = compose_router('CS1-GW1', ['10/1', '1/1', '1/2']) DEV_CS1GW2_ID, DEV_CS1GW2_EPS, DEV_CS1GW2 = compose_router('CS1-GW2', ['10/1', '1/1', '1/2']) DEV_CS2GW1_ID, DEV_CS2GW1_EPS, DEV_CS2GW1 = compose_router('CS2-GW1', ['10/1', '1/1', '1/2']) Loading
src/tests/ofc22/tests/test_functional_bootstrap.py +23 −37 Original line number Diff line number Diff line Loading @@ -14,27 +14,25 @@ # See the License for the specific language governing permissions and # limitations under the License. import copy, logging, pytest from common.Settings import get_setting import logging from common.proto.context_pb2 import ContextId, Empty from common.proto.monitoring_pb2 import KpiDescriptorList from common.tests.EventTools import EVENT_CREATE, EVENT_UPDATE, check_events from common.tools.object_factory.Context import json_context_id from common.tools.object_factory.Device import json_device_id from common.tools.object_factory.Link import json_link_id from common.tools.object_factory.Topology import json_topology_id from common.tests.LoadScenario import load_scenario_from_descriptor from context.client.ContextClient import ContextClient from monitoring.client.MonitoringClient import MonitoringClient from context.client.EventsCollector import EventsCollector from common.proto.context_pb2 import Context, ContextId, Device, Empty, Link, Topology from device.client.DeviceClient import DeviceClient from monitoring.client.MonitoringClient import MonitoringClient from tests.Fixtures import context_client, device_client, monitoring_client # pylint: disable=unused-import from .Objects import CONTEXT_ID, CONTEXTS, DEVICES, LINKS, TOPOLOGIES from tests.Fixtures import context_client, device_client, monitoring_client LOGGER = logging.getLogger(__name__) LOGGER.setLevel(logging.DEBUG) DESCRIPTOR_FILE = 'ofc22/descriptors_emulated.json' def test_scenario_empty(context_client : ContextClient): # pylint: disable=redefined-outer-name def test_scenario_empty( context_client : ContextClient, # pylint: disable=redefined-outer-name device_client : DeviceClient, # pylint: disable=redefined-outer-name ) -> None: # ----- List entities - Ensure database is empty ------------------------------------------------------------------- response = context_client.ListContexts(Empty()) assert len(response.contexts) == 0 Loading @@ -46,38 +44,26 @@ def test_scenario_empty(context_client : ContextClient): # pylint: disable=rede assert len(response.links) == 0 def test_prepare_environment(context_client : ContextClient): # pylint: disable=redefined-outer-name for context in CONTEXTS : context_client.SetContext (Context (**context )) for topology in TOPOLOGIES: context_client.SetTopology(Topology(**topology)) for device, connect_rules in DEVICES: device_with_connect_rules = copy.deepcopy(device) device_with_connect_rules['device_config']['config_rules'].extend(connect_rules) device_client.AddDevice(Device(**device_with_connect_rules)) for link in LINKS : context_client.SetLink (Link (**link )) # ----- Load Scenario ---------------------------------------------------------------------------------------------- load_scenario_from_descriptor(DESCRIPTOR_FILE, context_client, device_client, None, None) def test_scenario_ready(context_client : ContextClient): # pylint: disable=redefined-outer-name # ----- List entities - Ensure scenario is ready ------------------------------------------------------------------- response = context_client.ListContexts(Empty()) assert len(response.contexts) == len(CONTEXTS) #response = context_client.ListContexts(Empty()) #assert len(response.contexts) == len(CONTEXTS) response = context_client.ListTopologies(ContextId(**CONTEXT_ID)) assert len(response.topologies) == len(TOPOLOGIES) #response = context_client.ListTopologies(ContextId(**CONTEXT_ID)) #assert len(response.topologies) == len(TOPOLOGIES) response = context_client.ListDevices(Empty()) assert len(response.devices) == len(DEVICES) #response = context_client.ListDevices(Empty()) #assert len(response.devices) == len(DEVICES) response = context_client.ListLinks(Empty()) assert len(response.links) == len(LINKS) #response = context_client.ListLinks(Empty()) #assert len(response.links) == len(LINKS) #response = context_client.ListServices(ContextId(**CONTEXT_ID)) #assert len(response.services) == 0 response = context_client.ListServices(ContextId(**CONTEXT_ID)) assert len(response.services) == 0 def test_scenario_kpis_created(monitoring_client: MonitoringClient): """ Loading
