Loading src/tests/ofc25/tests/test_functional_create_vlinks.py +143 −51 Original line number Diff line number Diff line # Copyright 2022-2025 ETSI SDG TeraFlowSDN (TFS) (https://tfs.etsi.org/) # 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. Loading @@ -12,72 +12,164 @@ # See the License for the specific language governing permissions and # limitations under the License. import logging, os 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, ServiceStatusEnum, ServiceTypeEnum, ) from common.tools.descriptor.Loader import ( DescriptorLoader, check_descriptor_load_results, ) 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 from .Fixtures import ( PROFILE_IP, tfs_clients, ) # pylint: disable=unused-import from .Fixtures import PROFILE_E2E, PROFILE_IP, PROFILE_OPT, tfs_clients # pylint: disable=unused-import LOGGER = logging.getLogger(__name__) LOGGER.setLevel(logging.DEBUG) DESCRIPTOR_FILES = [ os.path.join(os.path.dirname(os.path.abspath(__file__)), '..', 'descriptors', fname) for fname in ['virtual_link_01.json', 'virtual_link_02.json', 'virtual_link_03.json'] ] ADMIN_CONTEXT_ID = ContextId(**json_context_id(DEFAULT_CONTEXT_NAME)) def test_create_virtual_link( tfs_clients, ) -> None: context_client = tfs_clients[PROFILE_IP].context device_client = tfs_clients[PROFILE_IP].device for descriptor_file in DESCRIPTOR_FILES: # Load descriptors and validate the base scenario pass descriptor_loader = DescriptorLoader( descriptors_file=descriptor_file, context_client=context_client, device_client=device_client ) results = descriptor_loader.process() check_descriptor_load_results(results, descriptor_loader) # Verify the scenario has 1 service and 0 slices response = context_client.GetContext(ADMIN_CONTEXT_ID) assert len(response.service_ids) == 1 assert len(response.slice_ids) == 0 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') # Check there are no slices response = context_client.ListSlices(ADMIN_CONTEXT_ID) LOGGER.warning('Slices[{:d}] = {:s}'.format(len(response.slices), grpc_message_to_json_string(response))) assert len(response.slices) == 0 # Check there is 1 service def _list_active_optical_services(context_client) -> List: response = context_client.ListServices(ADMIN_CONTEXT_ID) LOGGER.warning('Services[{:d}] = {:s}'.format(len(response.services), grpc_message_to_json_string(response))) assert len(response.services) == 1 LOGGER.info('Services[%d] = %s', len(response.services), grpc_message_to_json_string(response)) active_optical_services = [] for service in response.services: service_id = service.service_id 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 response = context_client.ListConnections(service_id) LOGGER.warning(' ServiceId[{:s}] => Connections[{:d}] = {:s}'.format( grpc_message_to_json_string(service_id), len(response.connections), grpc_message_to_json_string(response))) virtual_link_ids = _get_virtual_link_ids(ip_context_client) assert virtual_link_ids == expected_virtual_link_ids if service.service_type == ServiceTypeEnum.SERVICETYPE_OPTICAL_CONNECTIVITY: assert len(response.connections) == 2 e2e_services = _list_active_optical_services(e2e_context_client) if expected_e2e_services == 0: assert len(e2e_services) == 0 else: str_service = grpc_message_to_json_string(service) raise Exception('Unexpected ServiceType: {:s}'.format(str_service)) 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: 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 # Initial state: no services in any TFS and no virtual links in IP. _wait_for_state_or_raise( ip_context_client=ip_context_client, e2e_context_client=e2e_context_client, opt_context_client=opt_context_client, expected_virtual_link_ids=set(), expected_e2e_services=0, expected_opt_connections=0, ) expected_virtual_link_ids: Set[str] = set() for index, (descriptor_name, virtual_link_id) in enumerate(VIRTUAL_LINK_DESCRIPTORS, start=1): descriptor_file = os.path.join(DESCRIPTORS_DIR, descriptor_name) descriptor_loader = DescriptorLoader( descriptors_file=descriptor_file, context_client=ip_context_client, device_client=ip_device_client, ) results = descriptor_loader.process() check_descriptor_load_results(results, descriptor_loader) expected_virtual_link_ids.add(virtual_link_id) _wait_for_state_or_raise( 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=index, expected_opt_connections=1, ) src/tests/ofc25/tests/test_functional_delete_vlinks.py +41 −57 Original line number Diff line number Diff line # Copyright 2022-2025 ETSI SDG TeraFlowSDN (TFS) (https://tfs.etsi.org/) # 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. Loading @@ -12,69 +12,53 @@ # See the License for the specific language governing permissions and # limitations under the License. import logging from typing import Set, Tuple import os from common.Constants import DEFAULT_CONTEXT_NAME from common.proto.context_pb2 import ( ContextId, ServiceId, ServiceStatusEnum, ServiceTypeEnum, ) from common.tools.grpc.Tools import grpc_message_to_json_string from common.tools.object_factory.Context import json_context_id from .Fixtures import PROFILE_IP, tfs_clients # pylint: disable=unused-import from common.tools.descriptor.Loader import DescriptorLoader LOGGER = logging.getLogger(__name__) LOGGER.setLevel(logging.DEBUG) from .Fixtures import PROFILE_E2E, PROFILE_IP, PROFILE_OPT, tfs_clients # pylint: disable=unused-import from .test_functional_create_vlinks import ( DESCRIPTORS_DIR, VIRTUAL_LINK_DESCRIPTORS, _wait_for_state_or_raise, ) ADMIN_CONTEXT_ID = ContextId(**json_context_id(DEFAULT_CONTEXT_NAME)) def test_service_removal_bidir( tfs_clients, ): context_client = tfs_clients[PROFILE_IP].context # Verify the scenario has 1 service and 0 slices response = context_client.GetContext(ADMIN_CONTEXT_ID) assert len(response.service_ids) == 1 assert len(response.slice_ids) == 0 # Check there are no slices response = context_client.ListSlices(ADMIN_CONTEXT_ID) LOGGER.warning('Slices[{:d}] = {:s}'.format(len(response.slices), grpc_message_to_json_string(response))) assert len(response.slices) == 0 # Check there is 1 service response = context_client.ListServices(ADMIN_CONTEXT_ID) LOGGER.warning('Services[{:d}] = {:s}'.format(len(response.services), grpc_message_to_json_string(response))) assert len(response.services) == 1 context_service_uuids : Set[Tuple[str, str]] = set() for service in response.services: service_id = service.service_id assert service.service_status.service_status == ServiceStatusEnum.SERVICESTATUS_ACTIVE response = context_client.ListConnections(service_id) LOGGER.warning(' ServiceId[{:s}] => Connections[{:d}] = {:s}'.format( grpc_message_to_json_string(service_id), len(response.connections), grpc_message_to_json_string(response))) 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 if service.service_type == ServiceTypeEnum.SERVICETYPE_OPTICAL_CONNECTIVITY: assert len(response.connections) == 2 context_uuid = service_id.context_id.context_uuid.uuid service_uuid = service_id.service_uuid.uuid context_service_uuids.add((context_uuid, service_uuid)) else: str_service = grpc_message_to_json_string(service) raise Exception('Unexpected ServiceType: {:s}'.format(str_service)) # Identify service to delete assert len(context_service_uuids) == 1 context_uuid, service_uuid = set(context_service_uuids).pop() # Delete Service # service_client.DeleteService(ServiceId(**json_service_id(service_uuid, json_context_id(context_uuid)))) expected_virtual_link_ids = {link_id for _, link_id in VIRTUAL_LINK_DESCRIPTORS} _wait_for_state_or_raise( 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=3, expected_opt_connections=1, ) import delete_service for remaining, (descriptor_name, virtual_link_id) in zip( [2, 1, 0], reversed(VIRTUAL_LINK_DESCRIPTORS) ): descriptor_file = os.path.join(DESCRIPTORS_DIR, descriptor_name) descriptor_loader = DescriptorLoader( descriptors_file=descriptor_file, context_client=ip_context_client, device_client=ip_device_client, ) descriptor_loader.unload() # Verify the scenario has no services/slices response = context_client.GetContext(ADMIN_CONTEXT_ID) assert len(response.service_ids) == 0 assert len(response.slice_ids) == 0 expected_virtual_link_ids.remove(virtual_link_id) _wait_for_state_or_raise( 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=remaining, expected_opt_connections=(1 if remaining > 0 else 0), ) Loading
src/tests/ofc25/tests/test_functional_create_vlinks.py +143 −51 Original line number Diff line number Diff line # Copyright 2022-2025 ETSI SDG TeraFlowSDN (TFS) (https://tfs.etsi.org/) # 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. Loading @@ -12,72 +12,164 @@ # See the License for the specific language governing permissions and # limitations under the License. import logging, os 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, ServiceStatusEnum, ServiceTypeEnum, ) from common.tools.descriptor.Loader import ( DescriptorLoader, check_descriptor_load_results, ) 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 from .Fixtures import ( PROFILE_IP, tfs_clients, ) # pylint: disable=unused-import from .Fixtures import PROFILE_E2E, PROFILE_IP, PROFILE_OPT, tfs_clients # pylint: disable=unused-import LOGGER = logging.getLogger(__name__) LOGGER.setLevel(logging.DEBUG) DESCRIPTOR_FILES = [ os.path.join(os.path.dirname(os.path.abspath(__file__)), '..', 'descriptors', fname) for fname in ['virtual_link_01.json', 'virtual_link_02.json', 'virtual_link_03.json'] ] ADMIN_CONTEXT_ID = ContextId(**json_context_id(DEFAULT_CONTEXT_NAME)) def test_create_virtual_link( tfs_clients, ) -> None: context_client = tfs_clients[PROFILE_IP].context device_client = tfs_clients[PROFILE_IP].device for descriptor_file in DESCRIPTOR_FILES: # Load descriptors and validate the base scenario pass descriptor_loader = DescriptorLoader( descriptors_file=descriptor_file, context_client=context_client, device_client=device_client ) results = descriptor_loader.process() check_descriptor_load_results(results, descriptor_loader) # Verify the scenario has 1 service and 0 slices response = context_client.GetContext(ADMIN_CONTEXT_ID) assert len(response.service_ids) == 1 assert len(response.slice_ids) == 0 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') # Check there are no slices response = context_client.ListSlices(ADMIN_CONTEXT_ID) LOGGER.warning('Slices[{:d}] = {:s}'.format(len(response.slices), grpc_message_to_json_string(response))) assert len(response.slices) == 0 # Check there is 1 service def _list_active_optical_services(context_client) -> List: response = context_client.ListServices(ADMIN_CONTEXT_ID) LOGGER.warning('Services[{:d}] = {:s}'.format(len(response.services), grpc_message_to_json_string(response))) assert len(response.services) == 1 LOGGER.info('Services[%d] = %s', len(response.services), grpc_message_to_json_string(response)) active_optical_services = [] for service in response.services: service_id = service.service_id 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 response = context_client.ListConnections(service_id) LOGGER.warning(' ServiceId[{:s}] => Connections[{:d}] = {:s}'.format( grpc_message_to_json_string(service_id), len(response.connections), grpc_message_to_json_string(response))) virtual_link_ids = _get_virtual_link_ids(ip_context_client) assert virtual_link_ids == expected_virtual_link_ids if service.service_type == ServiceTypeEnum.SERVICETYPE_OPTICAL_CONNECTIVITY: assert len(response.connections) == 2 e2e_services = _list_active_optical_services(e2e_context_client) if expected_e2e_services == 0: assert len(e2e_services) == 0 else: str_service = grpc_message_to_json_string(service) raise Exception('Unexpected ServiceType: {:s}'.format(str_service)) 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: 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 # Initial state: no services in any TFS and no virtual links in IP. _wait_for_state_or_raise( ip_context_client=ip_context_client, e2e_context_client=e2e_context_client, opt_context_client=opt_context_client, expected_virtual_link_ids=set(), expected_e2e_services=0, expected_opt_connections=0, ) expected_virtual_link_ids: Set[str] = set() for index, (descriptor_name, virtual_link_id) in enumerate(VIRTUAL_LINK_DESCRIPTORS, start=1): descriptor_file = os.path.join(DESCRIPTORS_DIR, descriptor_name) descriptor_loader = DescriptorLoader( descriptors_file=descriptor_file, context_client=ip_context_client, device_client=ip_device_client, ) results = descriptor_loader.process() check_descriptor_load_results(results, descriptor_loader) expected_virtual_link_ids.add(virtual_link_id) _wait_for_state_or_raise( 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=index, expected_opt_connections=1, )
src/tests/ofc25/tests/test_functional_delete_vlinks.py +41 −57 Original line number Diff line number Diff line # Copyright 2022-2025 ETSI SDG TeraFlowSDN (TFS) (https://tfs.etsi.org/) # 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. Loading @@ -12,69 +12,53 @@ # See the License for the specific language governing permissions and # limitations under the License. import logging from typing import Set, Tuple import os from common.Constants import DEFAULT_CONTEXT_NAME from common.proto.context_pb2 import ( ContextId, ServiceId, ServiceStatusEnum, ServiceTypeEnum, ) from common.tools.grpc.Tools import grpc_message_to_json_string from common.tools.object_factory.Context import json_context_id from .Fixtures import PROFILE_IP, tfs_clients # pylint: disable=unused-import from common.tools.descriptor.Loader import DescriptorLoader LOGGER = logging.getLogger(__name__) LOGGER.setLevel(logging.DEBUG) from .Fixtures import PROFILE_E2E, PROFILE_IP, PROFILE_OPT, tfs_clients # pylint: disable=unused-import from .test_functional_create_vlinks import ( DESCRIPTORS_DIR, VIRTUAL_LINK_DESCRIPTORS, _wait_for_state_or_raise, ) ADMIN_CONTEXT_ID = ContextId(**json_context_id(DEFAULT_CONTEXT_NAME)) def test_service_removal_bidir( tfs_clients, ): context_client = tfs_clients[PROFILE_IP].context # Verify the scenario has 1 service and 0 slices response = context_client.GetContext(ADMIN_CONTEXT_ID) assert len(response.service_ids) == 1 assert len(response.slice_ids) == 0 # Check there are no slices response = context_client.ListSlices(ADMIN_CONTEXT_ID) LOGGER.warning('Slices[{:d}] = {:s}'.format(len(response.slices), grpc_message_to_json_string(response))) assert len(response.slices) == 0 # Check there is 1 service response = context_client.ListServices(ADMIN_CONTEXT_ID) LOGGER.warning('Services[{:d}] = {:s}'.format(len(response.services), grpc_message_to_json_string(response))) assert len(response.services) == 1 context_service_uuids : Set[Tuple[str, str]] = set() for service in response.services: service_id = service.service_id assert service.service_status.service_status == ServiceStatusEnum.SERVICESTATUS_ACTIVE response = context_client.ListConnections(service_id) LOGGER.warning(' ServiceId[{:s}] => Connections[{:d}] = {:s}'.format( grpc_message_to_json_string(service_id), len(response.connections), grpc_message_to_json_string(response))) 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 if service.service_type == ServiceTypeEnum.SERVICETYPE_OPTICAL_CONNECTIVITY: assert len(response.connections) == 2 context_uuid = service_id.context_id.context_uuid.uuid service_uuid = service_id.service_uuid.uuid context_service_uuids.add((context_uuid, service_uuid)) else: str_service = grpc_message_to_json_string(service) raise Exception('Unexpected ServiceType: {:s}'.format(str_service)) # Identify service to delete assert len(context_service_uuids) == 1 context_uuid, service_uuid = set(context_service_uuids).pop() # Delete Service # service_client.DeleteService(ServiceId(**json_service_id(service_uuid, json_context_id(context_uuid)))) expected_virtual_link_ids = {link_id for _, link_id in VIRTUAL_LINK_DESCRIPTORS} _wait_for_state_or_raise( 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=3, expected_opt_connections=1, ) import delete_service for remaining, (descriptor_name, virtual_link_id) in zip( [2, 1, 0], reversed(VIRTUAL_LINK_DESCRIPTORS) ): descriptor_file = os.path.join(DESCRIPTORS_DIR, descriptor_name) descriptor_loader = DescriptorLoader( descriptors_file=descriptor_file, context_client=ip_context_client, device_client=ip_device_client, ) descriptor_loader.unload() # Verify the scenario has no services/slices response = context_client.GetContext(ADMIN_CONTEXT_ID) assert len(response.service_ids) == 0 assert len(response.slice_ids) == 0 expected_virtual_link_ids.remove(virtual_link_id) _wait_for_state_or_raise( 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=remaining, expected_opt_connections=(1 if remaining > 0 else 0), )
