ECOC'22 functional test:

- Fixed endpoint UUIDs of OLS device, otherwise cleanup crashes due to different endpoint UUIDs randomly generated.
- Fixed create/delete service tests

src/tests/ecoc22/tests/Objects_DC_CSGW_OLS.py

@@ -115,7 +115,8 @@ DEV_CS2GW1_ID, DEV_CS2GW1_EPS, DEV_CS2GW1 = compose_router('CS2-GW1', ['10/1', '
 DEV_CS2GW2_ID, DEV_CS2GW2_EPS, DEV_CS2GW2 = compose_router('CS2-GW2', ['10/1', '1/1'])
 
 # Transport Network
-tols_ep_uuids = [str(uuid.uuid4()).split('-')[-1] for _ in range(4)]
+#tols_ep_uuids = [str(uuid.uuid4()).split('-')[-1] for _ in range(4)]
+tols_ep_uuids = ['afd8ffbb5403', '04b84e213e83', '3169ae676ac6', '93506f786270']
 DEV_TOLS_ID, DEV_TOLS_EPS, DEV_TOLS = compose_ols('TN-OLS', tols_ep_uuids)
 
 

src/tests/ecoc22/tests/Objects_DC_CSGW_TN_OLS.py

@@ -119,7 +119,9 @@ DEV_TNR1_ID, DEV_TNR1_EPS, DEV_TNR1 = compose_router('TN-R1', ['1/1', '1/2', '2/
 DEV_TNR2_ID, DEV_TNR2_EPS, DEV_TNR2 = compose_router('TN-R2', ['1/1', '1/2', '2/1'])
 DEV_TNR3_ID, DEV_TNR3_EPS, DEV_TNR3 = compose_router('TN-R3', ['1/1', '1/2', '2/1'])
 DEV_TNR4_ID, DEV_TNR4_EPS, DEV_TNR4 = compose_router('TN-R4', ['1/1', '1/2', '2/1'])
-tols_ep_uuids = [str(uuid.uuid4()).split('-')[-1] for _ in range(4)]
+
+#tols_ep_uuids = [str(uuid.uuid4()).split('-')[-1] for _ in range(4)]
+tols_ep_uuids = ['afd8ffbb5403', '04b84e213e83', '3169ae676ac6', '93506f786270']
 DEV_TOLS_ID, DEV_TOLS_EPS, DEV_TOLS = compose_ols('TN-OLS', tols_ep_uuids)
 
 

src/tests/ecoc22/tests/test_functional_create_service.py

@@ -13,7 +13,7 @@
 # limitations under the License.
 
 import logging
-from common.proto.context_pb2 import ContextId, Empty
+from common.proto.context_pb2 import ContextId, Empty, ServiceTypeEnum
 from common.tools.grpc.Tools import grpc_message_to_json_string
 from compute.tests.mock_osm.MockOSM import MockOSM
 from context.client.ContextClient import ContextClient
@@ -72,10 +72,13 @@ def test_scenario_service_created(context_client : ContextClient):  # pylint: di
 
     response = context_client.ListServices(ContextId(**CONTEXT_ID))
     LOGGER.info('Services[{:d}] = {:s}'.format(len(response.services), grpc_message_to_json_string(response)))
-    assert len(response.services) == 1 # L2NM
+    assert len(response.services) == 3 # 1xL2NM + 2xTAPI
     for service in response.services:
         service_id = service.service_id
         response = context_client.ListConnections(service_id)
         LOGGER.info('  ServiceId[{:s}] => Connections[{:d}] = {:s}'.format(
             grpc_message_to_json_string(service_id), len(response.connections), grpc_message_to_json_string(response)))
-        assert len(response.connections) == 2 # 2 connections per service (primary + backup)
+        if service.service_type == ServiceTypeEnum.SERVICETYPE_L2NM:
+            assert len(response.connections) == 2 # 2 connections per service (primary + backup)
+        elif service.service_type == ServiceTypeEnum.SERVICETYPE_TAPI_CONNECTIVITY_SERVICE:
+            assert len(response.connections) == 1 # 1 connection per service

src/tests/ecoc22/tests/test_functional_delete_service.py

@@ -30,7 +30,7 @@ from .Fixtures import context_client, osm_wim
 #    CONTEXT_ID, CONTEXTS, DEVICES, LINKS, TOPOLOGIES, WIM_SERVICE_CONNECTION_POINTS, WIM_SERVICE_TYPE)
 #from .Objects_DC_CSGW_TN_OLS import (
 #    CONTEXT_ID, CONTEXTS, DEVICES, LINKS, TOPOLOGIES, WIM_SERVICE_CONNECTION_POINTS, WIM_SERVICE_TYPE)
-from .Objects_DC_CSGW_TN_OLS import (
+from .Objects_DC_CSGW_OLS import (
     CONTEXT_ID, CONTEXTS, DEVICES, LINKS, TOPOLOGIES, WIM_SERVICE_CONNECTION_POINTS, WIM_SERVICE_TYPE)
 
 LOGGER = logging.getLogger(__name__)
@@ -55,13 +55,16 @@ def test_scenario_is_correct(context_client : ContextClient):  # pylint: disable
 
     response = context_client.ListServices(ContextId(**CONTEXT_ID))
     LOGGER.info('Services[{:d}] = {:s}'.format(len(response.services), grpc_message_to_json_string(response)))
-    assert len(response.services) == 1 # L2NM
+    assert len(response.services) == 3 # 1xL2NM + 2xTAPI
     for service in response.services:
         service_id = service.service_id
         response = context_client.ListConnections(service_id)
         LOGGER.info('  ServiceId[{:s}] => Connections[{:d}] = {:s}'.format(
             grpc_message_to_json_string(service_id), len(response.connections), grpc_message_to_json_string(response)))
-        assert len(response.connections) == 2 # 2 connections per service
+        if service.service_type == ServiceTypeEnum.SERVICETYPE_L2NM:
+            assert len(response.connections) == 2 # 2 connections per service (primary + backup)
+        elif service.service_type == ServiceTypeEnum.SERVICETYPE_TAPI_CONNECTIVITY_SERVICE:
+            assert len(response.connections) == 1 # 1 connection per service
 
 
 def test_service_removal(context_client : ContextClient, osm_wim : MockOSM): # pylint: disable=redefined-outer-name
@@ -72,7 +75,6 @@ def test_service_removal(context_client : ContextClient, osm_wim : MockOSM): # p
     for service in response.services:
         if service.service_type != ServiceTypeEnum.SERVICETYPE_L2NM: continue
         service_uuid = service.service_id.service_uuid.uuid
-        if service_uuid.endswith(':optical'): continue
         service_uuids.add(service_uuid)
         osm_wim.conn_info[service_uuid] = {}