From 44db1caefbf5067b0cf0073299c9f21265331412 Mon Sep 17 00:00:00 2001 From: Neale Ranns Date: Thu, 24 Dec 2020 09:16:09 +0000 Subject: linux-cp: Linux Interface Mirroring for Control Plane Integration Type: feature please see FEATURE.yaml for details. Signed-off-by: Neale Ranns Signed-off-by: Matthew Smith Signed-off-by: Jon Loeliger Signed-off-by: Pim van Pelt Change-Id: I04a45c15c0838906aa787e06660fa29f39f755fa --- src/plugins/linux-cp/test/test_linux_cp.py | 174 +++++++++++++++++++++++++++++ 1 file changed, 174 insertions(+) create mode 100644 src/plugins/linux-cp/test/test_linux_cp.py (limited to 'src/plugins/linux-cp/test/test_linux_cp.py') diff --git a/src/plugins/linux-cp/test/test_linux_cp.py b/src/plugins/linux-cp/test/test_linux_cp.py new file mode 100644 index 00000000000..df38681b16e --- /dev/null +++ b/src/plugins/linux-cp/test/test_linux_cp.py @@ -0,0 +1,174 @@ +#!/usr/bin/env python3 + +import unittest + +from scapy.layers.inet import IP, UDP +from scapy.layers.inet6 import IPv6, Raw +from scapy.layers.l2 import Ether, ARP, Dot1Q + +from vpp_object import VppObject +from framework import VppTestCase, VppTestRunner + + +class VppLcpPair(VppObject): + def __init__(self, test, phy, host): + self._test = test + self.phy = phy + self.host = host + + def add_vpp_config(self): + self._test.vapi.cli("test lcp add phy %s host %s" % + (self.phy, self.host)) + self._test.registry.register(self, self._test.logger) + return self + + def remove_vpp_config(self): + self._test.vapi.cli("test lcp del phy %s host %s" % + (self.phy, self.host)) + + def object_id(self): + return "lcp:%d:%d" % (self.phy.sw_if_index, + self.host.sw_if_index) + + def query_vpp_config(self): + pairs = list(self._test.vapi.vpp.details_iter( + self._test.vapi.lcp_itf_pair_get)) + + for p in pairs: + if p.phy_sw_if_index == self.phy.sw_if_index and \ + p.host_sw_if_index == self.host.sw_if_index: + return True + return False + + +class TestLinuxCP(VppTestCase): + """ Linux Control Plane """ + + extra_vpp_plugin_config = ["plugin", + "linux_cp_plugin.so", + "{", "enable", "}", + "plugin", + "linux_cp_unittest_plugin.so", + "{", "enable", "}"] + + @classmethod + def setUpClass(cls): + super(TestLinuxCP, cls).setUpClass() + + @classmethod + def tearDownClass(cls): + super(TestLinuxCP, cls).tearDownClass() + + def setUp(self): + super(TestLinuxCP, self).setUp() + + # create 4 pg interfaces so there are a few addresses + # in the FIB + self.create_pg_interfaces(range(4)) + + for i in self.pg_interfaces: + i.admin_up() + + def tearDown(self): + for i in self.pg_interfaces: + i.admin_down() + super(TestLinuxCP, self).tearDown() + + def test_linux_cp_tap(self): + """ Linux CP TAP """ + + # + # Setup + # + + arp_opts = {"who-has": 1, "is-at": 2} + + # create two pairs, wihch a bunch of hots on the phys + hosts = [self.pg0, self.pg1] + phys = [self.pg2, self.pg3] + N_HOSTS = 4 + + for phy in phys: + phy.config_ip4() + phy.generate_remote_hosts(4) + phy.configure_ipv4_neighbors() + + pair1 = VppLcpPair(self, phys[0], hosts[0]).add_vpp_config() + pair2 = VppLcpPair(self, phys[1], hosts[1]).add_vpp_config() + + self.logger.info(self.vapi.cli("sh lcp adj verbose")) + self.logger.info(self.vapi.cli("sh lcp")) + + # + # Traffic Tests + # + + # hosts to phys + for phy, host in zip(phys, hosts): + for j in range(N_HOSTS): + p = (Ether(src=phy.local_mac, + dst=phy.remote_hosts[j].mac) / + IP(src=phy.local_ip4, + dst=phy.remote_hosts[j].ip4) / + UDP(sport=1234, dport=1234) / + Raw()) + + rxs = self.send_and_expect(host, [p], phy) + + # verify packet is unchanged + for rx in rxs: + self.assertEqual(p.show2(True), rx.show2(True)) + + # ARPs x-connect to phy + p = (Ether(dst="ff:ff:ff:ff:ff:ff", + src=phy.remote_hosts[j].mac) / + ARP(op="who-has", + hwdst=phy.remote_hosts[j].mac, + hwsrc=phy.local_mac, + psrc=phy.local_ip4, + pdst=phy.remote_hosts[j].ip4)) + + rxs = self.send_and_expect(host, [p], phy) + + # verify packet is unchanged + for rx in rxs: + self.assertEqual(p.show2(True), rx.show2(True)) + + # phy to host + for phy, host in zip(phys, hosts): + for j in range(N_HOSTS): + p = (Ether(dst=phy.local_mac, + src=phy.remote_hosts[j].mac) / + IP(dst=phy.local_ip4, + src=phy.remote_hosts[j].ip4) / + UDP(sport=1234, dport=1234) / + Raw()) + + rxs = self.send_and_expect(phy, [p], host) + + # verify packet is unchanged + for rx in rxs: + self.assertEqual(p.show2(True), rx.show2(True)) + + # ARPs rx'd on the phy are sent to the host + p = (Ether(dst="ff:ff:ff:ff:ff:ff", + src=phy.remote_hosts[j].mac) / + ARP(op="is-at", + hwsrc=phy.remote_hosts[j].mac, + hwdst=phy.local_mac, + pdst=phy.local_ip4, + psrc=phy.remote_hosts[j].ip4)) + + rxs = self.send_and_expect(phy, [p], host) + + # verify packet is unchanged + for rx in rxs: + self.assertEqual(p.show2(True), rx.show2(True)) + + # cleanup + for phy in phys: + phy.unconfig_ip4() + + +if __name__ == '__main__': + unittest.main(testRunner=VppTestRunner) -- cgit 1.2.3-korg