xref: /dpdk/dts/framework/testbed_model/topology.py (revision 949d1488e50f0516df065232640617b5a921ed95)

# SPDX-License-Identifier: BSD-3-Clause
# Copyright(c) 2024 PANTHEON.tech s.r.o.

"""Testbed topology representation.

A topology of a testbed captures what links are available between the testbed's nodes.
The link information then implies what type of topology is available.
"""

from dataclasses import dataclass
from os import environ
from typing import TYPE_CHECKING, Iterable

if TYPE_CHECKING or environ.get("DTS_DOC_BUILD"):
    from enum import Enum as NoAliasEnum
else:
    from aenum import NoAliasEnum

from framework.config import PortConfig
from framework.exception import ConfigurationError

from .port import Port


class TopologyType(int, NoAliasEnum):
    """Supported topology types."""

    #: A topology with no Traffic Generator.
    no_link = 0
    #: A topology with one physical link between the SUT node and the TG node.
    one_link = 1
    #: A topology with two physical links between the Sut node and the TG node.
    two_links = 2
    #: The default topology required by test cases if not specified otherwise.
    default = 2

    @classmethod
    def get_from_value(cls, value: int) -> "TopologyType":
        r"""Get the corresponding instance from value.

        :class:`~enum.Enum`\s that don't allow aliases don't know which instance should be returned
        as there could be multiple valid instances. Except for the :attr:`default` value,
        :class:`TopologyType` is a regular :class:`~enum.Enum`.
        When getting an instance from value, we're not interested in the default,
        since we already know the value, allowing us to remove the ambiguity.
        """
        match value:
            case 0:
                return TopologyType.no_link
            case 1:
                return TopologyType.one_link
            case 2:
                return TopologyType.two_links
            case _:
                raise ConfigurationError("More than two links in a topology are not supported.")


class Topology:
    """Testbed topology.

    The topology contains ports processed into ingress and egress ports.
    If there are no ports on a node, dummy ports (ports with no actual values) are stored.
    If there is only one link available, the ports of this link are stored
    as both ingress and egress ports.

    The dummy ports shouldn't be used. It's up to :class:`~framework.runner.DTSRunner`
    to ensure no test case or suite requiring actual links is executed
    when the topology prohibits it and up to the developers to make sure that test cases
    not requiring any links don't use any ports. Otherwise, the underlying methods
    using the ports will fail.

    Attributes:
        type: The type of the topology.
        tg_port_egress: The egress port of the TG node.
        sut_port_ingress: The ingress port of the SUT node.
        sut_port_egress: The egress port of the SUT node.
        tg_port_ingress: The ingress port of the TG node.
    """

    type: TopologyType
    tg_port_egress: Port
    sut_port_ingress: Port
    sut_port_egress: Port
    tg_port_ingress: Port

    def __init__(self, sut_ports: Iterable[Port], tg_ports: Iterable[Port]):
        """Create the topology from `sut_ports` and `tg_ports`.

        Args:
            sut_ports: The SUT node's ports.
            tg_ports: The TG node's ports.
        """
        port_links = []
        for sut_port in sut_ports:
            for tg_port in tg_ports:
                if (sut_port.identifier, sut_port.peer) == (
                    tg_port.peer,
                    tg_port.identifier,
                ):
                    port_links.append(PortLink(sut_port=sut_port, tg_port=tg_port))

        self.type = TopologyType.get_from_value(len(port_links))
        dummy_port = Port(
            "",
            PortConfig(
                pci="0000:00:00.0",
                os_driver_for_dpdk="",
                os_driver="",
                peer_node="",
                peer_pci="0000:00:00.0",
            ),
        )
        self.tg_port_egress = dummy_port
        self.sut_port_ingress = dummy_port
        self.sut_port_egress = dummy_port
        self.tg_port_ingress = dummy_port
        if self.type > TopologyType.no_link:
            self.tg_port_egress = port_links[0].tg_port
            self.sut_port_ingress = port_links[0].sut_port
            self.sut_port_egress = self.sut_port_ingress
            self.tg_port_ingress = self.tg_port_egress
        if self.type > TopologyType.one_link:
            self.sut_port_egress = port_links[1].sut_port
            self.tg_port_ingress = port_links[1].tg_port


@dataclass(slots=True, frozen=True)
class PortLink:
    """The physical, cabled connection between the ports.

    Attributes:
        sut_port: The port on the SUT node connected to `tg_port`.
        tg_port: The port on the TG node connected to `sut_port`.
    """

    sut_port: Port
    tg_port: Port