Commit 8b5a94e4 authored by Lluis Gifre Renom's avatar Lluis Gifre Renom
Browse files

OFC25 test: 

- Fixed classification of links
parent dbbd110a

src/tests/ofc25/tests/Helper.py

+33 −6
Original line number Diff line number Diff line
@@ -63,17 +63,20 @@ def log_device_inventory(context_client, profile_name: str, log_prefix: str = 'D 




def split_descriptor_links(

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

    links: List[Link], descriptor_link_aliases: 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:

        runtime_aliases = {link.link_id.link_uuid.uuid}

        if link.name:

            runtime_aliases.add(link.name)



        if runtime_aliases.intersection(descriptor_link_aliases):

            descriptor_links.append(link)

        elif link.link_type == LinkTypeEnum.LINKTYPE_MANAGEMENT:

        elif _is_management_link(link):

            management_links.append(link)

        else:

            imported_links.append(link)
@@ -83,8 +86,18 @@ def split_descriptor_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)

    descriptor_link_aliases = set()

    for link in descriptor_loader.links:

        link_uuid = link.get('link_id', {}).get('link_uuid', {}).get('uuid')

        if link_uuid:

            descriptor_link_aliases.add(link_uuid)

        link_name = link.get('name')

        if link_name:

            descriptor_link_aliases.add(link_name)



    descriptor_links, management_links, imported_links = split_descriptor_links(

        response.links, descriptor_link_aliases

    )



    LOGGER.info(

        '[%s] Descriptor validation link inventory: total=%d descriptor=%d management=%d imported=%d',
@@ -100,6 +113,20 @@ def log_link_inventory(context_client, descriptor_loader, profile_name: str) -> 
    return descriptor_links, management_links, imported_links





def _is_management_link(link: Link) -> bool:

    if link.link_type == LinkTypeEnum.LINKTYPE_MANAGEMENT:

        return True



    if 'mgmt' in link.name.lower():

        return True



    for endpoint_id in link.link_endpoint_ids:

        if 'mgmt' in endpoint_id.endpoint_uuid.uuid.lower():

            return True



    return False





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