+++ /dev/null
-#!/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, VppTestRunner
-from lisp import VppLocalMapping, VppLispAdjacency, VppLispLocator, \
- VppLispLocatorSet, VppRemoteMapping, LispRemoteLocator
-from util import ppp
-
-# 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()
-
-
-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)
-
-
-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)
Code Review / vpp.git / blobdiff