1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
|
# SPDX-License-Identifier: BSD-3-Clause
# Copyright(c) 2024 PANTHEON.tech s.r.o.
# Copyright(c) 2025 Arm Limited
"""Testbed capabilities.
This module provides a protocol that defines the common attributes of test cases and suites
and support for test environment capabilities.
Many test cases are testing features not available on all hardware.
On the other hand, some test cases or suites may not need the most complex topology available.
The module allows developers to mark test cases or suites to require certain hardware capabilities
or a particular topology.
There are differences between hardware and topology capabilities:
* Hardware capabilities are assumed to not be required when not specified.
* However, some topology is always available, so each test case or suite is assigned
a default topology if no topology is specified in the decorator.
Examples:
.. code:: python
from framework.test_suite import TestSuite, func_test
from framework.testbed_model.capability import LinkTopology, requires_link_topology
# The whole test suite (each test case within) doesn't require any links.
@requires_link_topology(LinkTopology.NO_LINK)
@func_test
class TestHelloWorld(TestSuite):
def hello_world_single_core(self):
...
.. code:: python
from framework.test_suite import TestSuite, func_test
from framework.testbed_model.capability import NicCapability, requires_nic_capability
class TestPmdBufferScatter(TestSuite):
# only the test case requires the SCATTERED_RX_ENABLED capability
# other test cases may not require it
@requires_nic_capability(NicCapability.SCATTERED_RX_ENABLED)
@func_test
def test_scatter_mbuf_2048(self):
"""
from enum import IntEnum, auto
from typing import TYPE_CHECKING, Callable
if TYPE_CHECKING:
from framework.test_suite import TestProtocol
class LinkTopology(IntEnum):
"""Supported topology types."""
#: A topology with no Traffic Generator.
NO_LINK = 0
#: A topology with one physical link between the SUT node and the TG node.
ONE_LINK = auto()
#: A topology with two physical links between the Sut node and the TG node.
TWO_LINKS = auto()
@classmethod
def default(cls) -> "LinkTopology":
"""The default topology required by test cases if not specified otherwise."""
return cls.TWO_LINKS
class NicCapability(IntEnum):
"""DPDK NIC capabilities.
The capabilities are used to mark test cases or suites that require a specific
DPDK NIC capability to run. The capabilities are used by the test framework to
determine whether a test case or suite can be run on the current testbed.
"""
#: Scattered packets Rx enabled.
SCATTERED_RX_ENABLED = 0
#: Device supports VLAN stripping.
PORT_RX_OFFLOAD_VLAN_STRIP = auto()
QUEUE_RX_OFFLOAD_VLAN_STRIP = auto()
#: Device supports L3 checksum offload.
PORT_RX_OFFLOAD_IPV4_CKSUM = auto()
QUEUE_RX_OFFLOAD_IPV4_CKSUM = auto()
#: Device supports L4 checksum offload.
PORT_RX_OFFLOAD_UDP_CKSUM = auto()
QUEUE_RX_OFFLOAD_UDP_CKSUM = auto()
#: Device supports L4 checksum offload.
PORT_RX_OFFLOAD_TCP_CKSUM = auto()
QUEUE_RX_OFFLOAD_TCP_CKSUM = auto()
#: Device supports Large Receive Offload.
PORT_RX_OFFLOAD_TCP_LRO = auto()
QUEUE_RX_OFFLOAD_TCP_LRO = auto()
#: Device supports QinQ (queue in queue) offload.
PORT_RX_OFFLOAD_QINQ_STRIP = auto()
QUEUE_RX_OFFLOAD_QINQ_STRIP = auto()
#: Device supports inner packet L3 checksum.
PORT_RX_OFFLOAD_OUTER_IPV4_CKSUM = auto()
QUEUE_RX_OFFLOAD_OUTER_IPV4_CKSUM = auto()
#: Device supports MACsec.
PORT_RX_OFFLOAD_MACSEC_STRIP = auto()
QUEUE_RX_OFFLOAD_MACSEC_STRIP = auto()
#: Device supports filtering of a VLAN Tag identifier.
PORT_RX_OFFLOAD_VLAN_FILTER = auto()
QUEUE_RX_OFFLOAD_VLAN_FILTER = auto()
#: Device supports VLAN offload.
PORT_RX_OFFLOAD_VLAN_EXTEND = auto()
QUEUE_RX_OFFLOAD_VLAN_EXTEND = auto()
#: Device supports receiving segmented mbufs.
PORT_RX_OFFLOAD_SCATTER = auto()
QUEUE_RX_OFFLOAD_SCATTER = auto()
#: Device supports Timestamp.
PORT_RX_OFFLOAD_TIMESTAMP = auto()
QUEUE_RX_OFFLOAD_TIMESTAMP = auto()
#: Device supports crypto processing while packet is received in NIC.
PORT_RX_OFFLOAD_SECURITY = auto()
QUEUE_RX_OFFLOAD_SECURITY = auto()
#: Device supports CRC stripping.
PORT_RX_OFFLOAD_KEEP_CRC = auto()
QUEUE_RX_OFFLOAD_KEEP_CRC = auto()
#: Device supports L4 checksum offload.
PORT_RX_OFFLOAD_SCTP_CKSUM = auto()
QUEUE_RX_OFFLOAD_SCTP_CKSUM = auto()
#: Device supports inner packet L4 checksum.
PORT_RX_OFFLOAD_OUTER_UDP_CKSUM = auto()
QUEUE_RX_OFFLOAD_OUTER_UDP_CKSUM = auto()
#: Device supports RSS hashing.
PORT_RX_OFFLOAD_RSS_HASH = auto()
QUEUE_RX_OFFLOAD_RSS_HASH = auto()
#: Device supports scatter Rx packets to segmented mbufs.
PORT_RX_OFFLOAD_BUFFER_SPLIT = auto()
QUEUE_RX_OFFLOAD_BUFFER_SPLIT = auto()
#: Device supports all checksum capabilities.
PORT_RX_OFFLOAD_CHECKSUM = auto()
QUEUE_RX_OFFLOAD_CHECKSUM = auto()
#: Device supports all VLAN capabilities.
PORT_RX_OFFLOAD_VLAN = auto()
QUEUE_RX_OFFLOAD_VLAN = auto()
#: Device supports Rx queue setup after device started.
RUNTIME_RX_QUEUE_SETUP = auto()
#: Device supports Tx queue setup after device started.
RUNTIME_TX_QUEUE_SETUP = auto()
#: Device supports shared Rx queue among ports within Rx domain and switch domain.
RXQ_SHARE = auto()
#: Device supports keeping flow rules across restart.
FLOW_RULE_KEEP = auto()
#: Device supports keeping shared flow objects across restart.
FLOW_SHARED_OBJECT_KEEP = auto()
#: Device supports multicast address filtering.
MCAST_FILTERING = auto()
#: Device supports flow ctrl.
FLOW_CTRL = auto()
#: Device is running on a physical function.
PHYSICAL_FUNCTION = auto()
#:
PORT_TX_OFFLOAD_VLAN_INSERT = auto()
QUEUE_TX_OFFLOAD_VLAN_INSERT = auto()
#:
PORT_TX_OFFLOAD_IPV4_CKSUM = auto()
QUEUE_TX_OFFLOAD_IPV4_CKSUM = auto()
#:
PORT_TX_OFFLOAD_UDP_CKSUM = auto()
QUEUE_TX_OFFLOAD_UDP_CKSUM = auto()
#:
PORT_TX_OFFLOAD_TCP_CKSUM = auto()
QUEUE_TX_OFFLOAD_TCP_CKSUM = auto()
#:
PORT_TX_OFFLOAD_SCTP_CKSUM = auto()
QUEUE_TX_OFFLOAD_SCTP_CKSUM = auto()
#:
PORT_TX_OFFLOAD_TCP_TSO = auto()
QUEUE_TX_OFFLOAD_TCP_TSO = auto()
#:
PORT_TX_OFFLOAD_UDP_TSO = auto()
QUEUE_TX_OFFLOAD_UDP_TSO = auto()
#:
PORT_TX_OFFLOAD_OUTER_IPV4_CKSUM = auto()
QUEUE_TX_OFFLOAD_OUTER_IPV4_CKSUM = auto()
#:
PORT_TX_OFFLOAD_QINQ_INSERT = auto()
QUEUE_TX_OFFLOAD_QINQ_INSERT = auto()
#:
PORT_TX_OFFLOAD_VXLAN_TNL_TSO = auto()
QUEUE_TX_OFFLOAD_VXLAN_TNL_TSO = auto()
#:
PORT_TX_OFFLOAD_GRE_TNL_TSO = auto()
QUEUE_TX_OFFLOAD_GRE_TNL_TSO = auto()
#:
PORT_TX_OFFLOAD_IPIP_TNL_TSO = auto()
QUEUE_TX_OFFLOAD_IPIP_TNL_TSO = auto()
#:
PORT_TX_OFFLOAD_GENEVE_TNL_TSO = auto()
QUEUE_TX_OFFLOAD_GENEVE_TNL_TSO = auto()
#:
PORT_TX_OFFLOAD_MACSEC_INSERT = auto()
QUEUE_TX_OFFLOAD_MACSEC_INSERT = auto()
#:
PORT_TX_OFFLOAD_MT_LOCKFREE = auto()
QUEUE_TX_OFFLOAD_MT_LOCKFREE = auto()
#:
PORT_TX_OFFLOAD_MULTI_SEGS = auto()
QUEUE_TX_OFFLOAD_MULTI_SEGS = auto()
#:
PORT_TX_OFFLOAD_MBUF_FAST_FREE = auto()
QUEUE_TX_OFFLOAD_MBUF_FAST_FREE = auto()
#:
PORT_TX_OFFLOAD_SECURITY = auto()
QUEUE_TX_OFFLOAD_SECURITY = auto()
#:
PORT_TX_OFFLOAD_UDP_TNL_TSO = auto()
QUEUE_TX_OFFLOAD_UDP_TNL_TSO = auto()
#:
PORT_TX_OFFLOAD_IP_TNL_TSO = auto()
QUEUE_TX_OFFLOAD_IP_TNL_TSO = auto()
#:
PORT_TX_OFFLOAD_OUTER_UDP_CKSUM = auto()
QUEUE_TX_OFFLOAD_OUTER_UDP_CKSUM = auto()
#:
PORT_TX_OFFLOAD_SEND_ON_TIMESTAMP = auto()
QUEUE_TX_OFFLOAD_SEND_ON_TIMESTAMP = auto()
def requires_link_topology(
link_topology: LinkTopology,
) -> Callable[[type["TestProtocol"]], type["TestProtocol"]]:
"""Decorator to set required topology type for a test case or test suite.
Args:
link_topology: The topology type the test suite or case requires.
Returns:
The decorated test case or test suite.
"""
from framework.testbed_model.capability import TopologyCapability
def add_required_topology(
test_case_or_suite: type["TestProtocol"],
) -> type["TestProtocol"]:
topology_capability = TopologyCapability.get_unique(link_topology)
topology_capability.set_required(test_case_or_suite)
return test_case_or_suite
return add_required_topology
def requires_nic_capability(
nic_capability: NicCapability,
) -> Callable[[type["TestProtocol"]], type["TestProtocol"]]:
"""Decorator to add a single required NIC capability to a test case or test suite.
Args:
nic_capability: The NIC capability that is required by the test case or test suite.
Returns:
The decorated test case or test suite.
"""
from framework.testbed_model.capability import DecoratedNicCapability
def add_required_capability(
test_case_or_suite: type["TestProtocol"],
) -> type["TestProtocol"]:
decorated = DecoratedNicCapability.get_unique(nic_capability)
decorated.add_to_required(test_case_or_suite)
return test_case_or_suite
return add_required_capability
|
