src/plugins/lisp/test/test_lisp.py

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