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):
15 """ L2XC Test Case """
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)
56 cls.vapi.sw_interface_set_l2_xconnect(
57 cls.pg1.sw_if_index, cls.pg0.sw_if_index, enable=1)
59 # Create bi-directional cross-connects between pg2 and pg3
60 cls.vapi.sw_interface_set_l2_xconnect(
61 cls.pg2.sw_if_index, cls.pg3.sw_if_index, enable=1)
62 cls.vapi.sw_interface_set_l2_xconnect(
63 cls.pg3.sw_if_index, cls.pg2.sw_if_index, enable=1)
65 # mapping between packet-generator index and lists of test hosts
66 cls.hosts_by_pg_idx = dict()
68 # Create host MAC and IPv4 lists
69 cls.create_host_lists(cls.hosts_nr)
71 # setup all interfaces
72 for i in cls.interfaces:
76 super(TestL2xc, cls).tearDownClass()
80 super(TestL2xc, self).setUp()
81 self.reset_packet_infos()
85 Show various debug prints after each test.
87 super(TestL2xc, self).tearDown()
89 self.logger.info(self.vapi.ppcli("show l2patch"))
92 def create_host_lists(cls, count):
94 Method to create required number of MAC and IPv4 addresses.
95 Create required number of host MAC addresses and distribute them among
96 interfaces. Create host IPv4 address for every host MAC address too.
98 :param count: Number of hosts to create MAC and IPv4 addresses for.
100 for pg_if in cls.pg_interfaces:
101 cls.hosts_by_pg_idx[pg_if.sw_if_index] = []
102 hosts = cls.hosts_by_pg_idx[pg_if.sw_if_index]
103 for j in range(0, count):
105 "00:00:00:ff:%02x:%02x" % (pg_if.sw_if_index, j),
106 "172.17.1%02x.%u" % (pg_if.sw_if_index, j))
109 def create_stream(self, src_if, packet_sizes, packets_per_burst):
111 Create input packet stream for defined interface.
113 :param object src_if: Interface to create packet stream for.
114 :param list packet_sizes: List of required packet sizes.
115 :param int packets_per_burst: Number of packets in burst.
116 :return: Stream of packets.
119 for i in range(0, packets_per_burst):
120 dst_if = self.flows[src_if][0]
121 dst_host = random.choice(self.hosts_by_pg_idx[dst_if.sw_if_index])
122 src_host = random.choice(self.hosts_by_pg_idx[src_if.sw_if_index])
123 pkt_info = self.create_packet_info(src_if, dst_if)
124 payload = self.info_to_payload(pkt_info)
125 p = (Ether(dst=dst_host.mac, src=src_host.mac) /
126 IP(src=src_host.ip4, dst=dst_host.ip4) /
127 UDP(sport=1234, dport=1234) /
129 pkt_info.data = p.copy()
130 size = random.choice(packet_sizes)
131 self.extend_packet(p, size)
135 def verify_capture(self, pg_if, capture):
137 Verify captured input packet stream for defined interface.
139 :param object pg_if: Interface to verify captured packet stream for.
140 :param list capture: Captured packet stream.
143 for i in self.interfaces:
144 last_info[i.sw_if_index] = None
145 dst_sw_if_index = pg_if.sw_if_index
146 for packet in capture:
150 payload_info = self.payload_to_info(str(packet[Raw]))
151 packet_index = payload_info.index
152 self.assertEqual(payload_info.dst, dst_sw_if_index)
153 self.logger.debug("Got packet on port %s: src=%u (id=%u)" %
154 (pg_if.name, payload_info.src, packet_index))
155 next_info = self.get_next_packet_info_for_interface2(
156 payload_info.src, dst_sw_if_index,
157 last_info[payload_info.src])
158 last_info[payload_info.src] = next_info
159 self.assertTrue(next_info is not None)
160 self.assertEqual(packet_index, next_info.index)
161 saved_packet = next_info.data
162 # Check standard fields
163 self.assertEqual(ip.src, saved_packet[IP].src)
164 self.assertEqual(ip.dst, saved_packet[IP].dst)
165 self.assertEqual(udp.sport, saved_packet[UDP].sport)
166 self.assertEqual(udp.dport, saved_packet[UDP].dport)
168 self.logger.error(ppp("Unexpected or invalid packet:", packet))
170 for i in self.interfaces:
171 remaining_packet = self.get_next_packet_info_for_interface2(
172 i, dst_sw_if_index, last_info[i.sw_if_index])
173 self.assertTrue(remaining_packet is None,
174 "Port %u: Packet expected from source %u didn't"
175 " arrive" % (dst_sw_if_index, i.sw_if_index))
177 def run_l2xc_test(self, pkts_per_burst):
180 # Create incoming packet streams for packet-generator interfaces
181 for i in self.interfaces:
182 pkts = self.create_stream(i, self.pg_if_packet_sizes,
186 # Enable packet capturing and start packet sending
187 self.pg_enable_capture(self.pg_interfaces)
190 # Verify outgoing packet streams per packet-generator interface
191 for i in self.pg_interfaces:
192 capture = i.get_capture()
193 self.logger.info("Verifying capture on interface %s" % i.name)
194 self.verify_capture(i, capture)
196 def test_l2xc_sl(self):
197 """ L2XC single-loop test
201 2 pairs of 2 interfaces, l2xconnected
203 2. sending l2 eth packets between 4 interfaces
204 64B, 512B, 1518B, 9018B (ether_size)
205 burst of 2 packets per interface
208 self.run_l2xc_test(self.sl_pkts_per_burst)
210 def test_l2xc_dl(self):
211 """ L2XC dual-loop test
215 2 pairs of 2 interfaces, l2xconnected
217 2. sending l2 eth packets between 4 interfaces
218 64B, 512B, 1518B, 9018B (ether_size)
219 burst of 257 packets per interface
222 self.run_l2xc_test(self.dl_pkts_per_burst)
225 if __name__ == '__main__':
226 unittest.main(testRunner=VppTestRunner)