Re: [RFC PATCH 3/7] dts: revamp Topology model

[Nicholas Pratte <npratte@iol.unh.edu>]

Reviewed-by: Nicholas Pratte <npratte@iol.unh.edu>

On Mon, Feb 3, 2025 at 10:17 AM Luca Vizzarro <luca.vizzarro@arm.com> wrote:
>
> Change the Topology model to add further flexility in its usage as a
> standalone entry point to test suites.
>
> Signed-off-by: Luca Vizzarro <luca.vizzarro@arm.com>
> ---
>  dts/framework/testbed_model/topology.py | 85 +++++++++++++------------
>  1 file changed, 45 insertions(+), 40 deletions(-)
>
> diff --git a/dts/framework/testbed_model/topology.py b/dts/framework/testbed_model/topology.py
> index 814c3f3fe4..cf5c2c28ba 100644
> --- a/dts/framework/testbed_model/topology.py
> +++ b/dts/framework/testbed_model/topology.py
> @@ -9,10 +9,12 @@
>  """
>
>  from collections.abc import Iterator
> +from dataclasses import dataclass
>  from enum import Enum
>  from typing import NamedTuple
>
> -from framework.config.node import PortConfig
> +from typing_extensions import Self
> +
>  from framework.exception import ConfigurationError
>
>  from .port import Port
> @@ -43,35 +45,32 @@ class PortLink(NamedTuple):
>      tg_port: Port
>
>
> +@dataclass(frozen=True)
>  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
> +    If there are no ports on a node, accesses to :attr:`~Topology.tg_port_egress` and alike will
> +    raise an exception. 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.
> +    It's up to :class:`~framework.test_run.TestRun` 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.
> +        sut_ports: The SUT ports.
> +        tg_ports: The TG ports.
>      """
>
>      type: TopologyType
> -    tg_port_egress: Port
> -    sut_port_ingress: Port
> -    sut_port_egress: Port
> -    tg_port_ingress: Port
> +    sut_ports: list[Port]
> +    tg_ports: list[Port]
>
> -    def __init__(self, port_links: Iterator[PortLink]):
> +    @classmethod
> +    def from_port_links(cls, port_links: Iterator[PortLink]) -> Self:
>          """Create the topology from `port_links`.
>
>          Args:
> @@ -80,34 +79,40 @@ def __init__(self, port_links: Iterator[PortLink]):
>          Raises:
>              ConfigurationError: If an unsupported link topology is supplied.
>          """
> -        dummy_port = Port(
> -            "",
> -            PortConfig(
> -                name="dummy",
> -                pci="0000:00:00.0",
> -                os_driver_for_dpdk="",
> -                os_driver="",
> -            ),
> -        )
> -
> -        self.type = TopologyType.no_link
> -        self.tg_port_egress = dummy_port
> -        self.sut_port_ingress = dummy_port
> -        self.sut_port_egress = dummy_port
> -        self.tg_port_ingress = dummy_port
> +        type = TopologyType.no_link
>
>          if port_link := next(port_links, None):
> -            self.type = TopologyType.one_link
> -            self.tg_port_egress = port_link.tg_port
> -            self.sut_port_ingress = port_link.sut_port
> -            self.sut_port_egress = self.sut_port_ingress
> -            self.tg_port_ingress = self.tg_port_egress
> +            type = TopologyType.one_link
> +            sut_ports = [port_link.sut_port]
> +            tg_ports = [port_link.tg_port]
>
>              if port_link := next(port_links, None):
> -                self.type = TopologyType.two_links
> -                self.sut_port_egress = port_link.sut_port
> -                self.tg_port_ingress = port_link.tg_port
> +                type = TopologyType.two_links
> +                sut_ports.append(port_link.sut_port)
> +                tg_ports.append(port_link.tg_port)
>
>                  if next(port_links, None) is not None:
>                      msg = "More than two links in a topology are not supported."
>                      raise ConfigurationError(msg)
> +
> +        return cls(type, sut_ports, tg_ports)
> +
> +    @property
> +    def tg_port_egress(self) -> Port:
> +        """The egress port of the TG node."""
> +        return self.tg_ports[0]
> +
> +    @property
> +    def sut_port_ingress(self) -> Port:
> +        """The ingress port of the SUT node."""
> +        return self.sut_ports[0]
> +
> +    @property
> +    def sut_port_egress(self) -> Port:
> +        """The egress port of the SUT node."""
> +        return self.sut_ports[1 if self.type is TopologyType.two_links else 0]
> +
> +    @property
> +    def tg_port_ingress(self) -> Port:
> +        """The ingress port of the TG node."""
> +        return self.tg_ports[1 if self.type is TopologyType.two_links else 0]
> --
> 2.43.0
>