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