test/test_nsim.py

   1 from framework import VppTestCase
   2 from asfframework import VppTestRunner
   3 from config import config
   4 import unittest
   5 import re
   6
   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
  11 from util import ppp
  12
  13
  14 def create_stream(self, src_if, dst_if, count):
  15     packets = []
  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
  22         p = (
  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)
  26             / Raw(payload)
  27         )
  28         # store a copy of the packet in the packet info
  29         info.data = p.copy()
  30         # append the packet to the list
  31         packets.append(p)
  32
  33     # return the created packet list
  34     return packets
  35
  36
  37 def verify_capture(self, src_if, dst_if, capture, reply):
  38     packet_info = None
  39     for packet in capture:
  40         try:
  41             ip = packet[IP]
  42             udp = packet[UDP]
  43             # convert the payload to packet info object
  44             payload_info = self.payload_to_info(packet[Raw])
  45             # make sure the indexes match
  46             self.assert_equal(
  47                 payload_info.src, src_if.sw_if_index, "source sw_if_index"
  48             )
  49             self.assert_equal(
  50                 payload_info.dst, dst_if.sw_if_index, "destination sw_if_index"
  51             )
  52             packet_info = self.get_next_packet_info_for_interface2(
  53                 src_if.sw_if_index, dst_if.sw_if_index, packet_info
  54             )
  55             # make sure we didn't run out of saved packets
  56             self.assertIsNotNone(packet_info)
  57             self.assert_equal(
  58                 payload_info.index, packet_info.index, "packet info index"
  59             )
  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")
  65         except Exception:
  66             self.logger.error(ppp("Unexpected or invalid packet:", packet))
  67             raise
  68     remaining_packet = self.get_next_packet_info_for_interface2(
  69         src_if.sw_if_index, dst_if.sw_if_index, packet_info
  70     )
  71     self.assertIsNone(
  72         remaining_packet,
  73         "Interface %s: Packet expected from interface "
  74         "%s didn't arrive" % (dst_if.name, src_if.name),
  75     )
  76
  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:])
  81     self.assertTrue(
  82         actual_delay >= 100000, f"Delay is lower than expected: {actual_delay} < 100000"
  83     )
  84
  85
  86 @unittest.skipIf("nsim" in config.excluded_plugins, "Exclude NSIM plugin tests")
  87 class TestNsimCli(VppTestCase):
  88     """NSIM plugin tests [CLI]"""
  89
  90     @classmethod
  91     def setUpClass(cls):
  92         super(TestNsimCli, cls).setUpClass()
  93         try:
  94             cls.create_pg_interfaces(range(2))
  95             for i in cls.pg_interfaces:
  96                 i.config_ip4()
  97                 i.resolve_arp()
  98                 i.admin_up()
  99         except Exception:
 100             cls.tearDownClass()
 101             raise
 102
 103     @classmethod
 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:
 108             i.unconfig_ip4()
 109             i.admin_down()
 110         super(TestNsimCli, cls).tearDownClass()
 111
 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()
 118
 119         self.vapi.cli(
 120             "set nsim delay 100.0 ms bandwidth 1 gbit packet-size 128 drop-fraction 0.0"
 121         )
 122         self.vapi.cli("nsim cross-connect enable-disable pg0 pg1")
 123         self.vapi.cli("nsim output-feature enable-disable pg0")
 124
 125         self.pg_start()
 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)
 133
 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")
 140         self.vapi.cli(
 141             "set nsim delay 1 us bandwidth 1 gbit packet-size 128 drop-fraction 1.0 packets-per-drop 0"
 142         )
 143
 144         self.pg_start()
 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)
 150
 151
 152 @unittest.skipIf("nsim" in config.excluded_plugins, "Exclude NSIM plugin tests")
 153 class TestNsimApi(VppTestCase):
 154     """NSIM plugin tests [API]"""
 155
 156     @classmethod
 157     def setUpClass(cls):
 158         super(TestNsimApi, cls).setUpClass()
 159         try:
 160             cls.create_pg_interfaces(range(2))
 161             for i in cls.pg_interfaces:
 162                 i.config_ip4()
 163                 i.resolve_arp()
 164                 i.admin_up()
 165         except Exception:
 166             cls.tearDownClass()
 167             raise
 168
 169     @classmethod
 170     def tearDownClass(cls):
 171         cls.vapi.nsim_cross_connect_enable_disable(
 172             enable_disable=False, sw_if_index0=1, sw_if_index1=2
 173         )
 174         cls.vapi.nsim_output_feature_enable_disable(enable_disable=False, sw_if_index=1)
 175         for i in cls.pg_interfaces:
 176             i.unconfig_ip4()
 177             i.admin_down()
 178         super(TestNsimApi, cls).tearDownClass()
 179
 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()
 186
 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,
 192             packets_per_drop=0,
 193             packets_per_reorder=0,
 194         )
 195         self.vapi.nsim_cross_connect_enable_disable(
 196             enable_disable=True, sw_if_index0=1, sw_if_index1=2
 197         )
 198         self.vapi.nsim_output_feature_enable_disable(enable_disable=True, sw_if_index=1)
 199         self.pg_start()
 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)
 206
 207
 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]"""
 213
 214     @classmethod
 215     def setUpClass(cls):
 216         super(TestNsimApi2, cls).setUpClass()
 217         try:
 218             cls.create_pg_interfaces(range(2))
 219             for i in cls.pg_interfaces:
 220                 i.config_ip4()
 221                 i.resolve_arp()
 222                 i.admin_up()
 223         except Exception:
 224             cls.tearDownClass()
 225             raise
 226
 227     @classmethod
 228     def tearDownClass(cls):
 229         cls.vapi.nsim_cross_connect_enable_disable(
 230             enable_disable=False, sw_if_index0=1, sw_if_index1=2
 231         )
 232         cls.vapi.nsim_output_feature_enable_disable(enable_disable=False, sw_if_index=1)
 233         for i in cls.pg_interfaces:
 234             i.unconfig_ip4()
 235             i.admin_down()
 236         super(TestNsimApi2, cls).tearDownClass()
 237
 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")
 245
 246         self.vapi.nsim_configure2(
 247             delay_in_usec=10,
 248             average_packet_size=128,
 249             bandwidth_in_bits_per_second=100000000,
 250             packets_per_drop=1,
 251             packets_per_reorder=0,
 252         )
 253         self.vapi.nsim_cross_connect_enable_disable(
 254             enable_disable=True, sw_if_index0=1, sw_if_index1=2
 255         )
 256         self.vapi.nsim_output_feature_enable_disable(enable_disable=True, sw_if_index=1)
 257
 258         self.pg_start()
 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)
 264
 265
 266 if __name__ == "__main__":
 267     unittest.main(testRunner=VppTestRunner)