diff options
Diffstat (limited to 'test/test_nsim.py')
| -rw-r--r-- | test/test_nsim.py | 267 |
1 files changed, 267 insertions, 0 deletions
diff --git a/test/test_nsim.py b/test/test_nsim.py new file mode 100644 index 00000000000..16b5fd07882 --- /dev/null +++ b/test/test_nsim.py @@ -0,0 +1,267 @@ +from framework import VppTestCase +from asfframework import VppTestRunner +from config import config +import unittest +import re + +from scapy.layers.l2 import Ether +from scapy.layers.inet import IP, UDP +from scapy.packet import Raw +from random import randint +from util import ppp + + +def create_stream(self, src_if, dst_if, count): + packets = [] + for i in range(count): + # create packet info stored in the test case instance + info = self.create_packet_info(src_if, dst_if) + # convert the info into packet payload + payload = self.info_to_payload(info) + # create the packet itself + p = ( + Ether(dst=src_if.local_mac, src=src_if.remote_mac) + / IP(src=src_if.remote_ip4, dst=dst_if.remote_ip4) + / UDP(sport=randint(49152, 65535), dport=5678) + / Raw(payload) + ) + # store a copy of the packet in the packet info + info.data = p.copy() + # append the packet to the list + packets.append(p) + + # return the created packet list + return packets + + +def verify_capture(self, src_if, dst_if, capture, reply): + packet_info = None + for packet in capture: + try: + ip = packet[IP] + udp = packet[UDP] + # convert the payload to packet info object + payload_info = self.payload_to_info(packet[Raw]) + # make sure the indexes match + self.assert_equal( + payload_info.src, src_if.sw_if_index, "source sw_if_index" + ) + self.assert_equal( + payload_info.dst, dst_if.sw_if_index, "destination sw_if_index" + ) + packet_info = self.get_next_packet_info_for_interface2( + src_if.sw_if_index, dst_if.sw_if_index, packet_info + ) + # make sure we didn't run out of saved packets + self.assertIsNotNone(packet_info) + self.assert_equal( + payload_info.index, packet_info.index, "packet info index" + ) + saved_packet = packet_info.data # fetch the saved packet + # assert the values match + self.assert_equal(ip.src, saved_packet[IP].src, "IP source address") + # ... more assertions here + self.assert_equal(udp.sport, saved_packet[UDP].sport, "UDP source port") + except Exception: + self.logger.error(ppp("Unexpected or invalid packet:", packet)) + raise + remaining_packet = self.get_next_packet_info_for_interface2( + src_if.sw_if_index, dst_if.sw_if_index, packet_info + ) + self.assertIsNone( + remaining_packet, + "Interface %s: Packet expected from interface " + "%s didn't arrive" % (dst_if.name, src_if.name), + ) + + # find timestamps and get actual delay + pattern = r"\d{2}:\d{2}:\d{2}:\d{6}" + timestamps = re.findall(pattern, reply) + actual_delay = int(timestamps[2][9:]) - int(timestamps[0][9:]) + self.assertTrue( + actual_delay >= 100000, f"Delay is lower than expected: {actual_delay} < 100000" + ) + + +@unittest.skipIf("nsim" in config.excluded_plugins, "Exclude NSIM plugin tests") +class TestNsimCli(VppTestCase): + """NSIM plugin tests [CLI]""" + + @classmethod + def setUpClass(cls): + super(TestNsimCli, cls).setUpClass() + try: + cls.create_pg_interfaces(range(2)) + for i in cls.pg_interfaces: + i.config_ip4() + i.resolve_arp() + i.admin_up() + except Exception: + cls.tearDownClass() + raise + + @classmethod + def tearDownClass(cls): + cls.vapi.cli("nsim cross-connect enable-disable pg0 pg1 disable") + cls.vapi.cli("nsim output-feature enable-disable pg0 disable") + for i in cls.pg_interfaces: + i.unconfig_ip4() + i.admin_down() + super(TestNsimCli, cls).tearDownClass() + + def test_nsim_delay(self): + """Add 100ms delay""" + packets = create_stream(self, self.pg0, self.pg1, 5) + self.pg0.add_stream(packets) + self.pg0.enable_capture() + self.pg1.enable_capture() + + self.vapi.cli( + "set nsim delay 100.0 ms bandwidth 1 gbit packet-size 128 drop-fraction 0.0" + ) + self.vapi.cli("nsim cross-connect enable-disable pg0 pg1") + self.vapi.cli("nsim output-feature enable-disable pg0") + + self.pg_start() + capture = self.pg1.get_capture() + self.pg0.assert_nothing_captured() + reply = self.vapi.cli("show trace") + verify_capture(self, self.pg0, self.pg1, capture, reply) + self.assertIn("nsim", reply) + reply = self.vapi.cli("show nsim") + self.assertIn("delay: 100.0 ms", reply) + + def test_nsim_drop(self): + """Drop all packets""" + packets = create_stream(self, self.pg0, self.pg1, 5) + self.pg0.add_stream(packets) + self.vapi.cli("clear trace") + # test fails if running test-debug and no delay is set ("invalid delay 0.00") + self.vapi.cli( + "set nsim delay 1 us bandwidth 1 gbit packet-size 128 drop-fraction 1.0 packets-per-drop 0" + ) + + self.pg_start() + self.pg1.assert_nothing_captured() + reply = self.vapi.cli("show nsim") + self.assertIn("drop fraction: 1.0", reply) + reply = self.vapi.cli("show trace") + self.assertIn("sw_if_index -1", reply) + + +@unittest.skipIf("nsim" in config.excluded_plugins, "Exclude NSIM plugin tests") +class TestNsimApi(VppTestCase): + """NSIM plugin tests [API]""" + + @classmethod + def setUpClass(cls): + super(TestNsimApi, cls).setUpClass() + try: + cls.create_pg_interfaces(range(2)) + for i in cls.pg_interfaces: + i.config_ip4() + i.resolve_arp() + i.admin_up() + except Exception: + cls.tearDownClass() + raise + + @classmethod + def tearDownClass(cls): + cls.vapi.nsim_cross_connect_enable_disable( + enable_disable=False, sw_if_index0=1, sw_if_index1=2 + ) + cls.vapi.nsim_output_feature_enable_disable(enable_disable=False, sw_if_index=1) + for i in cls.pg_interfaces: + i.unconfig_ip4() + i.admin_down() + super(TestNsimApi, cls).tearDownClass() + + def test_nsim_delay(self): + """Add 100ms delay""" + packets = create_stream(self, self.pg0, self.pg1, 5) + self.pg0.add_stream(packets) + self.pg0.enable_capture() + self.pg1.enable_capture() + + # "show nsim" shows 99.9ms if delay is exactly 100000 + self.vapi.nsim_configure2( + delay_in_usec=100001, + average_packet_size=128, + bandwidth_in_bits_per_second=100000000000, + packets_per_drop=0, + packets_per_reorder=0, + ) + self.vapi.nsim_cross_connect_enable_disable( + enable_disable=True, sw_if_index0=1, sw_if_index1=2 + ) + self.vapi.nsim_output_feature_enable_disable(enable_disable=True, sw_if_index=1) + self.pg_start() + capture = self.pg1.get_capture() + reply = self.vapi.cli("show trace") + verify_capture(self, self.pg0, self.pg1, capture, reply) + self.assertIn("nsim", reply) + reply = self.vapi.cli("show nsim") + self.assertIn("delay: 100.0 ms", reply) + + +# has to be separated, otherwise we get "VPP API client: read failed" +# when configuring NSIM (nsim_configure2) and then VPP crashes on teardown +@unittest.skipIf("nsim" in config.excluded_plugins, "Exclude NSIM plugin tests") +class TestNsimApi2(VppTestCase): + """NSIM plugin tests [API]""" + + @classmethod + def setUpClass(cls): + super(TestNsimApi2, cls).setUpClass() + try: + cls.create_pg_interfaces(range(2)) + for i in cls.pg_interfaces: + i.config_ip4() + i.resolve_arp() + i.admin_up() + except Exception: + cls.tearDownClass() + raise + + @classmethod + def tearDownClass(cls): + cls.vapi.nsim_cross_connect_enable_disable( + enable_disable=False, sw_if_index0=1, sw_if_index1=2 + ) + cls.vapi.nsim_output_feature_enable_disable(enable_disable=False, sw_if_index=1) + for i in cls.pg_interfaces: + i.unconfig_ip4() + i.admin_down() + super(TestNsimApi2, cls).tearDownClass() + + def test_nsim_drop(self): + """Drop all packets""" + packets = create_stream(self, self.pg0, self.pg1, 5) + self.pg0.add_stream(packets) + self.pg0.enable_capture() + self.pg1.enable_capture() + self.vapi.cli("clear trace") + + self.vapi.nsim_configure2( + delay_in_usec=10, + average_packet_size=128, + bandwidth_in_bits_per_second=100000000, + packets_per_drop=1, + packets_per_reorder=0, + ) + self.vapi.nsim_cross_connect_enable_disable( + enable_disable=True, sw_if_index0=1, sw_if_index1=2 + ) + self.vapi.nsim_output_feature_enable_disable(enable_disable=True, sw_if_index=1) + + self.pg_start() + self.pg1.assert_nothing_captured() + reply = self.vapi.cli("show nsim") + self.assertIn("drop fraction: 1.0", reply) + reply = self.vapi.cli("show trace") + self.assertIn("sw_if_index -1", reply) + + +if __name__ == "__main__": + unittest.main(testRunner=VppTestRunner) |