test/test_lisp.py

#!/usr/bin/env python3

import abc
import unittest

from scapy.fields import BitField, ByteField, FlagsField, IntField
from scapy.packet import bind_layers, Packet, Raw
from scapy.layers.inet import IP, UDP, Ether
from scapy.layers.inet6 import IPv6

from framework import VppTestCase
from asfframework import VppTestRunner
from lisp import (
    VppLocalMapping,
    VppLispAdjacency,
    VppLispLocator,
    VppLispLocatorSet,
    VppRemoteMapping,
    LispRemoteLocator,
)
from util import ppp
from config import config

# From py_lispnetworking.lisp.py:  # GNU General Public License v2.0


class LISP_GPE_Header(Packet):
    name = "LISP GPE Header"
    fields_desc = [
        FlagsField("gpe_flags", None, 6, ["N", "L", "E", "V", "I", "P"]),
        BitField("reserved", 0, 18),
        ByteField("next_proto", 0),
        IntField("iid", 0),
    ]


bind_layers(UDP, LISP_GPE_Header, dport=4341)
bind_layers(UDP, LISP_GPE_Header, sport=4341)
bind_layers(LISP_GPE_Header, IP, next_proto=1)
bind_layers(LISP_GPE_Header, IPv6, next_proto=2)
bind_layers(LISP_GPE_Header, Ether, next_proto=3)


class ForeignAddressFactory(object):
    count = 0
    prefix_len = 24
    net_template = "10.10.10.{}"
    net = net_template.format(0) + "/" + str(prefix_len)

    def get_ip4(self):
        if self.count > 255:
            raise Exception("Network host address exhaustion")
        self.count += 1
        return self.net_template.format(self.count)


class Driver(metaclass=abc.ABCMeta):
    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

    @abc.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])

            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()


@unittest.skipIf("lisp" in config.excluded_plugins, "Exclude LISP plugin tests")
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

    @classmethod
    def tearDownClass(cls):
        super(TestLisp, cls).tearDownClass()

    def setUp(self):
        super(TestLisp, self).setUp()
        self.vapi.lisp_enable_disable(is_enable=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 = "{!s}/{!s}".format(self.pg0.local_ip4, 32)
        self.rloc_ip4 = self.pg1.remote_ip4

        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, [LispRemoteLocator(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)


@unittest.skipIf("lisp" in config.excluded_plugins, "Exclude LISP plugin tests")
class TestLispUT(VppTestCase):
    """Lisp UT"""

    @classmethod
    def setUpClass(cls):
        super(TestLispUT, cls).setUpClass()

    @classmethod
    def tearDownClass(cls):
        super(TestLispUT, cls).tearDownClass()

    def test_fib(self):
        """LISP Unit Tests"""
        error = self.vapi.cli("test lisp cp")

        if error:
            self.logger.critical(error)
        self.assertNotIn("Failed", error)


if __name__ == "__main__":
    unittest.main(testRunner=VppTestRunner)