Commit 33367cbb authored by Lluis Gifre Renom's avatar Lluis Gifre Renom
Browse files

OFC25 test: 



- Fixed check of created services/connections

Co-authored-by: default avatarCopilot <copilot@github.com>
parent 28437be0

src/tests/ofc25/tests/Helper.py

+59 −8
Original line number Diff line number Diff line
@@ -15,7 +15,7 @@ 
import logging

import os

import time

from typing import List, Optional, Set, Tuple

from typing import Dict, List, Optional, Set, Tuple



from common.Constants import DEFAULT_CONTEXT_NAME

from common.proto.context_pb2 import ContextId, Device, Empty, Link, LinkTypeEnum, ServiceStatusEnum, ServiceTypeEnum
@@ -261,6 +261,25 @@ def assert_expected_set(actual_items: Set[str], expected_items: Optional[Set[str 
    )





def get_service_identifiers(service) -> Set[str]:

    identifiers = {service.service_id.service_uuid.uuid}

    if len(service.name) > 0:

        identifiers.add(service.name)

    return identifiers





def build_expected_optical_connections(expected_virtual_link_names: Set[str]) -> Dict[str, int]:

    expected_connections = dict()

    first_optical_service_name = VIRTUAL_LINK_DESCRIPTORS[0][1]



    for _, virtual_link_name in VIRTUAL_LINK_DESCRIPTORS:

        if virtual_link_name not in expected_virtual_link_names:

            continue

        expected_connections[virtual_link_name] = 2 if virtual_link_name == first_optical_service_name else 1



    return expected_connections





def assert_global_state(

    ip_context_client,

    e2e_context_client,
@@ -268,7 +287,8 @@ def assert_global_state( 
    expected_virtual_link_uuids: Optional[Set[str]],

    expected_virtual_link_names: Optional[Set[str]],

    expected_e2e_services: int,

    expected_opt_connections: int,

    expected_opt_services: int,

    expected_opt_connections: Optional[Dict[str, int]],

) -> None:

    response = ip_context_client.ListServices(ADMIN_CONTEXT_ID)

    assert len(response.services) == 0
@@ -287,12 +307,39 @@ def assert_global_state( 
            assert count_service_connections(e2e_context_client, service) == 1



    opt_services = list_active_optical_services(opt_context_client)

    if expected_opt_connections == 0:

    if expected_opt_services == 0:

        response = opt_context_client.ListServices(ADMIN_CONTEXT_ID)

        assert len(response.services) == 0

    else:

        assert len(opt_services) == 1

        assert count_service_connections(opt_context_client, opt_services[0]) == expected_opt_connections

        assert len(opt_services) == expected_opt_services



        if expected_opt_connections is not None:

            unmatched_expected = dict(expected_opt_connections)

            for service in opt_services:

                service_identifiers = get_service_identifiers(service)

                matching_identifiers = [

                    identifier for identifier in service_identifiers if identifier in unmatched_expected

                ]

                assert len(matching_identifiers) == 1, (

                    'Unable to match optical service identifiers={:s} against expected={:s}'.format(

                        str(sorted(service_identifiers)), str(sorted(unmatched_expected.keys()))

                    )

                )



                service_identifier = matching_identifiers[0]

                actual_connections = count_service_connections(opt_context_client, service)

                expected_connections = unmatched_expected.pop(service_identifier)

                assert actual_connections == expected_connections, (

                    'Optical service {:s} connections mismatch: expected={:d} actual={:d}'.format(

                        service_identifier, expected_connections, actual_connections

                    )

                )



            assert len(unmatched_expected) == 0, (

                'Missing optical services for expected connection checks: {:s}'.format(

                    str(sorted(unmatched_expected.keys()))

                )

            )





def wait_for_state_or_raise(
@@ -302,7 +349,8 @@ def wait_for_state_or_raise( 
    expected_virtual_link_uuids: Optional[Set[str]],

    expected_virtual_link_names: Optional[Set[str]],

    expected_e2e_services: int,

    expected_opt_connections: int,

    expected_opt_services: int,

    expected_opt_connections: Optional[Dict[str, int]],

    max_retry: int = 5,

    wait_seconds: float = 15.0,

) -> None:
@@ -311,12 +359,13 @@ def wait_for_state_or_raise( 
        try:

            LOGGER.info(

                'Checking expected state attempt %d/%d: virtual_link_uuids=%s virtual_link_names=%s '

                'e2e_services=%d opt_connections=%d',

                'e2e_services=%d opt_services=%d opt_connections=%s',

                attempt,

                max_retry,

                '<ignored>' if expected_virtual_link_uuids is None else str(sorted(expected_virtual_link_uuids)),

                '<ignored>' if expected_virtual_link_names is None else str(sorted(expected_virtual_link_names)),

                expected_e2e_services,

                expected_opt_services,

                expected_opt_connections,

            )

            assert_global_state(
@@ -326,6 +375,7 @@ def wait_for_state_or_raise( 
                expected_virtual_link_uuids=expected_virtual_link_uuids,

                expected_virtual_link_names=expected_virtual_link_names,

                expected_e2e_services=expected_e2e_services,

                expected_opt_services=expected_opt_services,

                expected_opt_connections=expected_opt_connections,

            )

            return
@@ -340,10 +390,11 @@ def wait_for_state_or_raise( 


    LOGGER.error(

        'Timed out waiting expected state: virtual_link_uuids=%s virtual_link_names=%s '

        'e2e_services=%d opt_connections=%d',

        'e2e_services=%d opt_services=%d opt_connections=%s',

        '<ignored>' if expected_virtual_link_uuids is None else str(sorted(expected_virtual_link_uuids)),

        '<ignored>' if expected_virtual_link_names is None else str(sorted(expected_virtual_link_names)),

        expected_e2e_services,

        expected_opt_services,

        expected_opt_connections,

    )

    raise last_error

src/tests/ofc25/tests/test_functional_create_vlinks.py

+11 −3
Original line number Diff line number Diff line
@@ -20,7 +20,12 @@ from common.tools.descriptor.Loader import DescriptorLoader, check_descriptor_lo 


# 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

from .Helper import (

    DESCRIPTORS_DIR,

    VIRTUAL_LINK_DESCRIPTORS,

    build_expected_optical_connections,

    wait_for_state_or_raise,

)



LOGGER = logging.getLogger(__name__)

LOGGER.setLevel(logging.DEBUG)
@@ -44,7 +49,8 @@ def test_create_virtual_link( 
        expected_virtual_link_uuids=None,

        expected_virtual_link_names=set(),

        expected_e2e_services=0,

        expected_opt_connections=0,

        expected_opt_services=0,

        expected_opt_connections={},

    )



    expected_virtual_link_names: Set[str] = set()
@@ -70,6 +76,7 @@ def test_create_virtual_link( 
            'expected_e2e_services=%d',

            virtual_link_name, str(sorted(expected_virtual_link_names)), index

        )

        expected_opt_connections = build_expected_optical_connections(expected_virtual_link_names)

        wait_for_state_or_raise(

            ip_context_client=ip_context_client,

            e2e_context_client=e2e_context_client,
@@ -77,5 +84,6 @@ def test_create_virtual_link( 
            expected_virtual_link_uuids=None,

            expected_virtual_link_names=expected_virtual_link_names,

            expected_e2e_services=index,

            expected_opt_connections=1 + index, # 1 optical band + N optical lightpaths within the band

            expected_opt_services=index,

            expected_opt_connections=expected_opt_connections,

        )

src/tests/ofc25/tests/test_functional_delete_vlinks.py

+6 −9
Original line number Diff line number Diff line
@@ -22,6 +22,7 @@ from .Fixtures import PROFILE_E2E, PROFILE_IP, PROFILE_OPT, tfs_clients 
from .Helper import (

    DESCRIPTORS_DIR,

    VIRTUAL_LINK_DESCRIPTORS,

    build_expected_optical_connections,

    wait_for_state_or_raise,

)
@@ -47,8 +48,9 @@ def test_delete_virtual_links( 
        opt_context_client=opt_context_client,

        expected_virtual_link_uuids=None,

        expected_virtual_link_names=expected_virtual_link_names,

        expected_e2e_services=3,

        expected_opt_connections=1 + 3, # 1 optical band + N optical lightpaths within the band

        expected_e2e_services=len(VIRTUAL_LINK_DESCRIPTORS),

        expected_opt_services=len(VIRTUAL_LINK_DESCRIPTORS),

        expected_opt_connections=build_expected_optical_connections(expected_virtual_link_names),

    )



    for remaining, (descriptor_name, virtual_link_name) in zip(
@@ -74,12 +76,6 @@ def test_delete_virtual_links( 
            'expected_e2e_services=%d',

            virtual_link_name, str(sorted(expected_virtual_link_names)), remaining

        )

        

        # 1 optical band + N optical lightpaths within the band

        # if only remains optical band, it is assumed it is auto-removed

        opt_remaining = 1 + remaining

        if opt_remaining == 1: opt_remaining = 0



        wait_for_state_or_raise(

            ip_context_client=ip_context_client,

            e2e_context_client=e2e_context_client,
@@ -87,5 +83,6 @@ def test_delete_virtual_links( 
            expected_virtual_link_uuids=None,

            expected_virtual_link_names=expected_virtual_link_names,

            expected_e2e_services=remaining,

            expected_opt_connections=opt_remaining,

            expected_opt_services=remaining,

            expected_opt_connections=build_expected_optical_connections(expected_virtual_link_names),

        )