1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
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)
|
