test/test_lisp.py

   1 #!/usr/bin/env python3
   2
   3 import abc
   4 import unittest
   5
   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
  10
  11 from framework import VppTestCase, VppTestRunner
  12 from asf.lisp import (
  13     VppLocalMapping,
  14     VppLispAdjacency,
  15     VppLispLocator,
  16     VppLispLocatorSet,
  17     VppRemoteMapping,
  18     LispRemoteLocator,
  19 )
  20 from util import ppp
  21
  22 # From py_lispnetworking.lisp.py:  # GNU General Public License v2.0
  23
  24
  25 class LISP_GPE_Header(Packet):
  26     name = "LISP GPE Header"
  27     fields_desc = [
  28         FlagsField("gpe_flags", None, 6, ["N", "L", "E", "V", "I", "P"]),
  29         BitField("reserved", 0, 18),
  30         ByteField("next_proto", 0),
  31         IntField("iid", 0),
  32     ]
  33
  34
  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)
  40
  41
  42 class ForeignAddressFactory(object):
  43     count = 0
  44     prefix_len = 24
  45     net_template = "10.10.10.{}"
  46     net = net_template.format(0) + "/" + str(prefix_len)
  47
  48     def get_ip4(self):
  49         if self.count > 255:
  50             raise Exception("Network host address exhaustion")
  51         self.count += 1
  52         return self.net_template.format(self.count)
  53
  54
  55 class Driver(metaclass=abc.ABCMeta):
  56     config_order = [
  57         "locator-sets",
  58         "locators",
  59         "local-mappings",
  60         "remote-mappings",
  61         "adjacencies",
  62     ]
  63
  64     """ Basic class for data driven testing """
  65
  66     def __init__(self, test, test_cases):
  67         self._test_cases = test_cases
  68         self._test = test
  69
  70     @property
  71     def test_cases(self):
  72         return self._test_cases
  73
  74     @property
  75     def test(self):
  76         return self._test
  77
  78     def create_packet(self, src_if, dst_if, deid, payload=""):
  79         """
  80         Create IPv4 packet
  81
  82         param: src_if
  83         param: dst_if
  84         """
  85         packet = (
  86             Ether(dst=src_if.local_mac, src=src_if.remote_mac)
  87             / IP(src=src_if.remote_ip4, dst=deid)
  88             / Raw(payload)
  89         )
  90         return packet
  91
  92     @abc.abstractmethod
  93     def run(self):
  94         """testing procedure"""
  95         pass
  96
  97
  98 class SimpleDriver(Driver):
  99     """Implements simple test procedure"""
 100
 101     def __init__(self, test, test_cases):
 102         super(SimpleDriver, self).__init__(test, test_cases)
 103
 104     def verify_capture(self, src_loc, dst_loc, capture):
 105         """
 106         Verify captured packet
 107
 108         :param src_loc: source locator address
 109         :param dst_loc: destination locator address
 110         :param capture: list of captured packets
 111         """
 112         self.test.assertEqual(
 113             len(capture),
 114             1,
 115             "Unexpected number of "
 116             "packets! Expected 1 but {} received".format(len(capture)),
 117         )
 118         packet = capture[0]
 119         try:
 120             ip_hdr = packet[IP]
 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!")
 126             ih = gpe_hdr[IP]
 127             self.test.assertEqual(
 128                 ih.src, self.test.pg0.remote_ip4, "unexpected source EID!"
 129             )
 130             self.test.assertEqual(ih.dst, self.test.deid_ip4, "unexpected dest EID!")
 131         except:
 132             self.test.logger.error(ppp("Unexpected or invalid packet:", packet))
 133             raise
 134
 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()
 139
 140     def run(self, dest):
 141         """Send traffic for each test case and verify that it
 142         is encapsulated"""
 143         for tc in enumerate(self.test_cases):
 144             self.test.logger.info("Running {}".format(tc[1]["name"]))
 145             self.configure_tc(tc[1])
 146
 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()
 151             self.test.pg_start()
 152             capture = self.test.pg1.get_capture(1)
 153             self.verify_capture(
 154                 self.test.pg1.local_ip4, self.test.pg1.remote_ip4, capture
 155             )
 156             self.test.pg0.assert_nothing_captured()
 157
 158
 159 class TestLisp(VppTestCase):
 160     """Basic LISP test"""
 161
 162     @classmethod
 163     def setUpClass(cls):
 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
 171
 172     @classmethod
 173     def tearDownClass(cls):
 174         super(TestLisp, cls).tearDownClass()
 175
 176     def setUp(self):
 177         super(TestLisp, self).setUp()
 178         self.vapi.lisp_enable_disable(is_enable=1)
 179
 180     def test_lisp_basic_encap(self):
 181         """Test case for basic encapsulation"""
 182
 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
 187
 188         test_cases = [
 189             {
 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")],
 194                 "remote-mappings": [
 195                     VppRemoteMapping(
 196                         self, self.deid_ip4_net, [LispRemoteLocator(self.rloc_ip4)]
 197                     )
 198                 ],
 199                 "adjacencies": [
 200                     VppLispAdjacency(self, self.seid_ip4, self.deid_ip4_net)
 201                 ],
 202             }
 203         ]
 204         self.test_driver = SimpleDriver(self, test_cases)
 205         self.test_driver.run(self.deid_ip4)
 206
 207
 208 class TestLispUT(VppTestCase):
 209     """Lisp UT"""
 210
 211     @classmethod
 212     def setUpClass(cls):
 213         super(TestLispUT, cls).setUpClass()
 214
 215     @classmethod
 216     def tearDownClass(cls):
 217         super(TestLispUT, cls).tearDownClass()
 218
 219     def test_fib(self):
 220         """LISP Unit Tests"""
 221         error = self.vapi.cli("test lisp cp")
 222
 223         if error:
 224             self.logger.critical(error)
 225         self.assertNotIn("Failed", error)
 226
 227
 228 if __name__ == "__main__":
 229     unittest.main(testRunner=VppTestRunner)