test/test_lisp.py

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