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
11 from asfframework import VppTestRunner
12 from util import Host, ppp
15 class TestL2xc(VppTestCase):
21 Perform standard class setup (defined by class method setUpClass in
22 class VppTestCase) before running the test case, set test case related
23 variables and configure VPP.
25 :var int hosts_nr: Number of hosts to be created.
26 :var int dl_pkts_per_burst: Number of packets in burst for dual-loop
28 :var int sl_pkts_per_burst: Number of packets in burst for single-loop
31 super(TestL2xc, cls).setUpClass()
35 cls.dl_pkts_per_burst = 257
36 cls.sl_pkts_per_burst = 2
39 # create 4 pg interfaces
40 cls.create_pg_interfaces(range(4))
42 # packet flows mapping pg0 -> pg1, pg2 -> pg3, etc.
44 cls.flows[cls.pg0] = [cls.pg1]
45 cls.flows[cls.pg1] = [cls.pg0]
46 cls.flows[cls.pg2] = [cls.pg3]
47 cls.flows[cls.pg3] = [cls.pg2]
50 cls.pg_if_packet_sizes = [64, 512, 1518, 9018]
52 cls.interfaces = list(cls.pg_interfaces)
54 # Create bi-directional cross-connects between pg0 and pg1
55 cls.vapi.sw_interface_set_l2_xconnect(
56 cls.pg0.sw_if_index, cls.pg1.sw_if_index, enable=1
58 cls.vapi.sw_interface_set_l2_xconnect(
59 cls.pg1.sw_if_index, cls.pg0.sw_if_index, enable=1
62 # Create bi-directional cross-connects between pg2 and pg3
63 cls.vapi.sw_interface_set_l2_xconnect(
64 cls.pg2.sw_if_index, cls.pg3.sw_if_index, enable=1
66 cls.vapi.sw_interface_set_l2_xconnect(
67 cls.pg3.sw_if_index, cls.pg2.sw_if_index, enable=1
70 # mapping between packet-generator index and lists of test hosts
71 cls.hosts_by_pg_idx = dict()
73 # Create host MAC and IPv4 lists
74 cls.create_host_lists(cls.hosts_nr)
76 # setup all interfaces
77 for i in cls.interfaces:
81 super(TestL2xc, cls).tearDownClass()
85 def tearDownClass(cls):
86 super(TestL2xc, cls).tearDownClass()
89 super(TestL2xc, self).setUp()
90 self.reset_packet_infos()
94 Show various debug prints after each test.
96 super(TestL2xc, self).tearDown()
98 def show_commands_at_teardown(self):
99 self.logger.info(self.vapi.ppcli("show l2patch"))
102 def create_host_lists(cls, count):
104 Method to create required number of MAC and IPv4 addresses.
105 Create required number of host MAC addresses and distribute them among
106 interfaces. Create host IPv4 address for every host MAC address too.
108 :param count: Number of hosts to create MAC and IPv4 addresses for.
110 for pg_if in cls.pg_interfaces:
111 cls.hosts_by_pg_idx[pg_if.sw_if_index] = []
112 hosts = cls.hosts_by_pg_idx[pg_if.sw_if_index]
113 for j in range(0, count):
115 "00:00:00:ff:%02x:%02x" % (pg_if.sw_if_index, j),
116 "172.17.1%02x.%u" % (pg_if.sw_if_index, j),
120 def create_stream(self, src_if, packet_sizes, packets_per_burst):
122 Create input packet stream for defined interface.
124 :param object src_if: Interface to create packet stream for.
125 :param list packet_sizes: List of required packet sizes.
126 :param int packets_per_burst: Number of packets in burst.
127 :return: Stream of packets.
130 for i in range(0, packets_per_burst):
131 dst_if = self.flows[src_if][0]
132 dst_host = random.choice(self.hosts_by_pg_idx[dst_if.sw_if_index])
133 src_host = random.choice(self.hosts_by_pg_idx[src_if.sw_if_index])
134 pkt_info = self.create_packet_info(src_if, dst_if)
135 payload = self.info_to_payload(pkt_info)
137 Ether(dst=dst_host.mac, src=src_host.mac)
138 / IP(src=src_host.ip4, dst=dst_host.ip4)
139 / UDP(sport=1234, dport=1234)
142 pkt_info.data = p.copy()
143 size = random.choice(packet_sizes)
144 self.extend_packet(p, size)
148 def verify_capture(self, pg_if, capture):
150 Verify captured input packet stream for defined interface.
152 :param object pg_if: Interface to verify captured packet stream for.
153 :param list capture: Captured packet stream.
156 for i in self.interfaces:
157 last_info[i.sw_if_index] = None
158 dst_sw_if_index = pg_if.sw_if_index
159 for packet in capture:
163 payload_info = self.payload_to_info(packet[Raw])
164 packet_index = payload_info.index
165 self.assertEqual(payload_info.dst, dst_sw_if_index)
167 "Got packet on port %s: src=%u (id=%u)"
168 % (pg_if.name, payload_info.src, packet_index)
170 next_info = self.get_next_packet_info_for_interface2(
171 payload_info.src, dst_sw_if_index, last_info[payload_info.src]
173 last_info[payload_info.src] = next_info
174 self.assertTrue(next_info is not None)
175 self.assertEqual(packet_index, next_info.index)
176 saved_packet = next_info.data
177 # Check standard fields
178 self.assertEqual(ip.src, saved_packet[IP].src)
179 self.assertEqual(ip.dst, saved_packet[IP].dst)
180 self.assertEqual(udp.sport, saved_packet[UDP].sport)
181 self.assertEqual(udp.dport, saved_packet[UDP].dport)
183 self.logger.error(ppp("Unexpected or invalid packet:", packet))
185 for i in self.interfaces:
186 remaining_packet = self.get_next_packet_info_for_interface2(
187 i, dst_sw_if_index, last_info[i.sw_if_index]
190 remaining_packet is None,
191 "Port %u: Packet expected from source %u didn't"
192 " arrive" % (dst_sw_if_index, i.sw_if_index),
195 def run_l2xc_test(self, pkts_per_burst):
198 # Create incoming packet streams for packet-generator interfaces
199 for i in self.interfaces:
200 pkts = self.create_stream(i, self.pg_if_packet_sizes, pkts_per_burst)
203 # Enable packet capturing and start packet sending
204 self.pg_enable_capture(self.pg_interfaces)
207 # Verify outgoing packet streams per packet-generator interface
208 for i in self.pg_interfaces:
209 capture = i.get_capture()
210 self.logger.info("Verifying capture on interface %s" % i.name)
211 self.verify_capture(i, capture)
213 def test_l2xc_sl(self):
214 """L2XC single-loop test
218 2 pairs of 2 interfaces, l2xconnected
220 2. sending l2 eth packets between 4 interfaces
221 64B, 512B, 1518B, 9018B (ether_size)
222 burst of 2 packets per interface
225 self.run_l2xc_test(self.sl_pkts_per_burst)
227 def test_l2xc_dl(self):
228 """L2XC dual-loop test
232 2 pairs of 2 interfaces, l2xconnected
234 2. sending l2 eth packets between 4 interfaces
235 64B, 512B, 1518B, 9018B (ether_size)
236 burst of 257 packets per interface
239 self.run_l2xc_test(self.dl_pkts_per_burst)
242 if __name__ == "__main__":
243 unittest.main(testRunner=VppTestRunner)