xref: /dpdk/dts/framework/testbed_model/tg_node.py (revision 7917b0d38e92e8b9ec5a870415b791420e10f11a) 
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_packet_and_capture(
55        self,
56        packet: Packet,
57        send_port: Port,
58        receive_port: Port,
59        filter_config: PacketFilteringConfig = PacketFilteringConfig(),
60        duration: float = 1,
61    ) -> list[Packet]:
62        """Send `packet`, return received traffic.
63
64        Send `packet` 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            packet: The packet 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_packet_and_capture(
79            packet,
80            send_port,
81            receive_port,
82            filter_config,
83            duration,
84        )
85
86    def close(self) -> None:
87        """Free all resources used by the node.
88
89        This extends the superclass method with TG cleanup.
90        """
91        self.traffic_generator.close()
92        super().close()
93