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
14import unittest
15
16from scapy.layers.inet import IP, ICMP
17from scapy.layers.l2 import Ether
18
19from framework import VppTestCase, VppTestRunner
20
21
22class 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().resolve_arp()
34        except:
35            cls.tearDownClass()
36            raise
37
38    @classmethod
39    def tearDownClass(cls):
40        super(TestLoopbackInterfaceCRUD, cls).tearDownClass()
41
42    @staticmethod
43    def create_icmp_stream(src_if, dst_ifs):
44        """
45
46        :param VppInterface src_if: Packets are send to this interface,
47            using this interfaces remote host.
48        :param list dst_ifs: IPv4 ICMP requests are send to interfaces
49            addresses.
50        :return: List of generated packets.
51        """
52        pkts = []
53        for i in dst_ifs:
54            p = (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            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(name_filter_valid=True,
92                                              name_filter='loop')
93        fib4_dump = self.vapi.ip_route_dump(0)
94        for i in loopbacks:
95            self.assertTrue(i.is_interface_config_in_dump(if_dump))
96            self.assertTrue(i.is_ip4_entry_in_fib_dump(fib4_dump))
97
98        if_dump = self.vapi.sw_interface_dump(name_filter_valid=True,
99                                              name_filter='loopXYZ')
100        self.assertEqual(len(if_dump), 0)
101
102        # check ping
103        stream = self.create_icmp_stream(self.pg0, loopbacks)
104        self.pg0.add_stream(stream)
105        self.pg_enable_capture(self.pg_interfaces)
106        self.pg_start()
107        capture = self.pg0.get_capture(expected_count=len(stream))
108
109        self.verify_icmp(capture, self.pg0, loopbacks)
110
111        # delete
112        for i in loopbacks:
113            i.remove_vpp_config()
114
115        # read (check not in sw if dump, ip4 fib, ip6 fib)
116        if_dump = self.vapi.sw_interface_dump()
117        fib4_dump = self.vapi.ip_route_dump(0)
118        for i in loopbacks:
119            self.assertFalse(i.is_interface_config_in_dump(if_dump))
120            self.assertFalse(i.is_ip4_entry_in_fib_dump(fib4_dump))
121
122        #  check not ping
123        stream = self.create_icmp_stream(self.pg0, loopbacks)
124        self.pg0.add_stream(stream)
125        self.pg_enable_capture(self.pg_interfaces)
126        self.pg_start()
127        self.pg0.assert_nothing_captured()
128
129    def test_down(self):
130        # create
131        loopbacks = self.create_loopback_interfaces(20)
132        for i in loopbacks:
133            i.local_ip4_prefix_len = 32
134            i.config_ip4().admin_up()
135
136        # disable
137        for i in loopbacks:
138            i.admin_down().unconfig_ip4()
139
140        # read (check not in sw if dump, ip4 fib, ip6 fib)
141        if_dump = self.vapi.sw_interface_dump()
142        fib4_dump = self.vapi.ip_route_dump(0)
143        for i in loopbacks:
144            self.assertTrue(i.is_interface_config_in_dump(if_dump))
145            self.assertFalse(i.is_ip4_entry_in_fib_dump(fib4_dump))
146
147        #  check not ping
148        stream = self.create_icmp_stream(self.pg0, loopbacks)
149        self.pg0.add_stream(stream)
150        self.pg_enable_capture(self.pg_interfaces)
151        self.pg_start()
152        self.pg0.assert_nothing_captured()
153
154
155class TestInterfaceDumpApiLocalOnly(VppTestCase):
156    """test_interface_crud.TestInterfaceDumpApiLocalOnly"""
157
158    def test_sw_if_index_0(self):
159        rv = self.vapi.sw_interface_dump(sw_if_index=0)
160        self.assertEqual(rv[0].sw_if_index, 0)
161
162    def test_sw_if_index_twiddle0(self):
163        rv = self.vapi.sw_interface_dump(sw_if_index=0xffffffff)
164        self.assertEqual(rv[0].sw_if_index, 0)
165
166    def test_sw_if_index_1_not_existing(self):
167        rv = self.vapi.sw_interface_dump(sw_if_index=1)
168        self.assertEqual(len(rv), 0, 'expected no records.')
169
170
171class TestInterfaceDumpApi(VppTestCase):
172    """test_interface_crud.TestInterfaceDumpApi"""
173
174    def test_sw_if_index_1(self):
175        self.vapi.create_loopback_instance(is_specified=1,
176                                           user_instance=10)
177        self.vapi.create_loopback_instance(is_specified=1,
178                                           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
189if __name__ == '__main__':
190    unittest.main(testRunner=VppTestRunner)
191