[RFC,3/7] dts: revamp Topology model
Checks
Commit Message
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(-)
Comments
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
>
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>
Reviewed-by: Dean Marx <dmarx@iol.unh.edu>
@@ -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]