Loading src/tests/ofc25/tests/Helper.py 0 → 100644 +133 −0 Original line number Diff line number Diff line # Copyright 2022-2026 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 logging import os import time from typing import List, Set from common.Constants import DEFAULT_CONTEXT_NAME from common.proto.context_pb2 import ContextId, Empty, LinkTypeEnum, ServiceStatusEnum, ServiceTypeEnum from common.tools.grpc.Tools import grpc_message_to_json_string from common.tools.object_factory.Context import json_context_id LOGGER = logging.getLogger(__name__) LOGGER.setLevel(logging.DEBUG) ADMIN_CONTEXT_ID = ContextId(**json_context_id(DEFAULT_CONTEXT_NAME)) VIRTUAL_LINK_DESCRIPTORS = [ ('virtual_link_01.json', 'IP1/PORT-xe1==IP2/PORT-xe1'), ('virtual_link_02.json', 'IP1/PORT-xe2==IP2/PORT-xe2'), ('virtual_link_03.json', 'IP1/PORT-xe3==IP2/PORT-xe3'), ] DESCRIPTORS_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), '..', 'descriptors') def list_active_optical_services(context_client) -> List: response = context_client.ListServices(ADMIN_CONTEXT_ID) LOGGER.info('Services[%d] = %s', len(response.services), grpc_message_to_json_string(response)) active_optical_services = [] for service in response.services: assert service.service_type == ServiceTypeEnum.SERVICETYPE_OPTICAL_CONNECTIVITY assert service.service_status.service_status == ServiceStatusEnum.SERVICESTATUS_ACTIVE active_optical_services.append(service) return active_optical_services def count_service_connections(context_client, service) -> int: response = context_client.ListConnections(service.service_id) LOGGER.info( 'ServiceId[%s] => Connections[%d] = %s', grpc_message_to_json_string(service.service_id), len(response.connections), grpc_message_to_json_string(response), ) return len(response.connections) def get_virtual_link_ids(context_client) -> Set[str]: response = context_client.ListLinks(Empty()) virtual_link_ids = { link.link_id.link_uuid.uuid for link in response.links if link.link_type == LinkTypeEnum.LINKTYPE_VIRTUAL } LOGGER.info('VirtualLinks[%d] = %s', len(virtual_link_ids), str(sorted(virtual_link_ids))) return virtual_link_ids def assert_global_state( ip_context_client, e2e_context_client, opt_context_client, expected_virtual_link_ids: Set[str], expected_e2e_services: int, expected_opt_connections: int, ) -> None: response = ip_context_client.ListServices(ADMIN_CONTEXT_ID) assert len(response.services) == 0 virtual_link_ids = get_virtual_link_ids(ip_context_client) assert virtual_link_ids == expected_virtual_link_ids e2e_services = list_active_optical_services(e2e_context_client) if expected_e2e_services == 0: assert len(e2e_services) == 0 else: assert len(e2e_services) == expected_e2e_services for service in e2e_services: assert count_service_connections(e2e_context_client, service) == 1 opt_services = list_active_optical_services(opt_context_client) if expected_opt_connections == 0: assert len(opt_services) == 0 else: assert len(opt_services) == 1 assert count_service_connections(opt_context_client, opt_services[0]) == expected_opt_connections def wait_for_state_or_raise( ip_context_client, e2e_context_client, opt_context_client, expected_virtual_link_ids: Set[str], expected_e2e_services: int, expected_opt_connections: int, max_retry: int = 12, wait_seconds: float = 1.0, ) -> None: last_error: Exception = Exception('state not reached') for _ in range(max_retry): try: assert_global_state( ip_context_client=ip_context_client, e2e_context_client=e2e_context_client, opt_context_client=opt_context_client, expected_virtual_link_ids=expected_virtual_link_ids, expected_e2e_services=expected_e2e_services, expected_opt_connections=expected_opt_connections, ) return except Exception as error: # pylint: disable=broad-except last_error = error time.sleep(wait_seconds) MSG = 'Timed out waiting expected state: virtual_links={:s} e2e_services={:d} opt_connections={:d}; error={:s}' raise Exception(MSG.format( str(sorted(expected_virtual_link_ids)), expected_e2e_services, expected_opt_connections, str(last_error) )) src/tests/ofc25/tests/test_functional_create_vlinks.py +4 −115 Original line number Diff line number Diff line Loading @@ -14,128 +14,17 @@ import logging import os import time from typing import List, Set from typing import Set from common.Constants import DEFAULT_CONTEXT_NAME from common.proto.context_pb2 import ContextId, Empty, LinkTypeEnum, ServiceStatusEnum, ServiceTypeEnum from common.tools.descriptor.Loader import DescriptorLoader, check_descriptor_load_results from common.tools.grpc.Tools import grpc_message_to_json_string from common.tools.object_factory.Context import json_context_id # pylint: disable=unused-import from .Fixtures import PROFILE_E2E, PROFILE_IP, PROFILE_OPT, tfs_clients from .Helper import DESCRIPTORS_DIR, VIRTUAL_LINK_DESCRIPTORS, wait_for_state_or_raise LOGGER = logging.getLogger(__name__) LOGGER.setLevel(logging.DEBUG) ADMIN_CONTEXT_ID = ContextId(**json_context_id(DEFAULT_CONTEXT_NAME)) VIRTUAL_LINK_DESCRIPTORS = [ ('virtual_link_01.json', 'IP1/PORT-xe1==IP2/PORT-xe1'), ('virtual_link_02.json', 'IP1/PORT-xe2==IP2/PORT-xe2'), ('virtual_link_03.json', 'IP1/PORT-xe3==IP2/PORT-xe3'), ] DESCRIPTORS_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), '..', 'descriptors') def _list_active_optical_services(context_client) -> List: response = context_client.ListServices(ADMIN_CONTEXT_ID) LOGGER.info('Services[%d] = %s', len(response.services), grpc_message_to_json_string(response)) active_optical_services = [] for service in response.services: assert service.service_type == ServiceTypeEnum.SERVICETYPE_OPTICAL_CONNECTIVITY assert service.service_status.service_status == ServiceStatusEnum.SERVICESTATUS_ACTIVE active_optical_services.append(service) return active_optical_services def _count_service_connections(context_client, service) -> int: response = context_client.ListConnections(service.service_id) LOGGER.info( 'ServiceId[%s] => Connections[%d] = %s', grpc_message_to_json_string(service.service_id), len(response.connections), grpc_message_to_json_string(response), ) return len(response.connections) def _get_virtual_link_ids(context_client) -> Set[str]: response = context_client.ListLinks(Empty()) virtual_link_ids = { link.link_id.link_uuid.uuid for link in response.links if link.link_type == LinkTypeEnum.LINKTYPE_VIRTUAL } LOGGER.info('VirtualLinks[%d] = %s', len(virtual_link_ids), str(sorted(virtual_link_ids))) return virtual_link_ids def _assert_global_state( ip_context_client, e2e_context_client, opt_context_client, expected_virtual_link_ids: Set[str], expected_e2e_services: int, expected_opt_connections: int, ) -> None: # IP should not have services in this OFC25 workflow. response = ip_context_client.ListServices(ADMIN_CONTEXT_ID) assert len(response.services) == 0 virtual_link_ids = _get_virtual_link_ids(ip_context_client) assert virtual_link_ids == expected_virtual_link_ids e2e_services = _list_active_optical_services(e2e_context_client) if expected_e2e_services == 0: assert len(e2e_services) == 0 else: assert len(e2e_services) == expected_e2e_services for service in e2e_services: assert _count_service_connections(e2e_context_client, service) == 1 opt_services = _list_active_optical_services(opt_context_client) if expected_opt_connections == 0: assert len(opt_services) == 0 else: assert len(opt_services) == 1 assert _count_service_connections(opt_context_client, opt_services[0]) == expected_opt_connections def _wait_for_state_or_raise( ip_context_client, e2e_context_client, opt_context_client, expected_virtual_link_ids: Set[str], expected_e2e_services: int, expected_opt_connections: int, max_retry: int = 12, wait_seconds: float = 1.0, ) -> None: last_error: Exception = Exception('state not reached') for _ in range(max_retry): try: _assert_global_state( ip_context_client=ip_context_client, e2e_context_client=e2e_context_client, opt_context_client=opt_context_client, expected_virtual_link_ids=expected_virtual_link_ids, expected_e2e_services=expected_e2e_services, expected_opt_connections=expected_opt_connections, ) return except Exception as error: # pylint: disable=broad-except last_error = error time.sleep(wait_seconds) msg = ( 'Timed out waiting expected state: virtual_links={:s} e2e_services={:d} opt_connections={:d}; error={:s}' ) raise Exception(msg.format(str(sorted(expected_virtual_link_ids)), expected_e2e_services, expected_opt_connections, str(last_error))) def test_create_virtual_link( tfs_clients, ) -> None: Loading @@ -145,7 +34,7 @@ def test_create_virtual_link( opt_context_client = tfs_clients[PROFILE_OPT].context # Initial state: no services in any TFS and no virtual links in IP. _wait_for_state_or_raise( wait_for_state_or_raise( ip_context_client=ip_context_client, e2e_context_client=e2e_context_client, opt_context_client=opt_context_client, Loading @@ -166,7 +55,7 @@ def test_create_virtual_link( check_descriptor_load_results(results, descriptor_loader) expected_virtual_link_ids.add(virtual_link_id) _wait_for_state_or_raise( wait_for_state_or_raise( ip_context_client=ip_context_client, e2e_context_client=e2e_context_client, opt_context_client=opt_context_client, Loading src/tests/ofc25/tests/test_functional_delete_vlinks.py +6 −5 Original line number Diff line number Diff line Loading @@ -18,22 +18,23 @@ from common.tools.descriptor.Loader import DescriptorLoader # pylint: disable=unused-import from .Fixtures import PROFILE_E2E, PROFILE_IP, PROFILE_OPT, tfs_clients from .Helper import ( DESCRIPTORS_DIR, VIRTUAL_LINK_DESCRIPTORS, _wait_for_state_or_raise, wait_for_state_or_raise, ) def test_service_removal_bidir( def test_delete_virtual_links( tfs_clients, ): ) -> None: ip_context_client = tfs_clients[PROFILE_IP].context ip_device_client = tfs_clients[PROFILE_IP].device e2e_context_client = tfs_clients[PROFILE_E2E].context opt_context_client = tfs_clients[PROFILE_OPT].context expected_virtual_link_ids = {link_id for _, link_id in VIRTUAL_LINK_DESCRIPTORS} _wait_for_state_or_raise( wait_for_state_or_raise( ip_context_client=ip_context_client, e2e_context_client=e2e_context_client, opt_context_client=opt_context_client, Loading @@ -54,7 +55,7 @@ def test_service_removal_bidir( descriptor_loader.unload() expected_virtual_link_ids.remove(virtual_link_id) _wait_for_state_or_raise( wait_for_state_or_raise( ip_context_client=ip_context_client, e2e_context_client=e2e_context_client, opt_context_client=opt_context_client, Loading Loading
src/tests/ofc25/tests/Helper.py 0 → 100644 +133 −0 Original line number Diff line number Diff line # Copyright 2022-2026 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 logging import os import time from typing import List, Set from common.Constants import DEFAULT_CONTEXT_NAME from common.proto.context_pb2 import ContextId, Empty, LinkTypeEnum, ServiceStatusEnum, ServiceTypeEnum from common.tools.grpc.Tools import grpc_message_to_json_string from common.tools.object_factory.Context import json_context_id LOGGER = logging.getLogger(__name__) LOGGER.setLevel(logging.DEBUG) ADMIN_CONTEXT_ID = ContextId(**json_context_id(DEFAULT_CONTEXT_NAME)) VIRTUAL_LINK_DESCRIPTORS = [ ('virtual_link_01.json', 'IP1/PORT-xe1==IP2/PORT-xe1'), ('virtual_link_02.json', 'IP1/PORT-xe2==IP2/PORT-xe2'), ('virtual_link_03.json', 'IP1/PORT-xe3==IP2/PORT-xe3'), ] DESCRIPTORS_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), '..', 'descriptors') def list_active_optical_services(context_client) -> List: response = context_client.ListServices(ADMIN_CONTEXT_ID) LOGGER.info('Services[%d] = %s', len(response.services), grpc_message_to_json_string(response)) active_optical_services = [] for service in response.services: assert service.service_type == ServiceTypeEnum.SERVICETYPE_OPTICAL_CONNECTIVITY assert service.service_status.service_status == ServiceStatusEnum.SERVICESTATUS_ACTIVE active_optical_services.append(service) return active_optical_services def count_service_connections(context_client, service) -> int: response = context_client.ListConnections(service.service_id) LOGGER.info( 'ServiceId[%s] => Connections[%d] = %s', grpc_message_to_json_string(service.service_id), len(response.connections), grpc_message_to_json_string(response), ) return len(response.connections) def get_virtual_link_ids(context_client) -> Set[str]: response = context_client.ListLinks(Empty()) virtual_link_ids = { link.link_id.link_uuid.uuid for link in response.links if link.link_type == LinkTypeEnum.LINKTYPE_VIRTUAL } LOGGER.info('VirtualLinks[%d] = %s', len(virtual_link_ids), str(sorted(virtual_link_ids))) return virtual_link_ids def assert_global_state( ip_context_client, e2e_context_client, opt_context_client, expected_virtual_link_ids: Set[str], expected_e2e_services: int, expected_opt_connections: int, ) -> None: response = ip_context_client.ListServices(ADMIN_CONTEXT_ID) assert len(response.services) == 0 virtual_link_ids = get_virtual_link_ids(ip_context_client) assert virtual_link_ids == expected_virtual_link_ids e2e_services = list_active_optical_services(e2e_context_client) if expected_e2e_services == 0: assert len(e2e_services) == 0 else: assert len(e2e_services) == expected_e2e_services for service in e2e_services: assert count_service_connections(e2e_context_client, service) == 1 opt_services = list_active_optical_services(opt_context_client) if expected_opt_connections == 0: assert len(opt_services) == 0 else: assert len(opt_services) == 1 assert count_service_connections(opt_context_client, opt_services[0]) == expected_opt_connections def wait_for_state_or_raise( ip_context_client, e2e_context_client, opt_context_client, expected_virtual_link_ids: Set[str], expected_e2e_services: int, expected_opt_connections: int, max_retry: int = 12, wait_seconds: float = 1.0, ) -> None: last_error: Exception = Exception('state not reached') for _ in range(max_retry): try: assert_global_state( ip_context_client=ip_context_client, e2e_context_client=e2e_context_client, opt_context_client=opt_context_client, expected_virtual_link_ids=expected_virtual_link_ids, expected_e2e_services=expected_e2e_services, expected_opt_connections=expected_opt_connections, ) return except Exception as error: # pylint: disable=broad-except last_error = error time.sleep(wait_seconds) MSG = 'Timed out waiting expected state: virtual_links={:s} e2e_services={:d} opt_connections={:d}; error={:s}' raise Exception(MSG.format( str(sorted(expected_virtual_link_ids)), expected_e2e_services, expected_opt_connections, str(last_error) ))
src/tests/ofc25/tests/test_functional_create_vlinks.py +4 −115 Original line number Diff line number Diff line Loading @@ -14,128 +14,17 @@ import logging import os import time from typing import List, Set from typing import Set from common.Constants import DEFAULT_CONTEXT_NAME from common.proto.context_pb2 import ContextId, Empty, LinkTypeEnum, ServiceStatusEnum, ServiceTypeEnum from common.tools.descriptor.Loader import DescriptorLoader, check_descriptor_load_results from common.tools.grpc.Tools import grpc_message_to_json_string from common.tools.object_factory.Context import json_context_id # pylint: disable=unused-import from .Fixtures import PROFILE_E2E, PROFILE_IP, PROFILE_OPT, tfs_clients from .Helper import DESCRIPTORS_DIR, VIRTUAL_LINK_DESCRIPTORS, wait_for_state_or_raise LOGGER = logging.getLogger(__name__) LOGGER.setLevel(logging.DEBUG) ADMIN_CONTEXT_ID = ContextId(**json_context_id(DEFAULT_CONTEXT_NAME)) VIRTUAL_LINK_DESCRIPTORS = [ ('virtual_link_01.json', 'IP1/PORT-xe1==IP2/PORT-xe1'), ('virtual_link_02.json', 'IP1/PORT-xe2==IP2/PORT-xe2'), ('virtual_link_03.json', 'IP1/PORT-xe3==IP2/PORT-xe3'), ] DESCRIPTORS_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), '..', 'descriptors') def _list_active_optical_services(context_client) -> List: response = context_client.ListServices(ADMIN_CONTEXT_ID) LOGGER.info('Services[%d] = %s', len(response.services), grpc_message_to_json_string(response)) active_optical_services = [] for service in response.services: assert service.service_type == ServiceTypeEnum.SERVICETYPE_OPTICAL_CONNECTIVITY assert service.service_status.service_status == ServiceStatusEnum.SERVICESTATUS_ACTIVE active_optical_services.append(service) return active_optical_services def _count_service_connections(context_client, service) -> int: response = context_client.ListConnections(service.service_id) LOGGER.info( 'ServiceId[%s] => Connections[%d] = %s', grpc_message_to_json_string(service.service_id), len(response.connections), grpc_message_to_json_string(response), ) return len(response.connections) def _get_virtual_link_ids(context_client) -> Set[str]: response = context_client.ListLinks(Empty()) virtual_link_ids = { link.link_id.link_uuid.uuid for link in response.links if link.link_type == LinkTypeEnum.LINKTYPE_VIRTUAL } LOGGER.info('VirtualLinks[%d] = %s', len(virtual_link_ids), str(sorted(virtual_link_ids))) return virtual_link_ids def _assert_global_state( ip_context_client, e2e_context_client, opt_context_client, expected_virtual_link_ids: Set[str], expected_e2e_services: int, expected_opt_connections: int, ) -> None: # IP should not have services in this OFC25 workflow. response = ip_context_client.ListServices(ADMIN_CONTEXT_ID) assert len(response.services) == 0 virtual_link_ids = _get_virtual_link_ids(ip_context_client) assert virtual_link_ids == expected_virtual_link_ids e2e_services = _list_active_optical_services(e2e_context_client) if expected_e2e_services == 0: assert len(e2e_services) == 0 else: assert len(e2e_services) == expected_e2e_services for service in e2e_services: assert _count_service_connections(e2e_context_client, service) == 1 opt_services = _list_active_optical_services(opt_context_client) if expected_opt_connections == 0: assert len(opt_services) == 0 else: assert len(opt_services) == 1 assert _count_service_connections(opt_context_client, opt_services[0]) == expected_opt_connections def _wait_for_state_or_raise( ip_context_client, e2e_context_client, opt_context_client, expected_virtual_link_ids: Set[str], expected_e2e_services: int, expected_opt_connections: int, max_retry: int = 12, wait_seconds: float = 1.0, ) -> None: last_error: Exception = Exception('state not reached') for _ in range(max_retry): try: _assert_global_state( ip_context_client=ip_context_client, e2e_context_client=e2e_context_client, opt_context_client=opt_context_client, expected_virtual_link_ids=expected_virtual_link_ids, expected_e2e_services=expected_e2e_services, expected_opt_connections=expected_opt_connections, ) return except Exception as error: # pylint: disable=broad-except last_error = error time.sleep(wait_seconds) msg = ( 'Timed out waiting expected state: virtual_links={:s} e2e_services={:d} opt_connections={:d}; error={:s}' ) raise Exception(msg.format(str(sorted(expected_virtual_link_ids)), expected_e2e_services, expected_opt_connections, str(last_error))) def test_create_virtual_link( tfs_clients, ) -> None: Loading @@ -145,7 +34,7 @@ def test_create_virtual_link( opt_context_client = tfs_clients[PROFILE_OPT].context # Initial state: no services in any TFS and no virtual links in IP. _wait_for_state_or_raise( wait_for_state_or_raise( ip_context_client=ip_context_client, e2e_context_client=e2e_context_client, opt_context_client=opt_context_client, Loading @@ -166,7 +55,7 @@ def test_create_virtual_link( check_descriptor_load_results(results, descriptor_loader) expected_virtual_link_ids.add(virtual_link_id) _wait_for_state_or_raise( wait_for_state_or_raise( ip_context_client=ip_context_client, e2e_context_client=e2e_context_client, opt_context_client=opt_context_client, Loading
src/tests/ofc25/tests/test_functional_delete_vlinks.py +6 −5 Original line number Diff line number Diff line Loading @@ -18,22 +18,23 @@ from common.tools.descriptor.Loader import DescriptorLoader # pylint: disable=unused-import from .Fixtures import PROFILE_E2E, PROFILE_IP, PROFILE_OPT, tfs_clients from .Helper import ( DESCRIPTORS_DIR, VIRTUAL_LINK_DESCRIPTORS, _wait_for_state_or_raise, wait_for_state_or_raise, ) def test_service_removal_bidir( def test_delete_virtual_links( tfs_clients, ): ) -> None: ip_context_client = tfs_clients[PROFILE_IP].context ip_device_client = tfs_clients[PROFILE_IP].device e2e_context_client = tfs_clients[PROFILE_E2E].context opt_context_client = tfs_clients[PROFILE_OPT].context expected_virtual_link_ids = {link_id for _, link_id in VIRTUAL_LINK_DESCRIPTORS} _wait_for_state_or_raise( wait_for_state_or_raise( ip_context_client=ip_context_client, e2e_context_client=e2e_context_client, opt_context_client=opt_context_client, Loading @@ -54,7 +55,7 @@ def test_service_removal_bidir( descriptor_loader.unload() expected_virtual_link_ids.remove(virtual_link_id) _wait_for_state_or_raise( wait_for_state_or_raise( ip_context_client=ip_context_client, e2e_context_client=e2e_context_client, opt_context_client=opt_context_client, Loading
