test/test_lisp.py

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