Add basic 4o4 LISP unit test

Change-Id: I2d812153d7afe7980346382b525af89b3c47e796
Signed-off-by: Filip Tehlar <ftehlar@cisco.com>

1 files changed, 171 insertions, 0 deletions

diff --git a/test/test_lisp.py b/test/test_lisp.py
new file mode 100644
index 00000000000..a896698c7b4
--- /dev/null
+++ b/test/test_lisp.py
@@ -0,0 +1,171 @@
+#!/usr/bin/env python
+import unittest
+
+from scapy.packet import Raw
+from scapy.layers.inet import IP, UDP, Ether
+from py_lispnetworking.lisp import LISP_GPE_Header
+
+from util import ppp, ForeignAddressFactory
+from framework import VppTestCase, VppTestRunner
+from lisp import *
+
+
+class Driver(object):
+
+    config_order = ['locator-sets',
+                    'locators',
+                    'local-mappings',
+                    'remote-mappings',
+                    'adjacencies']
+
+    """ Basic class for data driven testing """
+    def __init__(self, test, test_cases):
+        self._test_cases = test_cases
+        self._test = test
+
+    @property
+    def test_cases(self):
+        return self._test_cases
+
+    @property
+    def test(self):
+        return self._test
+
+    def create_packet(self, src_if, dst_if, deid, payload=''):
+        """
+        Create IPv4 packet
+
+        param: src_if
+        param: dst_if
+        """
+        packet = (Ether(dst=src_if.local_mac, src=src_if.remote_mac) /
+                  IP(src=src_if.remote_ip4, dst=deid) /
+                  Raw(payload))
+        return packet
+
+    @abstractmethod
+    def run(self):
+        """ testing procedure """
+        pass
+
+
+class SimpleDriver(Driver):
+    """ Implements simple test procedure """
+    def __init__(self, test, test_cases):
+        super(SimpleDriver, self).__init__(test, test_cases)
+
+    def verify_capture(self, src_loc, dst_loc, capture):
+        """
+        Verify captured packet
+
+        :param src_loc: source locator address
+        :param dst_loc: destination locator address
+        :param capture: list of captured packets
+        """
+        self.test.assertEqual(len(capture), 1, "Unexpected number of "
+                              "packets! Expected 1 but {} received"
+                              .format(len(capture)))
+        packet = capture[0]
+        try:
+            ip_hdr = packet[IP]
+            # assert the values match
+            self.test.assertEqual(ip_hdr.src, src_loc, "IP source address")
+            self.test.assertEqual(ip_hdr.dst, dst_loc,
+                                  "IP destination address")
+            gpe_hdr = packet[LISP_GPE_Header]
+            self.test.assertEqual(gpe_hdr.next_proto, 1,
+                                  "next_proto is not ipv4!")
+            ih = gpe_hdr[IP]
+            self.test.assertEqual(ih.src, self.test.pg0.remote_ip4,
+                                  "unexpected source EID!")
+            self.test.assertEqual(ih.dst, self.test.deid_ip4,
+                                  "unexpected dest EID!")
+        except:
+            self.test.logger.error(ppp("Unexpected or invalid packet:",
+                                   packet))
+            raise
+
+    def configure_tc(self, tc):
+        for config_item in self.config_order:
+            for vpp_object in tc[config_item]:
+                vpp_object.add_vpp_config()
+
+    def run(self, dest):
+        """ Send traffic for each test case and verify that it
+            is encapsulated """
+        for tc in enumerate(self.test_cases):
+            self.test.logger.info('Running {}'.format(tc[1]['name']))
+            self.configure_tc(tc[1])
+
+            print self.test.vapi.cli("sh lisp loc")
+            print self.test.vapi.cli("sh lisp eid")
+            print self.test.vapi.cli("sh lisp adj vni 0")
+            print self.test.vapi.cli("sh lisp gpe entry")
+
+            packet = self.create_packet(self.test.pg0, self.test.pg1, dest,
+                                        'data')
+            self.test.pg0.add_stream(packet)
+            self.test.pg0.enable_capture()
+            self.test.pg1.enable_capture()
+            self.test.pg_start()
+            capture = self.test.pg1.get_capture(1)
+            self.verify_capture(self.test.pg1.local_ip4,
+                                self.test.pg1.remote_ip4, capture)
+            self.test.pg0.assert_nothing_captured()
+
+
+class TestLisp(VppTestCase):
+    """ Basic LISP test """
+
+    @classmethod
+    def setUpClass(cls):
+        super(TestLisp, cls).setUpClass()
+        cls.faf = ForeignAddressFactory()
+        cls.create_pg_interfaces(range(2))  # create pg0 and pg1
+        for i in cls.pg_interfaces:
+            i.admin_up()  # put the interface upsrc_if
+            i.config_ip4()  # configure IPv4 address on the interface
+            i.resolve_arp()  # resolve ARP, so that we know VPP MAC
+
+    def setUp(self):
+        super(TestLisp, self).setUp()
+        self.vapi.lisp_enable_disable(is_enabled=1)
+
+    def test_lisp_basic_encap(self):
+        """Test case for basic encapsulation"""
+
+        self.deid_ip4_net = self.faf.net
+        self.deid_ip4 = self.faf.get_ip4()
+        self.seid_ip4 = '{}/{}'.format(self.pg0.local_ip4, 32)
+        self.rloc_ip4 = self.pg1.remote_ip4n
+
+        test_cases = [
+            {
+                'name': 'basic ip4 over ip4',
+                'locator-sets': [VppLispLocatorSet(self, 'ls-4o4')],
+                'locators': [
+                    VppLispLocator(self, self.pg1.sw_if_index, 'ls-4o4')
+                ],
+                'local-mappings': [
+                    VppLocalMapping(self, self.seid_ip4, 'ls-4o4')
+                ],
+                'remote-mappings': [
+                    VppRemoteMapping(self, self.deid_ip4_net,
+                                     [{
+                                         "is_ip4": 1,
+                                         "priority": 1,
+                                         "weight": 1,
+                                         "addr": self.rloc_ip4
+                                     }])
+                ],
+                'adjacencies': [
+                    VppLispAdjacency(self, self.seid_ip4, self.deid_ip4_net)
+                ]
+            }
+        ]
+        self.test_driver = SimpleDriver(self, test_cases)
+        self.test_driver.run(self.deid_ip4)
+
+
+if __name__ == '__main__':
+    unittest.main(testRunner=VppTestRunner)