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
  20 from asfframework import VppTestRunner
  21
  22
  23 class TestLoopbackInterfaceCRUD(VppTestCase):
  24     """CRUD Loopback"""
  25
  26     @classmethod
  27     def setUpClass(cls):
  28         super(TestLoopbackInterfaceCRUD, cls).setUpClass()
  29         try:
  30             cls.create_pg_interfaces(range(1))
  31             for i in cls.pg_interfaces:
  32                 i.config_ip4().resolve_arp()
  33         except:
  34             cls.tearDownClass()
  35             raise
  36
  37     @classmethod
  38     def tearDownClass(cls):
  39         super(TestLoopbackInterfaceCRUD, cls).tearDownClass()
  40
  41     @staticmethod
  42     def create_icmp_stream(src_if, dst_ifs):
  43         """
  44
  45         :param VppInterface src_if: Packets are send to this interface,
  46             using this interfaces remote host.
  47         :param list dst_ifs: IPv4 ICMP requests are send to interfaces
  48             addresses.
  49         :return: List of generated packets.
  50         """
  51         pkts = []
  52         for i in dst_ifs:
  53             p = (
  54                 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             )
  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().admin_up()
  90
  91         # read (check sw if dump, ip4 fib, ip6 fib)
  92         if_dump = self.vapi.sw_interface_dump(
  93             name_filter_valid=True, name_filter="loop"
  94         )
  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(
 101             name_filter_valid=True, name_filter="loopXYZ"
 102         )
 103         self.assertEqual(len(if_dump), 0)
 104
 105         # check ping
 106         stream = self.create_icmp_stream(self.pg0, loopbacks)
 107         self.pg0.add_stream(stream)
 108         self.pg_enable_capture(self.pg_interfaces)
 109         self.pg_start()
 110         capture = self.pg0.get_capture(expected_count=len(stream))
 111
 112         self.verify_icmp(capture, self.pg0, loopbacks)
 113
 114         # delete
 115         for i in loopbacks:
 116             i.remove_vpp_config()
 117
 118         # read (check not in sw if dump, ip4 fib, ip6 fib)
 119         if_dump = self.vapi.sw_interface_dump()
 120         fib4_dump = self.vapi.ip_route_dump(0)
 121         for i in loopbacks:
 122             self.assertFalse(i.is_interface_config_in_dump(if_dump))
 123             self.assertFalse(i.is_ip4_entry_in_fib_dump(fib4_dump))
 124
 125         #  check not ping
 126         stream = self.create_icmp_stream(self.pg0, loopbacks)
 127         self.pg0.add_stream(stream)
 128         self.pg_enable_capture(self.pg_interfaces)
 129         self.pg_start()
 130         self.pg0.assert_nothing_captured()
 131
 132     def test_down(self):
 133         # create
 134         loopbacks = self.create_loopback_interfaces(20)
 135         for i in loopbacks:
 136             i.local_ip4_prefix_len = 32
 137             i.config_ip4().admin_up()
 138
 139         # disable
 140         for i in loopbacks:
 141             i.admin_down().unconfig_ip4()
 142
 143         # read (check not in sw if dump, ip4 fib, ip6 fib)
 144         if_dump = self.vapi.sw_interface_dump()
 145         fib4_dump = self.vapi.ip_route_dump(0)
 146         for i in loopbacks:
 147             self.assertTrue(i.is_interface_config_in_dump(if_dump))
 148             self.assertFalse(i.is_ip4_entry_in_fib_dump(fib4_dump))
 149
 150         #  check not ping
 151         stream = self.create_icmp_stream(self.pg0, loopbacks)
 152         self.pg0.add_stream(stream)
 153         self.pg_enable_capture(self.pg_interfaces)
 154         self.pg_start()
 155         self.pg0.assert_nothing_captured()
 156
 157
 158 class TestInterfaceDumpApiLocalOnly(VppTestCase):
 159     """test_interface_crud.TestInterfaceDumpApiLocalOnly"""
 160
 161     def test_sw_if_index_0(self):
 162         rv = self.vapi.sw_interface_dump(sw_if_index=0)
 163         self.assertEqual(rv[0].sw_if_index, 0)
 164
 165     def test_sw_if_index_twiddle0(self):
 166         rv = self.vapi.sw_interface_dump(sw_if_index=0xFFFFFFFF)
 167         self.assertEqual(rv[0].sw_if_index, 0)
 168
 169     def test_sw_if_index_1_not_existing(self):
 170         rv = self.vapi.sw_interface_dump(sw_if_index=1)
 171         self.assertEqual(len(rv), 0, "expected no records.")
 172
 173
 174 class TestInterfaceDumpApi(VppTestCase):
 175     """test_interface_crud.TestInterfaceDumpApi"""
 176
 177     def test_sw_if_index_1(self):
 178         self.vapi.create_loopback_instance(is_specified=1, user_instance=10)
 179         self.vapi.create_loopback_instance(is_specified=1, user_instance=5)
 180
 181         # Can I get back the specified record?
 182         rv = self.vapi.sw_interface_dump(sw_if_index=1)
 183         self.assertEqual(rv[0].sw_if_index, 1, rv)
 184
 185         # verify 3 interfaces
 186         rv = self.vapi.sw_interface_dump(sw_if_index=0xFFFFFFFF)
 187         self.assertEqual(len(rv), 3, "Expected 3 interfaces.")
 188
 189
 190 if __name__ == "__main__":
 191     unittest.main(testRunner=VppTestRunner)