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