3 # Module to provide L2 cross-connect test case.
5 # The module provides a set of tools for L2 cross-connect tests.
8 logging.getLogger("scapy.runtime").setLevel(logging.ERROR)
12 from framework import VppTestCase, VppTestRunner
13 from scapy.layers.l2 import Ether, Raw
14 from scapy.layers.inet import IP, UDP
17 ## Subclass of the VppTestCase class.
19 # This subclass is a class for L2 cross-connect test cases. It provides methods
20 # to create interfaces, configuring L2 cross-connects, creating and verifying
22 class TestL2xc(VppTestCase):
23 """ L2XC Test Case """
26 interf_nr = 4 # Number of interfaces
27 hosts_nr = 10 # Number of hosts
28 pkts_per_burst = 257 # Number of packets per burst
30 ## Class method to start the test case.
31 # Overrides setUpClass method in VppTestCase class.
32 # There is used try..except statement to ensure that the tear down of
33 # the class will be executed even if any exception is raised.
34 # @param cls The class pointer.
37 super(TestL2xc, cls).setUpClass()
41 cls.interfaces = range(TestL2xc.interf_nr)
42 cls.create_interfaces(cls.interfaces)
44 ## Create bi-directional cross-connects between pg0 and pg1
45 cls.api("sw_interface_set_l2_xconnect rx pg0 tx pg1 enable")
46 cls.api("sw_interface_set_l2_xconnect rx pg1 tx pg0 enable")
48 ## Create bi-directional cross-connects between pg2 and pg3
49 cls.api("sw_interface_set_l2_xconnect rx pg2 tx pg3 enable")
50 cls.api("sw_interface_set_l2_xconnect rx pg3 tx pg2 enable")
52 cls.cli(0, "show l2patch")
54 ## Create host MAC and IPv4 lists
55 cls.create_host_lists(TestL2xc.hosts_nr)
57 except Exception as e:
61 ## Method to define tear down VPP actions of the test case.
62 # Overrides tearDown method in VppTestCase class.
63 # @param self The object pointer.
65 self.cli(2, "show int")
66 self.cli(2, "show trace")
67 self.cli(2, "show hardware")
68 self.cli(2, "show l2patch")
69 self.cli(2, "show error")
70 self.cli(2, "show run")
72 ## Class method to create required number of MAC and IPv4 addresses.
73 # Create required number of host MAC addresses and distribute them among
74 # interfaces. Create host IPv4 address for every host MAC address too.
75 # @param cls The class pointer.
76 # @param count Integer variable to store the number of MAC addresses to be
79 def create_host_lists(cls, count):
80 for i in cls.interfaces:
83 for j in range(0, count):
84 cls.MY_MACS[i].append("00:00:00:ff:%02x:%02x" % (i, j))
85 cls.MY_IP4S[i].append("172.17.1%02x.%u" % (i, j))
87 # Dictionary variable to store list of MAC addresses per interface.
89 # Dictionary variable to store list of IPv4 addresses per interface.
91 ## Method to create packet stream for the packet generator interface.
92 # Create input packet stream for the given packet generator interface with
93 # packets of different length targeted for all other created packet
94 # generator interfaces.
95 # @param self The object pointer.
96 # @param pg_id Integer variable to store the index of the interface to
97 # create the input packet stream.
98 # @return pkts List variable to store created input stream of packets.
99 def create_stream(self, pg_id):
100 # TODO: use variables to create lists based on interface number
101 pg_targets = [None] * 4
107 for i in range(0, TestL2xc.pkts_per_burst):
108 target_pg_id = pg_targets[pg_id][0]
109 target_host_id = random.randrange(len(self.MY_MACS[target_pg_id]))
110 source_host_id = random.randrange(len(self.MY_MACS[pg_id]))
111 pkt_info = self.create_packet_info(pg_id, target_pg_id)
112 payload = self.info_to_payload(pkt_info)
113 p = (Ether(dst=self.MY_MACS[target_pg_id][target_host_id],
114 src=self.MY_MACS[pg_id][source_host_id]) /
115 IP(src=self.MY_IP4S[pg_id][source_host_id],
116 dst=self.MY_IP4S[target_pg_id][target_host_id]) /
117 UDP(sport=1234, dport=1234) /
119 pkt_info.data = p.copy()
120 packet_sizes = [64, 512, 1518, 9018]
121 size = packet_sizes[(i / 2) % len(packet_sizes)]
122 self.extend_packet(p, size)
126 # List variable to store list of indexes of target packet generator
127 # interfaces for every source packet generator interface.
129 # Integer variable to store the index of the random target packet
130 # generator interfaces.
131 ## @var target_host_id
132 # Integer variable to store the index of the randomly chosen
133 # destination host MAC/IPv4 address.
134 ## @var source_host_id
135 # Integer variable to store the index of the randomly chosen source
136 # host MAC/IPv4 address.
138 # Object variable to store the information about the generated packet.
140 # String variable to store the payload of the packet to be generated.
142 # Object variable to store the generated packet.
144 # List variable to store required packet sizes.
146 # List variable to store required packet sizes.
148 ## Method to verify packet stream received on the packet generator interface.
149 # Verify packet-by-packet the output stream captured on a given packet
150 # generator (pg) interface using following packet payload data - order of
151 # packet in the stream, index of the source and destination pg interface,
152 # src and dst host IPv4 addresses and src port and dst port values of UDP
154 # @param self The object pointer.
155 # @param o Integer variable to store the index of the interface to
156 # verify the output packet stream.
157 # @param capture List variable to store the captured output packet stream.
158 def verify_capture(self, o, capture):
160 for i in self.interfaces:
162 for packet in capture:
166 payload_info = self.payload_to_info(str(packet[Raw]))
167 self.assertEqual(payload_info.dst, o)
168 self.log("Got packet on port %u: src=%u (id=%u)"
169 % (o, payload_info.src, payload_info.index), 2)
170 next_info = self.get_next_packet_info_for_interface2(
171 payload_info.src, payload_info.dst,
172 last_info[payload_info.src])
173 last_info[payload_info.src] = next_info
174 self.assertTrue(next_info is not None)
175 self.assertEqual(payload_info.index, next_info.index)
176 # Check standard fields
177 self.assertEqual(ip.src, next_info.data[IP].src)
178 self.assertEqual(ip.dst, next_info.data[IP].dst)
179 self.assertEqual(udp.sport, next_info.data[UDP].sport)
180 self.assertEqual(udp.dport, next_info.data[UDP].dport)
182 self.log("Unexpected or invalid packet:")
185 for i in self.interfaces:
186 remaining_packet = self.get_next_packet_info_for_interface2(
188 self.assertTrue(remaining_packet is None,
189 "Port %u: Packet expected from source %u didn't"
192 # Dictionary variable to store verified packets per packet generator
195 # Object variable to store the IP layer of the packet.
197 # Object variable to store the UDP layer of the packet.
199 # Object variable to store required information about the packet.
201 # Object variable to store information about next packet.
202 ## @var remaining_packet
203 # Object variable to store information about remaining packet.
205 ## Method defining L2 cross-connect test case.
206 # Contains steps of the test case.
207 # @param self The object pointer.
213 2 pairs of 2 interfaces, l2xconnected
215 2.sending l2 eth packets between 4 interfaces
216 64B, 512B, 1518B, 9018B (ether_size)
217 burst of packets per interface
220 ## Create incoming packet streams for packet-generator interfaces
221 for i in self.interfaces:
222 pkts = self.create_stream(i)
223 self.pg_add_stream(i, pkts)
225 ## Enable packet capturing and start packet sending
226 self.pg_enable_capture(self.interfaces)
229 ## Verify outgoing packet streams per packet-generator interface
230 for i in self.interfaces:
231 out = self.pg_get_capture(i)
232 self.log("Verifying capture %u" % i)
233 self.verify_capture(i, out)
235 # List variable to store created input stream of packets for the packet
236 # generator interface.
238 # List variable to store captured output stream of packets for
239 # the packet generator interface.
242 if __name__ == '__main__':
243 unittest.main(testRunner = VppTestRunner)