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 asf.lisp import (
22 # From py_lispnetworking.lisp.py: # GNU General Public License v2.0
25 class LISP_GPE_Header(Packet):
26 name = "LISP GPE Header"
28 FlagsField("gpe_flags", None, 6, ["N", "L", "E", "V", "I", "P"]),
29 BitField("reserved", 0, 18),
30 ByteField("next_proto", 0),
35 bind_layers(UDP, LISP_GPE_Header, dport=4341)
36 bind_layers(UDP, LISP_GPE_Header, sport=4341)
37 bind_layers(LISP_GPE_Header, IP, next_proto=1)
38 bind_layers(LISP_GPE_Header, IPv6, next_proto=2)
39 bind_layers(LISP_GPE_Header, Ether, next_proto=3)
42 class ForeignAddressFactory(object):
45 net_template = "10.10.10.{}"
46 net = net_template.format(0) + "/" + str(prefix_len)
50 raise Exception("Network host address exhaustion")
52 return self.net_template.format(self.count)
55 class Driver(metaclass=abc.ABCMeta):
64 """ Basic class for data driven testing """
66 def __init__(self, test, test_cases):
67 self._test_cases = test_cases
72 return self._test_cases
78 def create_packet(self, src_if, dst_if, deid, payload=""):
86 Ether(dst=src_if.local_mac, src=src_if.remote_mac)
87 / IP(src=src_if.remote_ip4, dst=deid)
94 """testing procedure"""
98 class SimpleDriver(Driver):
99 """Implements simple test procedure"""
101 def __init__(self, test, test_cases):
102 super(SimpleDriver, self).__init__(test, test_cases)
104 def verify_capture(self, src_loc, dst_loc, capture):
106 Verify captured packet
108 :param src_loc: source locator address
109 :param dst_loc: destination locator address
110 :param capture: list of captured packets
112 self.test.assertEqual(
115 "Unexpected number of "
116 "packets! Expected 1 but {} received".format(len(capture)),
121 # assert the values match
122 self.test.assertEqual(ip_hdr.src, src_loc, "IP source address")
123 self.test.assertEqual(ip_hdr.dst, dst_loc, "IP destination address")
124 gpe_hdr = packet[LISP_GPE_Header]
125 self.test.assertEqual(gpe_hdr.next_proto, 1, "next_proto is not ipv4!")
127 self.test.assertEqual(
128 ih.src, self.test.pg0.remote_ip4, "unexpected source EID!"
130 self.test.assertEqual(ih.dst, self.test.deid_ip4, "unexpected dest EID!")
132 self.test.logger.error(ppp("Unexpected or invalid packet:", packet))
135 def configure_tc(self, tc):
136 for config_item in self.config_order:
137 for vpp_object in tc[config_item]:
138 vpp_object.add_vpp_config()
141 """Send traffic for each test case and verify that it
143 for tc in enumerate(self.test_cases):
144 self.test.logger.info("Running {}".format(tc[1]["name"]))
145 self.configure_tc(tc[1])
147 packet = self.create_packet(self.test.pg0, self.test.pg1, dest, "data")
148 self.test.pg0.add_stream(packet)
149 self.test.pg0.enable_capture()
150 self.test.pg1.enable_capture()
152 capture = self.test.pg1.get_capture(1)
154 self.test.pg1.local_ip4, self.test.pg1.remote_ip4, capture
156 self.test.pg0.assert_nothing_captured()
159 class TestLisp(VppTestCase):
160 """Basic LISP test"""
164 super(TestLisp, cls).setUpClass()
165 cls.faf = ForeignAddressFactory()
166 cls.create_pg_interfaces(range(2)) # create pg0 and pg1
167 for i in cls.pg_interfaces:
168 i.admin_up() # put the interface upsrc_if
169 i.config_ip4() # configure IPv4 address on the interface
170 i.resolve_arp() # resolve ARP, so that we know VPP MAC
173 def tearDownClass(cls):
174 super(TestLisp, cls).tearDownClass()
177 super(TestLisp, self).setUp()
178 self.vapi.lisp_enable_disable(is_enable=1)
180 def test_lisp_basic_encap(self):
181 """Test case for basic encapsulation"""
183 self.deid_ip4_net = self.faf.net
184 self.deid_ip4 = self.faf.get_ip4()
185 self.seid_ip4 = "{!s}/{!s}".format(self.pg0.local_ip4, 32)
186 self.rloc_ip4 = self.pg1.remote_ip4
190 "name": "basic ip4 over ip4",
191 "locator-sets": [VppLispLocatorSet(self, "ls-4o4")],
192 "locators": [VppLispLocator(self, self.pg1.sw_if_index, "ls-4o4")],
193 "local-mappings": [VppLocalMapping(self, self.seid_ip4, "ls-4o4")],
196 self, self.deid_ip4_net, [LispRemoteLocator(self.rloc_ip4)]
200 VppLispAdjacency(self, self.seid_ip4, self.deid_ip4_net)
204 self.test_driver = SimpleDriver(self, test_cases)
205 self.test_driver.run(self.deid_ip4)
208 class TestLispUT(VppTestCase):
213 super(TestLispUT, cls).setUpClass()
216 def tearDownClass(cls):
217 super(TestLispUT, cls).tearDownClass()
220 """LISP Unit Tests"""
221 error = self.vapi.cli("test lisp cp")
224 self.logger.critical(error)
225 self.assertNotIn("Failed", error)
228 if __name__ == "__main__":
229 unittest.main(testRunner=VppTestRunner)