diff options
Diffstat (limited to 'src/plugins/l3xc/test')
| -rw-r--r-- | src/plugins/l3xc/test/test_l3xc.py | 152 |
1 files changed, 0 insertions, 152 deletions
diff --git a/src/plugins/l3xc/test/test_l3xc.py b/src/plugins/l3xc/test/test_l3xc.py deleted file mode 100644 index d7a82976cf5..00000000000 --- a/src/plugins/l3xc/test/test_l3xc.py +++ /dev/null @@ -1,152 +0,0 @@ -#!/usr/bin/env python3 - -from socket import inet_pton, inet_ntop, AF_INET, AF_INET6 -import unittest - -from framework import VppTestCase, VppTestRunner -from vpp_ip import DpoProto -from vpp_ip_route import VppIpRoute, VppRoutePath, VppMplsLabel, VppIpTable - -from scapy.packet import Raw -from scapy.layers.l2 import Ether -from scapy.layers.inet import IP, UDP -from scapy.layers.inet6 import IPv6 - -from vpp_object import VppObject - -NUM_PKTS = 67 - - -def find_l3xc(test, sw_if_index, dump_sw_if_index=None): - if not dump_sw_if_index: - dump_sw_if_index = sw_if_index - xcs = test.vapi.l3xc_dump(dump_sw_if_index) - for xc in xcs: - if sw_if_index == xc.l3xc.sw_if_index: - return True - return False - - -class VppL3xc(VppObject): - - def __init__(self, test, intf, paths, is_ip6=False): - self._test = test - self.intf = intf - self.is_ip6 = is_ip6 - self.paths = paths - self.encoded_paths = [] - for path in self.paths: - self.encoded_paths.append(path.encode()) - - def add_vpp_config(self): - self._test.vapi.l3xc_update( - l3xc={ - 'is_ip6': self.is_ip6, - 'sw_if_index': self.intf.sw_if_index, - 'n_paths': len(self.paths), - 'paths': self.encoded_paths - }) - self._test.registry.register(self, self._test.logger) - - def remove_vpp_config(self): - self._test.vapi.l3xc_del( - is_ip6=self.is_ip6, - sw_if_index=self.intf.sw_if_index) - - def query_vpp_config(self): - return find_l3xc(self._test, self.intf.sw_if_index) - - def object_id(self): - return ("l3xc-%d" % self.intf.sw_if_index) - - -class TestL3xc(VppTestCase): - """ L3XC Test Case """ - - @classmethod - def setUpClass(cls): - super(TestL3xc, cls).setUpClass() - - @classmethod - def tearDownClass(cls): - super(TestL3xc, cls).tearDownClass() - - def setUp(self): - super(TestL3xc, self).setUp() - - self.create_pg_interfaces(range(6)) - - for i in self.pg_interfaces: - i.admin_up() - i.config_ip4() - i.resolve_arp() - i.config_ip6() - i.resolve_ndp() - - def tearDown(self): - for i in self.pg_interfaces: - i.unconfig_ip4() - i.unconfig_ip6() - i.admin_down() - super(TestL3xc, self).tearDown() - - def send_and_expect_load_balancing(self, input, pkts, outputs): - self.pg_send(input, pkts) - rxs = [] - for oo in outputs: - rx = oo._get_capture(1) - self.assertNotEqual(0, len(rx)) - for r in rx: - rxs.append(r) - return rxs - - def test_l3xc4(self): - """ IPv4 X-Connect """ - - # - # x-connect pg0 to pg1 and pg2 to pg3->5 - # - l3xc_1 = VppL3xc(self, self.pg0, - [VppRoutePath(self.pg1.remote_ip4, - self.pg1.sw_if_index)]) - l3xc_1.add_vpp_config() - l3xc_2 = VppL3xc(self, self.pg2, - [VppRoutePath(self.pg3.remote_ip4, - self.pg3.sw_if_index), - VppRoutePath(self.pg4.remote_ip4, - self.pg4.sw_if_index), - VppRoutePath(self.pg5.remote_ip4, - self.pg5.sw_if_index)]) - l3xc_2.add_vpp_config() - - self.assertTrue(find_l3xc(self, self.pg2.sw_if_index, 0xffffffff)) - - self.logger.info(self.vapi.cli("sh l3xc")) - - # - # fire in packets. If it's forwarded then the L3XC was successful, - # since default routing will drop it - # - p_1 = (Ether(src=self.pg0.remote_mac, - dst=self.pg0.local_mac) / - IP(src="1.1.1.1", dst="1.1.1.2") / - UDP(sport=1234, dport=1234) / - Raw(b'\xa5' * 100)) - # self.send_and_expect(self.pg0, p_1*NUM_PKTS, self.pg1) - - p_2 = [] - for ii in range(NUM_PKTS): - p_2.append(Ether(src=self.pg0.remote_mac, - dst=self.pg0.local_mac) / - IP(src="1.1.1.1", dst="1.1.1.2") / - UDP(sport=1000 + ii, dport=1234) / - Raw(b'\xa5' * 100)) - self.send_and_expect_load_balancing(self.pg2, p_2, - [self.pg3, self.pg4, self.pg5]) - - l3xc_2.remove_vpp_config() - self.send_and_assert_no_replies(self.pg2, p_2) - - -if __name__ == '__main__': - unittest.main(testRunner=VppTestRunner) |