1# SPDX-License-Identifier: BSD-3-Clause 2# Copyright(c) 2024 PANTHEON.tech s.r.o. 3 4"""Testbed topology representation. 5 6A topology of a testbed captures what links are available between the testbed's nodes. 7The link information then implies what type of topology is available. 8""" 9 10from dataclasses import dataclass 11from os import environ 12from typing import TYPE_CHECKING, Iterable 13 14if TYPE_CHECKING or environ.get("DTS_DOC_BUILD"): 15 from enum import Enum as NoAliasEnum 16else: 17 from aenum import NoAliasEnum 18 19from framework.config import PortConfig 20from framework.exception import ConfigurationError 21 22from .port import Port 23 24 25class TopologyType(int, NoAliasEnum): 26 """Supported topology types.""" 27 28 #: A topology with no Traffic Generator. 29 no_link = 0 30 #: A topology with one physical link between the SUT node and the TG node. 31 one_link = 1 32 #: A topology with two physical links between the Sut node and the TG node. 33 two_links = 2 34 #: The default topology required by test cases if not specified otherwise. 35 default = 2 36 37 @classmethod 38 def get_from_value(cls, value: int) -> "TopologyType": 39 r"""Get the corresponding instance from value. 40 41 :class:`~enum.Enum`\s that don't allow aliases don't know which instance should be returned 42 as there could be multiple valid instances. Except for the :attr:`default` value, 43 :class:`TopologyType` is a regular :class:`~enum.Enum`. 44 When getting an instance from value, we're not interested in the default, 45 since we already know the value, allowing us to remove the ambiguity. 46 """ 47 match value: 48 case 0: 49 return TopologyType.no_link 50 case 1: 51 return TopologyType.one_link 52 case 2: 53 return TopologyType.two_links 54 case _: 55 raise ConfigurationError("More than two links in a topology are not supported.") 56 57 58class Topology: 59 """Testbed topology. 60 61 The topology contains ports processed into ingress and egress ports. 62 If there are no ports on a node, dummy ports (ports with no actual values) are stored. 63 If there is only one link available, the ports of this link are stored 64 as both ingress and egress ports. 65 66 The dummy ports shouldn't be used. It's up to :class:`~framework.runner.DTSRunner` 67 to ensure no test case or suite requiring actual links is executed 68 when the topology prohibits it and up to the developers to make sure that test cases 69 not requiring any links don't use any ports. Otherwise, the underlying methods 70 using the ports will fail. 71 72 Attributes: 73 type: The type of the topology. 74 tg_port_egress: The egress port of the TG node. 75 sut_port_ingress: The ingress port of the SUT node. 76 sut_port_egress: The egress port of the SUT node. 77 tg_port_ingress: The ingress port of the TG node. 78 """ 79 80 type: TopologyType 81 tg_port_egress: Port 82 sut_port_ingress: Port 83 sut_port_egress: Port 84 tg_port_ingress: Port 85 86 def __init__(self, sut_ports: Iterable[Port], tg_ports: Iterable[Port]): 87 """Create the topology from `sut_ports` and `tg_ports`. 88 89 Args: 90 sut_ports: The SUT node's ports. 91 tg_ports: The TG node's ports. 92 """ 93 port_links = [] 94 for sut_port in sut_ports: 95 for tg_port in tg_ports: 96 if (sut_port.identifier, sut_port.peer) == ( 97 tg_port.peer, 98 tg_port.identifier, 99 ): 100 port_links.append(PortLink(sut_port=sut_port, tg_port=tg_port)) 101 102 self.type = TopologyType.get_from_value(len(port_links)) 103 dummy_port = Port( 104 "", 105 PortConfig( 106 pci="0000:00:00.0", 107 os_driver_for_dpdk="", 108 os_driver="", 109 peer_node="", 110 peer_pci="0000:00:00.0", 111 ), 112 ) 113 self.tg_port_egress = dummy_port 114 self.sut_port_ingress = dummy_port 115 self.sut_port_egress = dummy_port 116 self.tg_port_ingress = dummy_port 117 if self.type > TopologyType.no_link: 118 self.tg_port_egress = port_links[0].tg_port 119 self.sut_port_ingress = port_links[0].sut_port 120 self.sut_port_egress = self.sut_port_ingress 121 self.tg_port_ingress = self.tg_port_egress 122 if self.type > TopologyType.one_link: 123 self.sut_port_egress = port_links[1].sut_port 124 self.tg_port_ingress = port_links[1].tg_port 125 126 127@dataclass(slots=True, frozen=True) 128class PortLink: 129 """The physical, cabled connection between the ports. 130 131 Attributes: 132 sut_port: The port on the SUT node connected to `tg_port`. 133 tg_port: The port on the TG node connected to `sut_port`. 134 """ 135 136 sut_port: Port 137 tg_port: Port 138