xref: /dpdk/dts/framework/testbed_model/tg_node.py (revision 2e69387a656396a7382fe27d8d9f61f1b5890573) 
1# SPDX-License-Identifier: BSD-3-Clause
2# Copyright(c) 2010-2014 Intel Corporation
3# Copyright(c) 2022 University of New Hampshire
4# Copyright(c) 2023 PANTHEON.tech s.r.o.
5
6"""Traffic generator node.
7
8A traffic generator (TG) generates traffic that's sent towards the SUT node.
9A TG node is where the TG runs.
10"""
11
12from scapy.packet import Packet  # type: ignore[import-untyped]
13
14from framework.config import TGNodeConfiguration
15from framework.testbed_model.traffic_generator.capturing_traffic_generator import (
16    PacketFilteringConfig,
17)
18
19from .node import Node
20from .port import Port
21from .traffic_generator import CapturingTrafficGenerator, create_traffic_generator
22
23
24class TGNode(Node):
25    """The traffic generator node.
26
27    The TG node extends :class:`Node` with TG specific features:
28
29        * Traffic generator initialization,
30        * The sending of traffic and receiving packets,
31        * The sending of traffic without receiving packets.
32
33    Not all traffic generators are capable of capturing traffic, which is why there
34    must be a way to send traffic without that.
35
36    Attributes:
37        traffic_generator: The traffic generator running on the node.
38    """
39
40    traffic_generator: CapturingTrafficGenerator
41
42    def __init__(self, node_config: TGNodeConfiguration):
43        """Extend the constructor with TG node specifics.
44
45        Initialize the traffic generator on the TG node.
46
47        Args:
48            node_config: The TG node's test run configuration.
49        """
50        super().__init__(node_config)
51        self.traffic_generator = create_traffic_generator(self, node_config.traffic_generator)
52        self._logger.info(f"Created node: {self.name}")
53
54    def send_packets_and_capture(
55        self,
56        packets: list[Packet],
57        send_port: Port,
58        receive_port: Port,
59        filter_config: PacketFilteringConfig = PacketFilteringConfig(),
60        duration: float = 1,
61    ) -> list[Packet]:
62        """Send `packets`, return received traffic.
63
64        Send `packets` on `send_port` and then return all traffic captured
65        on `receive_port` for the given duration. Also record the captured traffic
66        in a pcap file.
67
68        Args:
69            packets: The packets to send.
70            send_port: The egress port on the TG node.
71            receive_port: The ingress port in the TG node.
72            filter_config: The filter to use when capturing packets.
73            duration: Capture traffic for this amount of time after sending `packet`.
74
75        Returns:
76             A list of received packets. May be empty if no packets are captured.
77        """
78        return self.traffic_generator.send_packets_and_capture(
79            packets,
80            send_port,
81            receive_port,
82            filter_config,
83            duration,
84        )
85
86    def send_packets(self, packets: list[Packet], port: Port):
87        """Send packets without capturing resulting received packets.
88
89        Args:
90            packets: Packets to send.
91            port: Port to send the packets on.
92        """
93        self.traffic_generator.send_packets(packets, port)
94
95    def close(self) -> None:
96        """Free all resources used by the node.
97
98        This extends the superclass method with TG cleanup.
99        """
100        self.traffic_generator.close()
101        super().close()
102