6 from scapy.fields import BitField, ByteField, FlagsField, IntField
7 from scapy.packet import bind_layers, Packet, Raw
8 from scapy.layers.inet import IP, UDP, Ether
9 from scapy.layers.inet6 import IPv6
11 from framework import VppTestCase, VppTestRunner
12 from lisp import VppLocalMapping, VppLispAdjacency, VppLispLocator, \
13 VppLispLocatorSet, VppRemoteMapping, LispRemoteLocator
16 # From py_lispnetworking.lisp.py: # GNU General Public License v2.0
19 class LISP_GPE_Header(Packet):
20 name = "LISP GPE Header"
22 FlagsField("gpe_flags", None, 6, ["N", "L", "E", "V", "I", "P"]),
23 BitField("reserved", 0, 18),
24 ByteField("next_proto", 0),
27 bind_layers(UDP, LISP_GPE_Header, dport=4341)
28 bind_layers(UDP, LISP_GPE_Header, sport=4341)
29 bind_layers(LISP_GPE_Header, IP, next_proto=1)
30 bind_layers(LISP_GPE_Header, IPv6, next_proto=2)
31 bind_layers(LISP_GPE_Header, Ether, next_proto=3)
34 class ForeignAddressFactory(object):
37 net_template = '10.10.10.{}'
38 net = net_template.format(0) + '/' + str(prefix_len)
42 raise Exception("Network host address exhaustion")
44 return self.net_template.format(self.count)
47 class Driver(metaclass=abc.ABCMeta):
49 config_order = ['locator-sets',
55 """ Basic class for data driven testing """
56 def __init__(self, test, test_cases):
57 self._test_cases = test_cases
62 return self._test_cases
68 def create_packet(self, src_if, dst_if, deid, payload=''):
75 packet = (Ether(dst=src_if.local_mac, src=src_if.remote_mac) /
76 IP(src=src_if.remote_ip4, dst=deid) /
82 """ testing procedure """
86 class SimpleDriver(Driver):
87 """ Implements simple test procedure """
88 def __init__(self, test, test_cases):
89 super(SimpleDriver, self).__init__(test, test_cases)
91 def verify_capture(self, src_loc, dst_loc, capture):
93 Verify captured packet
95 :param src_loc: source locator address
96 :param dst_loc: destination locator address
97 :param capture: list of captured packets
99 self.test.assertEqual(len(capture), 1, "Unexpected number of "
100 "packets! Expected 1 but {} received"
101 .format(len(capture)))
105 # assert the values match
106 self.test.assertEqual(ip_hdr.src, src_loc, "IP source address")
107 self.test.assertEqual(ip_hdr.dst, dst_loc,
108 "IP destination address")
109 gpe_hdr = packet[LISP_GPE_Header]
110 self.test.assertEqual(gpe_hdr.next_proto, 1,
111 "next_proto is not ipv4!")
113 self.test.assertEqual(ih.src, self.test.pg0.remote_ip4,
114 "unexpected source EID!")
115 self.test.assertEqual(ih.dst, self.test.deid_ip4,
116 "unexpected dest EID!")
118 self.test.logger.error(ppp("Unexpected or invalid packet:",
122 def configure_tc(self, tc):
123 for config_item in self.config_order:
124 for vpp_object in tc[config_item]:
125 vpp_object.add_vpp_config()
128 """ Send traffic for each test case and verify that it
130 for tc in enumerate(self.test_cases):
131 self.test.logger.info('Running {}'.format(tc[1]['name']))
132 self.configure_tc(tc[1])
134 packet = self.create_packet(self.test.pg0, self.test.pg1, dest,
136 self.test.pg0.add_stream(packet)
137 self.test.pg0.enable_capture()
138 self.test.pg1.enable_capture()
140 capture = self.test.pg1.get_capture(1)
141 self.verify_capture(self.test.pg1.local_ip4,
142 self.test.pg1.remote_ip4, capture)
143 self.test.pg0.assert_nothing_captured()
146 class TestLisp(VppTestCase):
147 """ Basic LISP test """
151 super(TestLisp, cls).setUpClass()
152 cls.faf = ForeignAddressFactory()
153 cls.create_pg_interfaces(range(2)) # create pg0 and pg1
154 for i in cls.pg_interfaces:
155 i.admin_up() # put the interface upsrc_if
156 i.config_ip4() # configure IPv4 address on the interface
157 i.resolve_arp() # resolve ARP, so that we know VPP MAC
160 def tearDownClass(cls):
161 super(TestLisp, cls).tearDownClass()
164 super(TestLisp, self).setUp()
165 self.vapi.lisp_enable_disable(is_enable=1)
167 def test_lisp_basic_encap(self):
168 """Test case for basic encapsulation"""
170 self.deid_ip4_net = self.faf.net
171 self.deid_ip4 = self.faf.get_ip4()
172 self.seid_ip4 = '{!s}/{!s}'.format(self.pg0.local_ip4, 32)
173 self.rloc_ip4 = self.pg1.remote_ip4
177 'name': 'basic ip4 over ip4',
178 'locator-sets': [VppLispLocatorSet(self, 'ls-4o4')],
180 VppLispLocator(self, self.pg1.sw_if_index, 'ls-4o4')
183 VppLocalMapping(self, self.seid_ip4, 'ls-4o4')
186 VppRemoteMapping(self, self.deid_ip4_net,
187 [LispRemoteLocator(self.rloc_ip4)])
190 VppLispAdjacency(self, self.seid_ip4, self.deid_ip4_net)
194 self.test_driver = SimpleDriver(self, test_cases)
195 self.test_driver.run(self.deid_ip4)
198 class TestLispUT(VppTestCase):
203 super(TestLispUT, cls).setUpClass()
206 def tearDownClass(cls):
207 super(TestLispUT, cls).tearDownClass()
210 """ LISP Unit Tests """
211 error = self.vapi.cli("test lisp cp")
214 self.logger.critical(error)
215 self.assertNotIn("Failed", error)
218 if __name__ == '__main__':
219 unittest.main(testRunner=VppTestRunner)