1 from framework import VppTestCase
2 from asfframework import VppTestRunner
3 from config import config
7 from scapy.layers.l2 import Ether
8 from scapy.layers.inet import IP, UDP
9 from scapy.packet import Raw
10 from random import randint
14 def create_stream(self, src_if, dst_if, count):
16 for i in range(count):
17 # create packet info stored in the test case instance
18 info = self.create_packet_info(src_if, dst_if)
19 # convert the info into packet payload
20 payload = self.info_to_payload(info)
21 # create the packet itself
23 Ether(dst=src_if.local_mac, src=src_if.remote_mac)
24 / IP(src=src_if.remote_ip4, dst=dst_if.remote_ip4)
25 / UDP(sport=randint(49152, 65535), dport=5678)
28 # store a copy of the packet in the packet info
30 # append the packet to the list
33 # return the created packet list
37 def verify_capture(self, src_if, dst_if, capture, reply):
39 for packet in capture:
43 # convert the payload to packet info object
44 payload_info = self.payload_to_info(packet[Raw])
45 # make sure the indexes match
47 payload_info.src, src_if.sw_if_index, "source sw_if_index"
50 payload_info.dst, dst_if.sw_if_index, "destination sw_if_index"
52 packet_info = self.get_next_packet_info_for_interface2(
53 src_if.sw_if_index, dst_if.sw_if_index, packet_info
55 # make sure we didn't run out of saved packets
56 self.assertIsNotNone(packet_info)
58 payload_info.index, packet_info.index, "packet info index"
60 saved_packet = packet_info.data # fetch the saved packet
61 # assert the values match
62 self.assert_equal(ip.src, saved_packet[IP].src, "IP source address")
63 # ... more assertions here
64 self.assert_equal(udp.sport, saved_packet[UDP].sport, "UDP source port")
66 self.logger.error(ppp("Unexpected or invalid packet:", packet))
68 remaining_packet = self.get_next_packet_info_for_interface2(
69 src_if.sw_if_index, dst_if.sw_if_index, packet_info
73 "Interface %s: Packet expected from interface "
74 "%s didn't arrive" % (dst_if.name, src_if.name),
77 # find timestamps and get actual delay
78 pattern = r"\d{2}:\d{2}:\d{2}:\d{6}"
79 timestamps = re.findall(pattern, reply)
80 actual_delay = int(timestamps[2][9:]) - int(timestamps[0][9:])
82 actual_delay >= 100000, f"Delay is lower than expected: {actual_delay} < 100000"
86 @unittest.skipIf("nsim" in config.excluded_plugins, "Exclude NSIM plugin tests")
87 class TestNsimCli(VppTestCase):
88 """NSIM plugin tests [CLI]"""
92 super(TestNsimCli, cls).setUpClass()
94 cls.create_pg_interfaces(range(2))
95 for i in cls.pg_interfaces:
104 def tearDownClass(cls):
105 cls.vapi.cli("nsim cross-connect enable-disable pg0 pg1 disable")
106 cls.vapi.cli("nsim output-feature enable-disable pg0 disable")
107 for i in cls.pg_interfaces:
110 super(TestNsimCli, cls).tearDownClass()
112 def test_nsim_delay(self):
113 """Add 100ms delay"""
114 packets = create_stream(self, self.pg0, self.pg1, 5)
115 self.pg0.add_stream(packets)
116 self.pg0.enable_capture()
117 self.pg1.enable_capture()
120 "set nsim delay 100.0 ms bandwidth 1 gbit packet-size 128 drop-fraction 0.0"
122 self.vapi.cli("nsim cross-connect enable-disable pg0 pg1")
123 self.vapi.cli("nsim output-feature enable-disable pg0")
126 capture = self.pg1.get_capture()
127 self.pg0.assert_nothing_captured()
128 reply = self.vapi.cli("show trace")
129 verify_capture(self, self.pg0, self.pg1, capture, reply)
130 self.assertIn("nsim", reply)
131 reply = self.vapi.cli("show nsim")
132 self.assertIn("delay: 100.0 ms", reply)
134 def test_nsim_drop(self):
135 """Drop all packets"""
136 packets = create_stream(self, self.pg0, self.pg1, 5)
137 self.pg0.add_stream(packets)
138 self.vapi.cli("clear trace")
139 # test fails if running test-debug and no delay is set ("invalid delay 0.00")
141 "set nsim delay 1 us bandwidth 1 gbit packet-size 128 drop-fraction 1.0 packets-per-drop 0"
145 self.pg1.assert_nothing_captured()
146 reply = self.vapi.cli("show nsim")
147 self.assertIn("drop fraction: 1.0", reply)
148 reply = self.vapi.cli("show trace")
149 self.assertIn("sw_if_index -1", reply)
152 @unittest.skipIf("nsim" in config.excluded_plugins, "Exclude NSIM plugin tests")
153 class TestNsimApi(VppTestCase):
154 """NSIM plugin tests [API]"""
158 super(TestNsimApi, cls).setUpClass()
160 cls.create_pg_interfaces(range(2))
161 for i in cls.pg_interfaces:
170 def tearDownClass(cls):
171 cls.vapi.nsim_cross_connect_enable_disable(
172 enable_disable=False, sw_if_index0=1, sw_if_index1=2
174 cls.vapi.nsim_output_feature_enable_disable(enable_disable=False, sw_if_index=1)
175 for i in cls.pg_interfaces:
178 super(TestNsimApi, cls).tearDownClass()
180 def test_nsim_delay(self):
181 """Add 100ms delay"""
182 packets = create_stream(self, self.pg0, self.pg1, 5)
183 self.pg0.add_stream(packets)
184 self.pg0.enable_capture()
185 self.pg1.enable_capture()
187 # "show nsim" shows 99.9ms if delay is exactly 100000
188 self.vapi.nsim_configure2(
189 delay_in_usec=100001,
190 average_packet_size=128,
191 bandwidth_in_bits_per_second=100000000000,
193 packets_per_reorder=0,
195 self.vapi.nsim_cross_connect_enable_disable(
196 enable_disable=True, sw_if_index0=1, sw_if_index1=2
198 self.vapi.nsim_output_feature_enable_disable(enable_disable=True, sw_if_index=1)
200 capture = self.pg1.get_capture()
201 reply = self.vapi.cli("show trace")
202 verify_capture(self, self.pg0, self.pg1, capture, reply)
203 self.assertIn("nsim", reply)
204 reply = self.vapi.cli("show nsim")
205 self.assertIn("delay: 100.0 ms", reply)
208 # has to be separated, otherwise we get "VPP API client: read failed"
209 # when configuring NSIM (nsim_configure2) and then VPP crashes on teardown
210 @unittest.skipIf("nsim" in config.excluded_plugins, "Exclude NSIM plugin tests")
211 class TestNsimApi2(VppTestCase):
212 """NSIM plugin tests [API]"""
216 super(TestNsimApi2, cls).setUpClass()
218 cls.create_pg_interfaces(range(2))
219 for i in cls.pg_interfaces:
228 def tearDownClass(cls):
229 cls.vapi.nsim_cross_connect_enable_disable(
230 enable_disable=False, sw_if_index0=1, sw_if_index1=2
232 cls.vapi.nsim_output_feature_enable_disable(enable_disable=False, sw_if_index=1)
233 for i in cls.pg_interfaces:
236 super(TestNsimApi2, cls).tearDownClass()
238 def test_nsim_drop(self):
239 """Drop all packets"""
240 packets = create_stream(self, self.pg0, self.pg1, 5)
241 self.pg0.add_stream(packets)
242 self.pg0.enable_capture()
243 self.pg1.enable_capture()
244 self.vapi.cli("clear trace")
246 self.vapi.nsim_configure2(
248 average_packet_size=128,
249 bandwidth_in_bits_per_second=100000000,
251 packets_per_reorder=0,
253 self.vapi.nsim_cross_connect_enable_disable(
254 enable_disable=True, sw_if_index0=1, sw_if_index1=2
256 self.vapi.nsim_output_feature_enable_disable(enable_disable=True, sw_if_index=1)
259 self.pg1.assert_nothing_captured()
260 reply = self.vapi.cli("show nsim")
261 self.assertIn("drop fraction: 1.0", reply)
262 reply = self.vapi.cli("show trace")
263 self.assertIn("sw_if_index -1", reply)
266 if __name__ == "__main__":
267 unittest.main(testRunner=VppTestRunner)