test/test_lisp.py

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