2 """CRUD tests of APIs (Create, Read, Update, Delete) HLD:
4 - interface up/down/add/delete - interface type:
10 - tuntap (root privileges needed)
16 from scapy.layers.inet import IP, ICMP
17 from scapy.layers.l2 import Ether
19 from framework import VppTestCase
20 from asfframework import VppTestRunner
23 class TestLoopbackInterfaceCRUD(VppTestCase):
28 super(TestLoopbackInterfaceCRUD, cls).setUpClass()
30 cls.create_pg_interfaces(range(1))
31 for i in cls.pg_interfaces:
32 i.config_ip4().resolve_arp()
38 def tearDownClass(cls):
39 super(TestLoopbackInterfaceCRUD, cls).tearDownClass()
42 def create_icmp_stream(src_if, dst_ifs):
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
49 :return: List of generated packets.
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")
61 def verify_icmp(self, capture, request_src_if, dst_ifs):
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.
76 info = (ip.src, ip.dst, icmp.type, icmp.id)
77 rcvd_icmp_pkts.append(info)
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)
86 loopbacks = self.create_loopback_interfaces(20)
88 i.local_ip4_prefix_len = 32
89 i.config_ip4().admin_up()
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"
95 fib4_dump = self.vapi.ip_route_dump(0)
97 self.assertTrue(i.is_interface_config_in_dump(if_dump))
98 self.assertTrue(i.is_ip4_entry_in_fib_dump(fib4_dump))
100 if_dump = self.vapi.sw_interface_dump(
101 name_filter_valid=True, name_filter="loopXYZ"
103 self.assertEqual(len(if_dump), 0)
106 stream = self.create_icmp_stream(self.pg0, loopbacks)
107 self.pg0.add_stream(stream)
108 self.pg_enable_capture(self.pg_interfaces)
110 capture = self.pg0.get_capture(expected_count=len(stream))
112 self.verify_icmp(capture, self.pg0, loopbacks)
116 i.remove_vpp_config()
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)
122 self.assertFalse(i.is_interface_config_in_dump(if_dump))
123 self.assertFalse(i.is_ip4_entry_in_fib_dump(fib4_dump))
126 stream = self.create_icmp_stream(self.pg0, loopbacks)
127 self.pg0.add_stream(stream)
128 self.pg_enable_capture(self.pg_interfaces)
130 self.pg0.assert_nothing_captured()
134 loopbacks = self.create_loopback_interfaces(20)
136 i.local_ip4_prefix_len = 32
137 i.config_ip4().admin_up()
141 i.admin_down().unconfig_ip4()
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)
147 self.assertTrue(i.is_interface_config_in_dump(if_dump))
148 self.assertFalse(i.is_ip4_entry_in_fib_dump(fib4_dump))
151 stream = self.create_icmp_stream(self.pg0, loopbacks)
152 self.pg0.add_stream(stream)
153 self.pg_enable_capture(self.pg_interfaces)
155 self.pg0.assert_nothing_captured()
158 class TestInterfaceDumpApiLocalOnly(VppTestCase):
159 """test_interface_crud.TestInterfaceDumpApiLocalOnly"""
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)
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)
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.")
174 class TestInterfaceDumpApi(VppTestCase):
175 """test_interface_crud.TestInterfaceDumpApi"""
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)
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)
185 # verify 3 interfaces
186 rv = self.vapi.sw_interface_dump(sw_if_index=0xFFFFFFFF)
187 self.assertEqual(len(rv), 3, "Expected 3 interfaces.")
190 if __name__ == "__main__":
191 unittest.main(testRunner=VppTestRunner)