6 from scapy.packet import Raw
7 from scapy.layers.l2 import Ether
8 from scapy.layers.inet import IP, UDP
11 from framework import VppTestCase, VppTestRunner
12 from util import TestHost
15 class TestL2xc(VppTestCase):
16 """ L2XC Test Case """
19 hosts_nr = 10 # Number of hosts
20 pkts_per_burst = 257 # Number of packets per burst
24 super(TestL2xc, cls).setUpClass()
27 super(TestL2xc, self).setUp()
29 # create 4 pg interfaces
30 self.create_pg_interfaces(range(4))
32 # packet flows mapping pg0 -> pg1, pg2 -> pg3, etc.
34 self.flows[self.pg0] = [self.pg1]
35 self.flows[self.pg1] = [self.pg0]
36 self.flows[self.pg2] = [self.pg3]
37 self.flows[self.pg3] = [self.pg2]
40 self.pg_if_packet_sizes = [64, 512, 1518, 9018]
42 self.interfaces = list(self.pg_interfaces)
44 # Create bi-directional cross-connects between pg0 and pg1
45 self.vapi.sw_interface_set_l2_xconnect(
46 self.pg0.sw_if_index, self.pg1.sw_if_index, enable=1)
47 self.vapi.sw_interface_set_l2_xconnect(
48 self.pg1.sw_if_index, self.pg0.sw_if_index, enable=1)
50 # Create bi-directional cross-connects between pg2 and pg3
51 self.vapi.sw_interface_set_l2_xconnect(
52 self.pg2.sw_if_index, self.pg3.sw_if_index, enable=1)
53 self.vapi.sw_interface_set_l2_xconnect(
54 self.pg3.sw_if_index, self.pg2.sw_if_index, enable=1)
56 info(self.vapi.cli("show l2patch"))
58 # mapping between packet-generator index and lists of test hosts
59 self.hosts_by_pg_idx = dict()
61 # Create host MAC and IPv4 lists
62 # self.MY_MACS = dict()
63 # self.MY_IP4S = dict()
64 self.create_host_lists(TestL2xc.hosts_nr)
66 # setup all interfaces
67 for i in self.interfaces:
71 super(TestL2xc, self).tearDown()
73 info(self.vapi.cli("show l2patch"))
75 def create_host_lists(self, count):
76 """ Method to create required number of MAC and IPv4 addresses.
77 Create required number of host MAC addresses and distribute them among
78 interfaces. Create host IPv4 address for every host MAC address too.
80 :param count: Number of hosts to create MAC and IPv4 addresses for.
82 for pg_if in self.pg_interfaces:
83 # self.MY_MACS[i.sw_if_index] = []
84 # self.MY_IP4S[i.sw_if_index] = []
85 self.hosts_by_pg_idx[pg_if.sw_if_index] = []
86 hosts = self.hosts_by_pg_idx[pg_if.sw_if_index]
87 for j in range(0, count):
89 "00:00:00:ff:%02x:%02x" % (pg_if.sw_if_index, j),
90 "172.17.1%02x.%u" % (pg_if.sw_if_index, j))
93 def create_stream(self, src_if, packet_sizes):
95 for i in range(0, TestL2xc.pkts_per_burst):
96 dst_if = self.flows[src_if][0]
97 dst_host = random.choice(self.hosts_by_pg_idx[dst_if.sw_if_index])
98 src_host = random.choice(self.hosts_by_pg_idx[src_if.sw_if_index])
99 pkt_info = self.create_packet_info(
100 src_if.sw_if_index, dst_if.sw_if_index)
101 payload = self.info_to_payload(pkt_info)
102 p = (Ether(dst=dst_host.mac, src=src_host.mac) /
103 IP(src=src_host.ip4, dst=dst_host.ip4) /
104 UDP(sport=1234, dport=1234) /
106 pkt_info.data = p.copy()
107 size = packet_sizes[(i / 2) % len(packet_sizes)]
108 self.extend_packet(p, size)
112 def verify_capture(self, pg_if, capture):
114 for i in self.interfaces:
115 last_info[i.sw_if_index] = None
116 dst_sw_if_index = pg_if.sw_if_index
117 for packet in capture:
121 payload_info = self.payload_to_info(str(packet[Raw]))
122 packet_index = payload_info.index
123 self.assertEqual(payload_info.dst, dst_sw_if_index)
124 debug("Got packet on port %s: src=%u (id=%u)" %
125 (pg_if.name, payload_info.src, packet_index))
126 next_info = self.get_next_packet_info_for_interface2(
127 payload_info.src, dst_sw_if_index,
128 last_info[payload_info.src])
129 last_info[payload_info.src] = next_info
130 self.assertTrue(next_info is not None)
131 self.assertEqual(packet_index, next_info.index)
132 saved_packet = next_info.data
133 # Check standard fields
134 self.assertEqual(ip.src, saved_packet[IP].src)
135 self.assertEqual(ip.dst, saved_packet[IP].dst)
136 self.assertEqual(udp.sport, saved_packet[UDP].sport)
137 self.assertEqual(udp.dport, saved_packet[UDP].dport)
139 error("Unexpected or invalid packet:")
142 for i in self.interfaces:
143 remaining_packet = self.get_next_packet_info_for_interface2(
144 i, dst_sw_if_index, last_info[i.sw_if_index])
145 self.assertTrue(remaining_packet is None,
146 "Port %u: Packet expected from source %u didn't"
147 " arrive" % (dst_sw_if_index, i.sw_if_index))
154 2 pairs of 2 interfaces, l2xconnected
155 2. sending l2 eth packets between 4 interfaces
156 64B, 512B, 1518B, 9018B (ether_size)
157 burst of packets per interface
160 # Create incoming packet streams for packet-generator interfaces
161 for i in self.interfaces:
162 pkts = self.create_stream(i, self.pg_if_packet_sizes)
165 # Enable packet capturing and start packet sending
166 self.pg_enable_capture(self.pg_interfaces)
169 # Verify outgoing packet streams per packet-generator interface
170 for i in self.pg_interfaces:
171 capture = i.get_capture()
172 info("Verifying capture on interface %s" % i.name)
173 self.verify_capture(i, capture)
176 if __name__ == '__main__':
177 unittest.main(testRunner=VppTestRunner)