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     @classmethod
  26     def setUpClass(cls):
  27         super(TestLoopbackInterfaceCRUD, cls).setUpClass()
  28         try:
  29             cls.create_pg_interfaces(range(1))
  30             for i in cls.pg_interfaces:
  31                 i.config_ip4().resolve_arp()
  32         except:
  33             cls.tearDownClass()
  34             raise
  35
  36     @classmethod
  37     def tearDownClass(cls):
  38         super(TestLoopbackInterfaceCRUD, cls).tearDownClass()
  39
  40     @staticmethod
  41     def create_icmp_stream(src_if, dst_ifs):
  42         """
  43
  44         :param VppInterface src_if: Packets are send to this interface,
  45             using this interfaces remote host.
  46         :param list dst_ifs: IPv4 ICMP requests are send to interfaces
  47             addresses.
  48         :return: List of generated packets.
  49         """
  50         pkts = []
  51         for i in dst_ifs:
  52             p = (
  53                 Ether(dst=src_if.local_mac, src=src_if.remote_mac)
  54                 / IP(src=src_if.remote_ip4, dst=i.local_ip4)
  55                 / ICMP(id=i.sw_if_index, type="echo-request")
  56             )
  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(
  92             name_filter_valid=True, name_filter="loop"
  93         )
  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         if_dump = self.vapi.sw_interface_dump(
 100             name_filter_valid=True, name_filter="loopXYZ"
 101         )
 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().admin_up()
 137
 138         # disable
 139         for i in loopbacks:
 140             i.admin_down().unconfig_ip4()
 141
 142         # read (check not in sw if dump, ip4 fib, ip6 fib)
 143         if_dump = self.vapi.sw_interface_dump()
 144         fib4_dump = self.vapi.ip_route_dump(0)
 145         for i in loopbacks:
 146             self.assertTrue(i.is_interface_config_in_dump(if_dump))
 147             self.assertFalse(i.is_ip4_entry_in_fib_dump(fib4_dump))
 148
 149         #  check not ping
 150         stream = self.create_icmp_stream(self.pg0, loopbacks)
 151         self.pg0.add_stream(stream)
 152         self.pg_enable_capture(self.pg_interfaces)
 153         self.pg_start()
 154         self.pg0.assert_nothing_captured()
 155
 156
 157 class TestInterfaceDumpApiLocalOnly(VppTestCase):
 158     """test_interface_crud.TestInterfaceDumpApiLocalOnly"""
 159
 160     def test_sw_if_index_0(self):
 161         rv = self.vapi.sw_interface_dump(sw_if_index=0)
 162         self.assertEqual(rv[0].sw_if_index, 0)
 163
 164     def test_sw_if_index_twiddle0(self):
 165         rv = self.vapi.sw_interface_dump(sw_if_index=0xFFFFFFFF)
 166         self.assertEqual(rv[0].sw_if_index, 0)
 167
 168     def test_sw_if_index_1_not_existing(self):
 169         rv = self.vapi.sw_interface_dump(sw_if_index=1)
 170         self.assertEqual(len(rv), 0, "expected no records.")
 171
 172
 173 class TestInterfaceDumpApi(VppTestCase):
 174     """test_interface_crud.TestInterfaceDumpApi"""
 175
 176     def test_sw_if_index_1(self):
 177         self.vapi.create_loopback_instance(is_specified=1, user_instance=10)
 178         self.vapi.create_loopback_instance(is_specified=1, 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)