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