lb: remove api boilerplate
[vpp.git] / test / test_lisp.py
1 #!/usr/bin/env python
2
3 import abc
4 import six
5 import unittest
6
7 from scapy.fields import BitField, ByteField, FlagsField, IntField
8 from scapy.packet import bind_layers, Packet, Raw
9 from scapy.layers.inet import IP, UDP, Ether
10 from scapy.layers.inet6 import IPv6
11
12 from framework import VppTestCase, VppTestRunner
13 from lisp import VppLocalMapping, VppLispAdjacency, VppLispLocator, \
14     VppLispLocatorSet, VppRemoteMapping
15 from util import ppp, ForeignAddressFactory
16
17 # From py_lispnetworking.lisp.py:  # GNU General Public License v2.0
18
19
20 class LISP_GPE_Header(Packet):
21     name = "LISP GPE Header"
22     fields_desc = [
23         FlagsField("gpe_flags", None, 6, ["N", "L", "E", "V", "I", "P"]),
24         BitField("reserved", 0, 18),
25         ByteField("next_proto", 0),
26         IntField("iid", 0),
27     ]
28 bind_layers(UDP, LISP_GPE_Header, dport=4341)
29 bind_layers(UDP, LISP_GPE_Header, sport=4341)
30 bind_layers(LISP_GPE_Header, IP, next_proto=1)
31 bind_layers(LISP_GPE_Header, IPv6, next_proto=2)
32 bind_layers(LISP_GPE_Header, Ether, next_proto=3)
33
34
35 @six.add_metaclass(abc.ABCMeta)
36 class Driver(object):
37
38     config_order = ['locator-sets',
39                     'locators',
40                     'local-mappings',
41                     'remote-mappings',
42                     'adjacencies']
43
44     """ Basic class for data driven testing """
45     def __init__(self, test, test_cases):
46         self._test_cases = test_cases
47         self._test = test
48
49     @property
50     def test_cases(self):
51         return self._test_cases
52
53     @property
54     def test(self):
55         return self._test
56
57     def create_packet(self, src_if, dst_if, deid, payload=''):
58         """
59         Create IPv4 packet
60
61         param: src_if
62         param: dst_if
63         """
64         packet = (Ether(dst=src_if.local_mac, src=src_if.remote_mac) /
65                   IP(src=src_if.remote_ip4, dst=deid) /
66                   Raw(payload))
67         return packet
68
69     @abc.abstractmethod
70     def run(self):
71         """ testing procedure """
72         pass
73
74
75 class SimpleDriver(Driver):
76     """ Implements simple test procedure """
77     def __init__(self, test, test_cases):
78         super(SimpleDriver, self).__init__(test, test_cases)
79
80     def verify_capture(self, src_loc, dst_loc, capture):
81         """
82         Verify captured packet
83
84         :param src_loc: source locator address
85         :param dst_loc: destination locator address
86         :param capture: list of captured packets
87         """
88         self.test.assertEqual(len(capture), 1, "Unexpected number of "
89                               "packets! Expected 1 but {} received"
90                               .format(len(capture)))
91         packet = capture[0]
92         try:
93             ip_hdr = packet[IP]
94             # assert the values match
95             self.test.assertEqual(ip_hdr.src, src_loc, "IP source address")
96             self.test.assertEqual(ip_hdr.dst, dst_loc,
97                                   "IP destination address")
98             gpe_hdr = packet[LISP_GPE_Header]
99             self.test.assertEqual(gpe_hdr.next_proto, 1,
100                                   "next_proto is not ipv4!")
101             ih = gpe_hdr[IP]
102             self.test.assertEqual(ih.src, self.test.pg0.remote_ip4,
103                                   "unexpected source EID!")
104             self.test.assertEqual(ih.dst, self.test.deid_ip4,
105                                   "unexpected dest EID!")
106         except:
107             self.test.logger.error(ppp("Unexpected or invalid packet:",
108                                    packet))
109             raise
110
111     def configure_tc(self, tc):
112         for config_item in self.config_order:
113             for vpp_object in tc[config_item]:
114                 vpp_object.add_vpp_config()
115
116     def run(self, dest):
117         """ Send traffic for each test case and verify that it
118             is encapsulated """
119         for tc in enumerate(self.test_cases):
120             self.test.logger.info('Running {}'.format(tc[1]['name']))
121             self.configure_tc(tc[1])
122
123             packet = self.create_packet(self.test.pg0, self.test.pg1, dest,
124                                         'data')
125             self.test.pg0.add_stream(packet)
126             self.test.pg0.enable_capture()
127             self.test.pg1.enable_capture()
128             self.test.pg_start()
129             capture = self.test.pg1.get_capture(1)
130             self.verify_capture(self.test.pg1.local_ip4,
131                                 self.test.pg1.remote_ip4, capture)
132             self.test.pg0.assert_nothing_captured()
133
134
135 class TestLisp(VppTestCase):
136     """ Basic LISP test """
137
138     @classmethod
139     def setUpClass(cls):
140         super(TestLisp, cls).setUpClass()
141         cls.faf = ForeignAddressFactory()
142         cls.create_pg_interfaces(range(2))  # create pg0 and pg1
143         for i in cls.pg_interfaces:
144             i.admin_up()  # put the interface upsrc_if
145             i.config_ip4()  # configure IPv4 address on the interface
146             i.resolve_arp()  # resolve ARP, so that we know VPP MAC
147
148     @classmethod
149     def tearDownClass(cls):
150         super(TestLisp, cls).tearDownClass()
151
152     def setUp(self):
153         super(TestLisp, self).setUp()
154         self.vapi.lisp_enable_disable(is_enabled=1)
155
156     def test_lisp_basic_encap(self):
157         """Test case for basic encapsulation"""
158
159         self.deid_ip4_net = self.faf.net
160         self.deid_ip4 = self.faf.get_ip4()
161         self.seid_ip4 = '{!s}/{!s}'.format(self.pg0.local_ip4, 32)
162         self.rloc_ip4 = self.pg1.remote_ip4n
163
164         test_cases = [
165             {
166                 'name': 'basic ip4 over ip4',
167                 'locator-sets': [VppLispLocatorSet(self, b'ls-4o4')],
168                 'locators': [
169                     VppLispLocator(self, self.pg1.sw_if_index, b'ls-4o4')
170                 ],
171                 'local-mappings': [
172                     VppLocalMapping(self, self.seid_ip4, b'ls-4o4')
173                 ],
174                 'remote-mappings': [
175                     VppRemoteMapping(self, self.deid_ip4_net,
176                                      [{
177                                          "is_ip4": 1,
178                                          "priority": 1,
179                                          "weight": 1,
180                                          "addr": self.rloc_ip4
181                                      }])
182                 ],
183                 'adjacencies': [
184                     VppLispAdjacency(self, self.seid_ip4, self.deid_ip4_net)
185                 ]
186             }
187         ]
188         self.test_driver = SimpleDriver(self, test_cases)
189         self.test_driver.run(self.deid_ip4)
190
191
192 if __name__ == '__main__':
193     unittest.main(testRunner=VppTestRunner)