tests: run a test inside a QEMU VM

Use the script test/run.py to run a test named test_vm_tap
inside a QEMU VM. The run script builds out a virtual env,
launches a light weight QEMU VM, mounts host directories,
starts VPP inside the VM and runs the test. The test named
test_vm_tap, creates two tap v2 interfaces in separate Linux
namespaces and using iPerf, streams traffic between the VM
and VPP. All data files are stored in the directory named
/tmp/vpp-vm-tests. To clean up, use the make test-wipe
command.
Usage:
test/run.py --vm --debug --test test_vm_tap

Type: improvement

Change-Id: I4425dbef52acee1e5b8af5acaa169b89a2c0f171
Signed-off-by: Naveen Joy <najoy@cisco.com>

1 files changed, 102 insertions, 0 deletions

diff --git a/test/test_vm_tap.py b/test/test_vm_tap.py
new file mode 100644
index 00000000000..6787ca1e3d7
--- /dev/null
+++ b/test/test_vm_tap.py
@@ -0,0 +1,102 @@
+#!/usr/bin/env python3
+import unittest
+from ipaddress import ip_interface
+from vpp_qemu_utils import create_namespace
+from vpp_iperf import VppIperf
+from framework import VppTestCase, VppTestRunner
+from config import config
+
+
+class TestTapQemu(VppTestCase):
+    """Test Tap interfaces inside a QEMU VM.
+
+    Start an iPerf connection stream between QEMU and VPP via
+    tap v2 interfaces.
+
+    Linux_ns1 -- iperf_client -- tap1 -- VPP-BD -- tap2 --
+                              -- iperfServer -- Linux_ns2
+    """
+
+    @classmethod
+    def setUpClass(cls):
+        super(TestTapQemu, cls).setUpClass()
+
+    @classmethod
+    def tearDownClass(cls):
+        super(TestTapQemu, cls).tearDownClass()
+
+    def setUp(self):
+        """Perform test setup before running QEMU tests.
+
+        1. Create a namespace for the iPerf Server & Client.
+        2. Create 2 tap interfaces in VPP & add them to each namespace.
+        3. Add the tap interfaces to a bridge-domain.
+        """
+        super(TestTapQemu, self).setUp()
+        self.client_namespace = "iprf_client_ns"
+        self.server_namespace = "iprf_server_ns"
+        self.client_ip4_prefix = "10.0.0.101/24"
+        self.server_ip4_prefix = "10.0.0.102/24"
+        create_namespace(self.client_namespace)
+        create_namespace(self.server_namespace)
+        tap1_if_idx = self.create_tap(
+            101, self.client_namespace, self.client_ip4_prefix
+        )
+        tap2_if_idx = self.create_tap(
+            102, self.server_namespace, self.server_ip4_prefix
+        )
+        self.l2_connect_interfaces(tap1_if_idx, tap2_if_idx)
+
+    def create_tap(self, id, host_namespace, host_ip4_prefix):
+        result = self.vapi.api(
+            self.vapi.papi.tap_create_v2,
+            {
+                "id": id,
+                "use_random_mac": True,
+                "host_namespace_set": True,
+                "host_namespace": host_namespace,
+                "host_if_name_set": False,
+                "host_bridge_set": False,
+                "host_mac_addr_set": False,
+                "host_ip4_prefix": ip_interface(host_ip4_prefix),
+                "host_ip4_prefix_set": True,
+            },
+        )
+        sw_if_index = result.sw_if_index
+        self.vapi.api(
+            self.vapi.papi.sw_interface_set_flags,
+            {"sw_if_index": sw_if_index, "flags": 1},
+        )
+        return sw_if_index
+
+    def dump_vpp_tap_interfaces(self):
+        return self.vapi.api(self.vapi.papi.sw_interface_tap_v2_dump, {})
+
+    def dump_bridge_domain_details(self):
+        return self.vapi.api(self.vapi.papi.bridge_domain_dump, {"bd_id": 1})
+
+    def l2_connect_interfaces(self, *sw_if_idxs):
+        for if_idx in sw_if_idxs:
+            self.vapi.api(
+                self.vapi.papi.sw_interface_set_l2_bridge,
+                {
+                    "rx_sw_if_index": if_idx,
+                    "bd_id": 1,
+                    "shg": 0,
+                    "port_type": 0,
+                    "enable": True,
+                },
+            )
+
+    @unittest.skipUnless(config.extended, "part of extended tests")
+    def test_tap_iperf(self):
+        """Start an iperf connection stream between QEMU & VPP via tap."""
+        iperf = VppIperf()
+        iperf.client_ns = self.client_namespace
+        iperf.server_ns = self.server_namespace
+        iperf.server_ip = str(ip_interface(self.server_ip4_prefix).ip)
+        iperf.start()
+
+
+if __name__ == "__main__":
+    unittest.main(testRunner=VppTestRunner)