lb: remove api boilerplate
[vpp.git] / 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)