test/test_interface_crud.py

   1 #!/usr/bin/env python
   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()
  94         fib4_dump = self.vapi.ip_route_dump(0)
  95         for i in loopbacks:
  96             self.assertTrue(i.is_interface_config_in_dump(if_dump))
  97             self.assertTrue(i.is_ip4_entry_in_fib_dump(fib4_dump))
  98
  99         # check ping
 100         stream = self.create_icmp_stream(self.pg0, loopbacks)
 101         self.pg0.add_stream(stream)
 102         self.pg_enable_capture(self.pg_interfaces)
 103         self.pg_start()
 104         capture = self.pg0.get_capture(expected_count=len(stream))
 105
 106         self.verify_icmp(capture, self.pg0, loopbacks)
 107
 108         # delete
 109         for i in loopbacks:
 110             i.remove_vpp_config()
 111
 112         # read (check not in sw if dump, ip4 fib, ip6 fib)
 113         if_dump = self.vapi.sw_interface_dump()
 114         fib4_dump = self.vapi.ip_route_dump(0)
 115         for i in loopbacks:
 116             self.assertFalse(i.is_interface_config_in_dump(if_dump))
 117             self.assertFalse(i.is_ip4_entry_in_fib_dump(fib4_dump))
 118
 119         #  check not ping
 120         stream = self.create_icmp_stream(self.pg0, loopbacks)
 121         self.pg0.add_stream(stream)
 122         self.pg_enable_capture(self.pg_interfaces)
 123         self.pg_start()
 124         self.pg0.assert_nothing_captured()
 125
 126     def test_down(self):
 127         # create
 128         loopbacks = self.create_loopback_interfaces(20)
 129         for i in loopbacks:
 130             i.local_ip4_prefix_len = 32
 131             i.config_ip4()
 132             i.admin_up()
 133
 134         # disable
 135         for i in loopbacks:
 136             i.admin_down()
 137             i.unconfig_ip4()
 138
 139         # read (check not in sw if dump, ip4 fib, ip6 fib)
 140         if_dump = self.vapi.sw_interface_dump()
 141         fib4_dump = self.vapi.ip_route_dump(0)
 142         for i in loopbacks:
 143             self.assertTrue(i.is_interface_config_in_dump(if_dump))
 144             self.assertFalse(i.is_ip4_entry_in_fib_dump(fib4_dump))
 145
 146         #  check not ping
 147         stream = self.create_icmp_stream(self.pg0, loopbacks)
 148         self.pg0.add_stream(stream)
 149         self.pg_enable_capture(self.pg_interfaces)
 150         self.pg_start()
 151         self.pg0.assert_nothing_captured()
 152
 153
 154 class TestInterfaceDumpApiLocalOnly(VppTestCase):
 155     """test_interface_crud.TestInterfaceDumpApiLocalOnly"""
 156
 157     def test_sw_if_index_0(self):
 158         rv = self.vapi.sw_interface_dump(sw_if_index=0)
 159         self.assertEqual(rv[0].sw_if_index, 0)
 160
 161     def test_sw_if_index_twiddle0(self):
 162         rv = self.vapi.sw_interface_dump(sw_if_index=0xffffffff)
 163         self.assertEqual(rv[0].sw_if_index, 0)
 164
 165     def test_sw_if_index_1_not_existing(self):
 166         rv = self.vapi.sw_interface_dump(sw_if_index=1)
 167         self.assertEqual(len(rv), 0, 'expected no records.')
 168
 169
 170 class TestInterfaceDumpApi(VppTestCase):
 171     """test_interface_crud.TestInterfaceDumpApi"""
 172
 173     def test_sw_if_index_1(self):
 174         self.vapi.create_loopback_instance(is_specified=1,
 175                                            user_instance=10)
 176         self.vapi.create_loopback_instance(is_specified=1,
 177                                            user_instance=5)
 178
 179         # Can I get back the specified record?
 180         rv = self.vapi.sw_interface_dump(sw_if_index=1)
 181         self.assertEqual(rv[0].sw_if_index, 1, rv)
 182
 183         # verify 3 interfaces
 184         rv = self.vapi.sw_interface_dump(sw_if_index=0xffffffff)
 185         self.assertEqual(len(rv), 3, 'Expected 3 interfaces.')
 186
 187 if __name__ == '__main__':
 188     unittest.main(testRunner=VppTestRunner)