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()
81 Clear trace and packet infos before running each test.
83 super(TestL2xc, self).setUp()
84 self.packet_infos = {}
88 Show various debug prints after each test.
90 super(TestL2xc, self).tearDown()
92 self.logger.info(self.vapi.ppcli("show l2patch"))
95 def create_host_lists(cls, count):
97 Method to create required number of MAC and IPv4 addresses.
98 Create required number of host MAC addresses and distribute them among
99 interfaces. Create host IPv4 address for every host MAC address too.
101 :param count: Number of hosts to create MAC and IPv4 addresses for.
103 for pg_if in cls.pg_interfaces:
104 cls.hosts_by_pg_idx[pg_if.sw_if_index] = []
105 hosts = cls.hosts_by_pg_idx[pg_if.sw_if_index]
106 for j in range(0, count):
108 "00:00:00:ff:%02x:%02x" % (pg_if.sw_if_index, j),
109 "172.17.1%02x.%u" % (pg_if.sw_if_index, j))
112 def create_stream(self, src_if, packet_sizes, packets_per_burst):
114 Create input packet stream for defined interface.
116 :param object src_if: Interface to create packet stream for.
117 :param list packet_sizes: List of required packet sizes.
118 :param int packets_per_burst: Number of packets in burst.
119 :return: Stream of packets.
122 for i in range(0, packets_per_burst):
123 dst_if = self.flows[src_if][0]
124 dst_host = random.choice(self.hosts_by_pg_idx[dst_if.sw_if_index])
125 src_host = random.choice(self.hosts_by_pg_idx[src_if.sw_if_index])
126 pkt_info = self.create_packet_info(
127 src_if.sw_if_index, dst_if.sw_if_index)
128 payload = self.info_to_payload(pkt_info)
129 p = (Ether(dst=dst_host.mac, src=src_host.mac) /
130 IP(src=src_host.ip4, dst=dst_host.ip4) /
131 UDP(sport=1234, dport=1234) /
133 pkt_info.data = p.copy()
134 size = random.choice(packet_sizes)
135 self.extend_packet(p, size)
139 def verify_capture(self, pg_if, capture):
141 Verify captured input packet stream for defined interface.
143 :param object pg_if: Interface to verify captured packet stream for.
144 :param list capture: Captured packet stream.
147 for i in self.interfaces:
148 last_info[i.sw_if_index] = None
149 dst_sw_if_index = pg_if.sw_if_index
150 for packet in capture:
154 payload_info = self.payload_to_info(str(packet[Raw]))
155 packet_index = payload_info.index
156 self.assertEqual(payload_info.dst, dst_sw_if_index)
157 self.logger.debug("Got packet on port %s: src=%u (id=%u)" %
158 (pg_if.name, payload_info.src, packet_index))
159 next_info = self.get_next_packet_info_for_interface2(
160 payload_info.src, dst_sw_if_index,
161 last_info[payload_info.src])
162 last_info[payload_info.src] = next_info
163 self.assertTrue(next_info is not None)
164 self.assertEqual(packet_index, next_info.index)
165 saved_packet = next_info.data
166 # Check standard fields
167 self.assertEqual(ip.src, saved_packet[IP].src)
168 self.assertEqual(ip.dst, saved_packet[IP].dst)
169 self.assertEqual(udp.sport, saved_packet[UDP].sport)
170 self.assertEqual(udp.dport, saved_packet[UDP].dport)
172 self.logger.error(ppp("Unexpected or invalid packet:", packet))
174 for i in self.interfaces:
175 remaining_packet = self.get_next_packet_info_for_interface2(
176 i, dst_sw_if_index, last_info[i.sw_if_index])
177 self.assertTrue(remaining_packet is None,
178 "Port %u: Packet expected from source %u didn't"
179 " arrive" % (dst_sw_if_index, i.sw_if_index))
181 def run_l2xc_test(self, pkts_per_burst):
184 # Create incoming packet streams for packet-generator interfaces
185 for i in self.interfaces:
186 pkts = self.create_stream(i, self.pg_if_packet_sizes,
190 # Enable packet capturing and start packet sending
191 self.pg_enable_capture(self.pg_interfaces)
194 # Verify outgoing packet streams per packet-generator interface
195 for i in self.pg_interfaces:
196 capture = i.get_capture()
197 self.logger.info("Verifying capture on interface %s" % i.name)
198 self.verify_capture(i, capture)
200 def test_l2xc_sl(self):
201 """ L2XC single-loop test
205 2 pairs of 2 interfaces, l2xconnected
207 2. sending l2 eth packets between 4 interfaces
208 64B, 512B, 1518B, 9018B (ether_size)
209 burst of 2 packets per interface
212 self.run_l2xc_test(self.sl_pkts_per_burst)
214 def test_l2xc_dl(self):
215 """ L2XC dual-loop test
219 2 pairs of 2 interfaces, l2xconnected
221 2. sending l2 eth packets between 4 interfaces
222 64B, 512B, 1518B, 9018B (ether_size)
223 burst of 257 packets per interface
226 self.run_l2xc_test(self.dl_pkts_per_burst)
229 if __name__ == '__main__':
230 unittest.main(testRunner=VppTestRunner)