Commit dbbd110a authored by Lluis Gifre Renom's avatar Lluis Gifre Renom
Browse files

OFC25 test: 

- Fixed validation of imported links
parent 3af2558b

src/tests/ofc25/tests/Helper.py

+55 −3
Original line number Diff line number Diff line
@@ -18,7 +18,7 @@ import time 
from typing import List, Set, Tuple



from common.Constants import DEFAULT_CONTEXT_NAME

from common.proto.context_pb2 import ContextId, Device, Empty, LinkTypeEnum, ServiceStatusEnum, ServiceTypeEnum

from common.proto.context_pb2 import ContextId, Device, Empty, Link, LinkTypeEnum, ServiceStatusEnum, ServiceTypeEnum

from common.tools.grpc.Tools import grpc_message_list_to_json_string, grpc_message_to_json_string

from common.tools.object_factory.Context import json_context_id
@@ -62,6 +62,44 @@ def log_device_inventory(context_client, profile_name: str, log_prefix: str = 'D 
    return local_devices, imported_devices





def split_descriptor_links(

    links: List[Link], descriptor_link_ids: Set[str]

) -> Tuple[List[Link], List[Link], List[Link]]:

    descriptor_links = []

    management_links = []

    imported_links = []



    for link in links:

        link_uuid = link.link_id.link_uuid.uuid

        if link_uuid in descriptor_link_ids:

            descriptor_links.append(link)

        elif link.link_type == LinkTypeEnum.LINKTYPE_MANAGEMENT:

            management_links.append(link)

        else:

            imported_links.append(link)



    return descriptor_links, management_links, imported_links





def log_link_inventory(context_client, descriptor_loader, profile_name: str) -> Tuple[List[Link], List[Link], List[Link]]:

    response = context_client.ListLinks(Empty())

    descriptor_link_ids = {link['link_id']['link_uuid']['uuid'] for link in descriptor_loader.links}

    descriptor_links, management_links, imported_links = split_descriptor_links(response.links, descriptor_link_ids)



    LOGGER.info(

        '[%s] Descriptor validation link inventory: total=%d descriptor=%d management=%d imported=%d',

        profile_name,

        len(response.links),

        len(descriptor_links),

        len(management_links),

        len(imported_links),

    )

    LOGGER.info('[%s] Descriptor links: %s', profile_name, grpc_message_list_to_json_string(descriptor_links))

    LOGGER.info('[%s] Management links: %s', profile_name, grpc_message_list_to_json_string(management_links))

    LOGGER.info('[%s] Imported links: %s', profile_name, grpc_message_list_to_json_string(imported_links))

    return descriptor_links, management_links, imported_links





def validate_descriptor_state(context_client, descriptor_loader, profile_name: str) -> None:

    contexts = context_client.ListContexts(Empty())

    assert len(contexts.contexts) == descriptor_loader.num_contexts
@@ -81,8 +119,22 @@ def validate_descriptor_state(context_client, descriptor_loader, profile_name: s 
            len(imported_devices),

        )



    response = context_client.ListLinks(Empty())

    assert len(response.links) == descriptor_loader.num_links

    descriptor_links, management_links, imported_links = log_link_inventory(

        context_client, descriptor_loader, profile_name

    )

    assert len(descriptor_links) == descriptor_loader.num_links

    if management_links:

        LOGGER.info(

            '[%s] Found %d management links auto-added during device import',

            profile_name,

            len(management_links),

        )

    if imported_links:

        LOGGER.info(

            '[%s] Ignoring %d imported non-descriptor links during descriptor validation',

            profile_name,

            len(imported_links),

        )



    response = context_client.GetOpticalLinkList(Empty())

    assert len(response.optical_links) == descriptor_loader.num_optical_links