6 from scapy.packet import Raw
7 from scapy.layers.l2 import Ether
8 from scapy.layers.inet import IP, UDP
10 from framework import VppTestCase, VppTestRunner
11 from util import Host, ppp
14 class TestL2xc(VppTestCase):
20 Perform standard class setup (defined by class method setUpClass in
21 class VppTestCase) before running the test case, set test case related
22 variables and configure VPP.
24 :var int hosts_nr: Number of hosts to be created.
25 :var int dl_pkts_per_burst: Number of packets in burst for dual-loop
27 :var int sl_pkts_per_burst: Number of packets in burst for single-loop
30 super(TestL2xc, cls).setUpClass()
34 cls.dl_pkts_per_burst = 257
35 cls.sl_pkts_per_burst = 2
38 # create 4 pg interfaces
39 cls.create_pg_interfaces(range(4))
41 # packet flows mapping pg0 -> pg1, pg2 -> pg3, etc.
43 cls.flows[cls.pg0] = [cls.pg1]
44 cls.flows[cls.pg1] = [cls.pg0]
45 cls.flows[cls.pg2] = [cls.pg3]
46 cls.flows[cls.pg3] = [cls.pg2]
49 cls.pg_if_packet_sizes = [64, 512, 1518, 9018]
51 cls.interfaces = list(cls.pg_interfaces)
53 # Create bi-directional cross-connects between pg0 and pg1
54 cls.vapi.sw_interface_set_l2_xconnect(
55 cls.pg0.sw_if_index, cls.pg1.sw_if_index, enable=1
57 cls.vapi.sw_interface_set_l2_xconnect(
58 cls.pg1.sw_if_index, cls.pg0.sw_if_index, enable=1
61 # Create bi-directional cross-connects between pg2 and pg3
62 cls.vapi.sw_interface_set_l2_xconnect(
63 cls.pg2.sw_if_index, cls.pg3.sw_if_index, enable=1
65 cls.vapi.sw_interface_set_l2_xconnect(
66 cls.pg3.sw_if_index, cls.pg2.sw_if_index, enable=1
69 # mapping between packet-generator index and lists of test hosts
70 cls.hosts_by_pg_idx = dict()
72 # Create host MAC and IPv4 lists
73 cls.create_host_lists(cls.hosts_nr)
75 # setup all interfaces
76 for i in cls.interfaces:
80 super(TestL2xc, cls).tearDownClass()
84 def tearDownClass(cls):
85 super(TestL2xc, cls).tearDownClass()
88 super(TestL2xc, self).setUp()
89 self.reset_packet_infos()
93 Show various debug prints after each test.
95 super(TestL2xc, self).tearDown()
97 def show_commands_at_teardown(self):
98 self.logger.info(self.vapi.ppcli("show l2patch"))
101 def create_host_lists(cls, count):
103 Method to create required number of MAC and IPv4 addresses.
104 Create required number of host MAC addresses and distribute them among
105 interfaces. Create host IPv4 address for every host MAC address too.
107 :param count: Number of hosts to create MAC and IPv4 addresses for.
109 for pg_if in cls.pg_interfaces:
110 cls.hosts_by_pg_idx[pg_if.sw_if_index] = []
111 hosts = cls.hosts_by_pg_idx[pg_if.sw_if_index]
112 for j in range(0, count):
114 "00:00:00:ff:%02x:%02x" % (pg_if.sw_if_index, j),
115 "172.17.1%02x.%u" % (pg_if.sw_if_index, j),
119 def create_stream(self, src_if, packet_sizes, packets_per_burst):
121 Create input packet stream for defined interface.
123 :param object src_if: Interface to create packet stream for.
124 :param list packet_sizes: List of required packet sizes.
125 :param int packets_per_burst: Number of packets in burst.
126 :return: Stream of packets.
129 for i in range(0, packets_per_burst):
130 dst_if = self.flows[src_if][0]
131 dst_host = random.choice(self.hosts_by_pg_idx[dst_if.sw_if_index])
132 src_host = random.choice(self.hosts_by_pg_idx[src_if.sw_if_index])
133 pkt_info = self.create_packet_info(src_if, dst_if)
134 payload = self.info_to_payload(pkt_info)
136 Ether(dst=dst_host.mac, src=src_host.mac)
137 / IP(src=src_host.ip4, dst=dst_host.ip4)
138 / UDP(sport=1234, dport=1234)
141 pkt_info.data = p.copy()
142 size = random.choice(packet_sizes)
143 self.extend_packet(p, size)
147 def verify_capture(self, pg_if, capture):
149 Verify captured input packet stream for defined interface.
151 :param object pg_if: Interface to verify captured packet stream for.
152 :param list capture: Captured packet stream.
155 for i in self.interfaces:
156 last_info[i.sw_if_index] = None
157 dst_sw_if_index = pg_if.sw_if_index
158 for packet in capture:
162 payload_info = self.payload_to_info(packet[Raw])
163 packet_index = payload_info.index
164 self.assertEqual(payload_info.dst, dst_sw_if_index)
166 "Got packet on port %s: src=%u (id=%u)"
167 % (pg_if.name, payload_info.src, packet_index)
169 next_info = self.get_next_packet_info_for_interface2(
170 payload_info.src, dst_sw_if_index, last_info[payload_info.src]
172 last_info[payload_info.src] = next_info
173 self.assertTrue(next_info is not None)
174 self.assertEqual(packet_index, next_info.index)
175 saved_packet = next_info.data
176 # Check standard fields
177 self.assertEqual(ip.src, saved_packet[IP].src)
178 self.assertEqual(ip.dst, saved_packet[IP].dst)
179 self.assertEqual(udp.sport, saved_packet[UDP].sport)
180 self.assertEqual(udp.dport, saved_packet[UDP].dport)
182 self.logger.error(ppp("Unexpected or invalid packet:", packet))
184 for i in self.interfaces:
185 remaining_packet = self.get_next_packet_info_for_interface2(
186 i, dst_sw_if_index, last_info[i.sw_if_index]
189 remaining_packet is None,
190 "Port %u: Packet expected from source %u didn't"
191 " arrive" % (dst_sw_if_index, i.sw_if_index),
194 def run_l2xc_test(self, pkts_per_burst):
197 # Create incoming packet streams for packet-generator interfaces
198 for i in self.interfaces:
199 pkts = self.create_stream(i, self.pg_if_packet_sizes, pkts_per_burst)
202 # Enable packet capturing and start packet sending
203 self.pg_enable_capture(self.pg_interfaces)
206 # Verify outgoing packet streams per packet-generator interface
207 for i in self.pg_interfaces:
208 capture = i.get_capture()
209 self.logger.info("Verifying capture on interface %s" % i.name)
210 self.verify_capture(i, capture)
212 def test_l2xc_sl(self):
213 """L2XC single-loop test
217 2 pairs of 2 interfaces, l2xconnected
219 2. sending l2 eth packets between 4 interfaces
220 64B, 512B, 1518B, 9018B (ether_size)
221 burst of 2 packets per interface
224 self.run_l2xc_test(self.sl_pkts_per_burst)
226 def test_l2xc_dl(self):
227 """L2XC dual-loop test
231 2 pairs of 2 interfaces, l2xconnected
233 2. sending l2 eth packets between 4 interfaces
234 64B, 512B, 1518B, 9018B (ether_size)
235 burst of 257 packets per interface
238 self.run_l2xc_test(self.dl_pkts_per_burst)
241 if __name__ == "__main__":
242 unittest.main(testRunner=VppTestRunner)