test/test_interface_crud.py

   1 #!/usr/bin/env python3
   2 """CRUD tests of APIs (Create, Read, Update, Delete) HLD:
   3
   4 - interface up/down/add/delete - interface type:
   5     - pg (TBD)
   6     - loopback
   7     - vhostuser (TBD)
   8     - af_packet (TBD)
   9     - netmap (TBD)
  10     - tuntap (root privileges needed)
  11     - vxlan (TBD)
  12 """
  13
  14 import unittest
  15
  16 from scapy.layers.inet import IP, ICMP
  17 from scapy.layers.l2 import Ether
  18
  19 from framework import VppTestCase, VppTestRunner
  20
  21
  22 class TestLoopbackInterfaceCRUD(VppTestCase):
  23     """CRUD Loopback
  24
  25     """
  26
  27     @classmethod
  28     def setUpClass(cls):
  29         super(TestLoopbackInterfaceCRUD, cls).setUpClass()
  30         try:
  31             cls.create_pg_interfaces(range(1))
  32             for i in cls.pg_interfaces:
  33                 i.config_ip4().resolve_arp()
  34         except:
  35             cls.tearDownClass()
  36             raise
  37
  38     @classmethod
  39     def tearDownClass(cls):
  40         super(TestLoopbackInterfaceCRUD, cls).tearDownClass()
  41
  42     @staticmethod
  43     def create_icmp_stream(src_if, dst_ifs):
  44         """
  45
  46         :param VppInterface src_if: Packets are send to this interface,
  47             using this interfaces remote host.
  48         :param list dst_ifs: IPv4 ICMP requests are send to interfaces
  49             addresses.
  50         :return: List of generated packets.
  51         """
  52         pkts = []
  53         for i in dst_ifs:
  54             p = (Ether(dst=src_if.local_mac, src=src_if.remote_mac) /
  55                  IP(src=src_if.remote_ip4, dst=i.local_ip4) /
  56                  ICMP(id=i.sw_if_index, type='echo-request'))
  57             pkts.append(p)
  58         return pkts
  59
  60     def verify_icmp(self, capture, request_src_if, dst_ifs):
  61         """
  62
  63         :param capture: Capture to verify.
  64         :param VppInterface request_src_if: Interface where was send packets.
  65         :param list dst_ifs: Interfaces where was generated IPv4 ICMP requests.
  66         """
  67         rcvd_icmp_pkts = []
  68         for pkt in capture:
  69             try:
  70                 ip = pkt[IP]
  71                 icmp = pkt[ICMP]
  72             except IndexError:
  73                 pass
  74             else:
  75                 info = (ip.src, ip.dst, icmp.type, icmp.id)
  76                 rcvd_icmp_pkts.append(info)
  77
  78         for i in dst_ifs:
  79             # 0 - icmp echo response
  80             info = (i.local_ip4, request_src_if.remote_ip4, 0, i.sw_if_index)
  81             self.assertIn(info, rcvd_icmp_pkts)
  82
  83     def test_crud(self):
  84         # create
  85         loopbacks = self.create_loopback_interfaces(20)
  86         for i in loopbacks:
  87             i.local_ip4_prefix_len = 32
  88             i.config_ip4().admin_up()
  89
  90         # read (check sw if dump, ip4 fib, ip6 fib)
  91         if_dump = self.vapi.sw_interface_dump(name_filter_valid=True,
  92                                               name_filter='loop')
  93         fib4_dump = self.vapi.ip_route_dump(0)
  94         for i in loopbacks:
  95             self.assertTrue(i.is_interface_config_in_dump(if_dump))
  96             self.assertTrue(i.is_ip4_entry_in_fib_dump(fib4_dump))
  97
  98         if_dump = self.vapi.sw_interface_dump(name_filter_valid=True,
  99                                               name_filter='loopXYZ')
 100         self.assertEqual(len(if_dump), 0)
 101
 102         # check ping
 103         stream = self.create_icmp_stream(self.pg0, loopbacks)
 104         self.pg0.add_stream(stream)
 105         self.pg_enable_capture(self.pg_interfaces)
 106         self.pg_start()
 107         capture = self.pg0.get_capture(expected_count=len(stream))
 108
 109         self.verify_icmp(capture, self.pg0, loopbacks)
 110
 111         # delete
 112         for i in loopbacks:
 113             i.remove_vpp_config()
 114
 115         # read (check not in sw if dump, ip4 fib, ip6 fib)
 116         if_dump = self.vapi.sw_interface_dump()
 117         fib4_dump = self.vapi.ip_route_dump(0)
 118         for i in loopbacks:
 119             self.assertFalse(i.is_interface_config_in_dump(if_dump))
 120             self.assertFalse(i.is_ip4_entry_in_fib_dump(fib4_dump))
 121
 122         #  check not ping
 123         stream = self.create_icmp_stream(self.pg0, loopbacks)
 124         self.pg0.add_stream(stream)
 125         self.pg_enable_capture(self.pg_interfaces)
 126         self.pg_start()
 127         self.pg0.assert_nothing_captured()
 128
 129     def test_down(self):
 130         # create
 131         loopbacks = self.create_loopback_interfaces(20)
 132         for i in loopbacks:
 133             i.local_ip4_prefix_len = 32
 134             i.config_ip4().admin_up()
 135
 136         # disable
 137         for i in loopbacks:
 138             i.admin_down().unconfig_ip4()
 139
 140         # read (check not in sw if dump, ip4 fib, ip6 fib)
 141         if_dump = self.vapi.sw_interface_dump()
 142         fib4_dump = self.vapi.ip_route_dump(0)
 143         for i in loopbacks:
 144             self.assertTrue(i.is_interface_config_in_dump(if_dump))
 145             self.assertFalse(i.is_ip4_entry_in_fib_dump(fib4_dump))
 146
 147         #  check not ping
 148         stream = self.create_icmp_stream(self.pg0, loopbacks)
 149         self.pg0.add_stream(stream)
 150         self.pg_enable_capture(self.pg_interfaces)
 151         self.pg_start()
 152         self.pg0.assert_nothing_captured()
 153
 154
 155 class TestInterfaceDumpApiLocalOnly(VppTestCase):
 156     """test_interface_crud.TestInterfaceDumpApiLocalOnly"""
 157
 158     def test_sw_if_index_0(self):
 159         rv = self.vapi.sw_interface_dump(sw_if_index=0)
 160         self.assertEqual(rv[0].sw_if_index, 0)
 161
 162     def test_sw_if_index_twiddle0(self):
 163         rv = self.vapi.sw_interface_dump(sw_if_index=0xffffffff)
 164         self.assertEqual(rv[0].sw_if_index, 0)
 165
 166     def test_sw_if_index_1_not_existing(self):
 167         rv = self.vapi.sw_interface_dump(sw_if_index=1)
 168         self.assertEqual(len(rv), 0, 'expected no records.')
 169
 170
 171 class TestInterfaceDumpApi(VppTestCase):
 172     """test_interface_crud.TestInterfaceDumpApi"""
 173
 174     def test_sw_if_index_1(self):
 175         self.vapi.create_loopback_instance(is_specified=1,
 176                                            user_instance=10)
 177         self.vapi.create_loopback_instance(is_specified=1,
 178                                            user_instance=5)
 179
 180         # Can I get back the specified record?
 181         rv = self.vapi.sw_interface_dump(sw_if_index=1)
 182         self.assertEqual(rv[0].sw_if_index, 1, rv)
 183
 184         # verify 3 interfaces
 185         rv = self.vapi.sw_interface_dump(sw_if_index=0xffffffff)
 186         self.assertEqual(len(rv), 3, 'Expected 3 interfaces.')
 187
 188
 189 if __name__ == '__main__':
 190     unittest.main(testRunner=VppTestRunner)