6 from scapy.packet import Raw
7 from scapy.layers.l2 import Ether, Dot1Q
8 from scapy.layers.inet import IP, UDP
11 from framework import VppTestCase, VppTestRunner
12 from vpp_sub_interface import L2_VTR_OP, VppDot1QSubint, VppDot1ADSubint
13 from collections import namedtuple
15 Tag = namedtuple('Tag', ['dot1', 'vlan'])
20 class TestVtr(VppTestCase):
25 super(TestVtr, cls).setUpClass()
29 cls.mac_entries_count = 5
32 cls.dot1ad_sub_id = 20
36 cls.create_pg_interfaces(ifs)
38 cls.sub_interfaces = [
39 VppDot1ADSubint(cls, cls.pg1, cls.dot1ad_sub_id,
41 VppDot1QSubint(cls, cls.pg2, cls.Btag)]
43 interfaces = list(cls.pg_interfaces)
44 interfaces.extend(cls.sub_interfaces)
46 # Create BD with MAC learning enabled and put interfaces and
47 # sub-interfaces to this BD
48 for pg_if in cls.pg_interfaces:
49 sw_if_index = pg_if.sub_if.sw_if_index \
50 if hasattr(pg_if, 'sub_if') else pg_if.sw_if_index
51 cls.vapi.sw_interface_set_l2_bridge(rx_sw_if_index=sw_if_index,
54 # setup all interfaces
58 # mapping between packet-generator index and lists of test hosts
59 cls.hosts_by_pg_idx = dict()
61 # create test host entries and inject packets to learn MAC entries
62 # in the bridge-domain
63 cls.create_hosts_and_learn(cls.mac_entries_count)
64 cls.logger.info(cls.vapi.ppcli("show l2fib"))
67 super(TestVtr, cls).tearDownClass()
71 def tearDownClass(cls):
72 super(TestVtr, cls).tearDownClass()
76 Clear trace and packet infos before running each test.
78 super(TestVtr, self).setUp()
79 self.reset_packet_infos()
83 Show various debug prints after each test.
85 super(TestVtr, self).tearDown()
87 def show_commands_at_teardown(self):
88 self.logger.info(self.vapi.ppcli("show l2fib verbose"))
89 self.logger.info(self.vapi.ppcli("show bridge-domain %s detail" %
93 def create_hosts_and_learn(cls, count):
94 for pg_if in cls.pg_interfaces:
95 cls.hosts_by_pg_idx[pg_if.sw_if_index] = []
96 hosts = cls.hosts_by_pg_idx[pg_if.sw_if_index]
98 for j in range(1, count + 1):
100 "00:00:00:ff:%02x:%02x" % (pg_if.sw_if_index, j),
101 "172.17.1%02x.%u" % (pg_if.sw_if_index, j))
102 packet = (Ether(dst="ff:ff:ff:ff:ff:ff", src=host.mac))
104 if hasattr(pg_if, 'sub_if'):
105 packet = pg_if.sub_if.add_dot1_layer(packet)
106 packets.append(packet)
107 pg_if.add_stream(packets)
108 cls.logger.info("Sending broadcast eth frames for MAC learning")
109 cls.pg_enable_capture(cls.pg_interfaces)
112 def create_packet(self, src_if, dst_if, do_dot1=True):
113 packet_sizes = [64, 512, 1518, 9018]
114 dst_host = random.choice(self.hosts_by_pg_idx[dst_if.sw_if_index])
115 src_host = random.choice(self.hosts_by_pg_idx[src_if.sw_if_index])
116 pkt_info = self.create_packet_info(src_if, dst_if)
117 payload = self.info_to_payload(pkt_info)
118 p = (Ether(dst=dst_host.mac, src=src_host.mac) /
119 IP(src=src_host.ip4, dst=dst_host.ip4) /
120 UDP(sport=1234, dport=1234) /
122 pkt_info.data = p.copy()
123 if do_dot1 and hasattr(src_if, 'sub_if'):
124 p = src_if.sub_if.add_dot1_layer(p)
125 size = random.choice(packet_sizes)
126 self.extend_packet(p, size)
129 def _add_tag(self, packet, vlan, tag_type):
130 payload = packet.payload
131 inner_type = packet.type
132 packet.remove_payload()
133 packet.add_payload(Dot1Q(vlan=vlan) / payload)
134 packet.payload.type = inner_type
135 packet.payload.vlan = vlan
136 packet.type = tag_type
139 def _remove_tag(self, packet, vlan=None, tag_type=None):
141 self.assertEqual(packet.type, tag_type)
143 payload = packet.payload
145 self.assertEqual(payload.vlan, vlan)
146 inner_type = payload.type
147 payload = payload.payload
148 packet.remove_payload()
149 packet.add_payload(payload)
150 packet.type = inner_type
152 def add_tags(self, packet, tags):
153 for t in reversed(tags):
154 self._add_tag(packet, t.vlan, t.dot1)
156 def remove_tags(self, packet, tags):
158 self._remove_tag(packet, t.vlan, t.dot1)
160 def vtr_test(self, swif, tags):
161 p = self.create_packet(swif, self.pg0)
163 self.pg_enable_capture(self.pg_interfaces)
165 rx = self.pg0.get_capture(1)
168 self.remove_tags(rx[0], tags)
169 self.assertTrue(Dot1Q not in rx[0])
174 i = VppDot1QSubint(self, self.pg0, tags[0].vlan)
175 self.vapi.sw_interface_set_l2_bridge(rx_sw_if_index=i.sw_if_index,
176 bd_id=self.bd_id, enable=1)
179 p = self.create_packet(self.pg0, swif, do_dot1=False)
180 self.add_tags(p, tags)
181 self.pg0.add_stream(p)
182 self.pg_enable_capture(self.pg_interfaces)
184 rx = swif.get_capture(1)
185 swif.sub_if.remove_dot1_layer(rx[0])
186 self.assertTrue(Dot1Q not in rx[0])
188 self.vapi.sw_interface_set_l2_bridge(rx_sw_if_index=i.sw_if_index,
189 bd_id=self.bd_id, enable=0)
190 i.remove_vpp_config()
192 def test_1ad_vtr_pop_1(self):
193 """ 1AD VTR pop 1 test
195 self.pg1.sub_if.set_vtr(L2_VTR_OP.L2_POP_1)
196 self.vtr_test(self.pg1, [Tag(dot1=DOT1Q, vlan=100)])
198 def test_1ad_vtr_pop_2(self):
199 """ 1AD VTR pop 2 test
201 self.pg1.sub_if.set_vtr(L2_VTR_OP.L2_POP_2)
202 self.vtr_test(self.pg1, [])
204 def test_1ad_vtr_push_1ad(self):
205 """ 1AD VTR push 1 1AD test
207 self.pg1.sub_if.set_vtr(L2_VTR_OP.L2_PUSH_1, tag=300)
208 self.vtr_test(self.pg1, [Tag(dot1=DOT1AD, vlan=300),
209 Tag(dot1=DOT1AD, vlan=200),
210 Tag(dot1=DOT1Q, vlan=100)])
212 def test_1ad_vtr_push_2ad(self):
213 """ 1AD VTR push 2 1AD test
215 self.pg1.sub_if.set_vtr(L2_VTR_OP.L2_PUSH_2, outer=400, inner=300)
216 self.vtr_test(self.pg1, [Tag(dot1=DOT1AD, vlan=400),
217 Tag(dot1=DOT1Q, vlan=300),
218 Tag(dot1=DOT1AD, vlan=200),
219 Tag(dot1=DOT1Q, vlan=100)])
221 def test_1ad_vtr_push_1q(self):
222 """ 1AD VTR push 1 1Q test
224 self.pg1.sub_if.set_vtr(L2_VTR_OP.L2_PUSH_1, tag=300, push1q=1)
225 self.vtr_test(self.pg1, [Tag(dot1=DOT1Q, vlan=300),
226 Tag(dot1=DOT1AD, vlan=200),
227 Tag(dot1=DOT1Q, vlan=100)])
229 def test_1ad_vtr_push_2q(self):
230 """ 1AD VTR push 2 1Q test
232 self.pg1.sub_if.set_vtr(L2_VTR_OP.L2_PUSH_2,
233 outer=400, inner=300, push1q=1)
234 self.vtr_test(self.pg1, [Tag(dot1=DOT1Q, vlan=400),
235 Tag(dot1=DOT1Q, vlan=300),
236 Tag(dot1=DOT1AD, vlan=200),
237 Tag(dot1=DOT1Q, vlan=100)])
239 def test_1ad_vtr_translate_1_1ad(self):
240 """ 1AD VTR translate 1 -> 1 1AD test
242 self.pg1.sub_if.set_vtr(L2_VTR_OP.L2_TRANSLATE_1_1, tag=300)
243 self.vtr_test(self.pg1, [Tag(dot1=DOT1AD, vlan=300),
244 Tag(dot1=DOT1Q, vlan=100)])
246 def test_1ad_vtr_translate_1_2ad(self):
247 """ 1AD VTR translate 1 -> 2 1AD test
249 self.pg1.sub_if.set_vtr(
250 L2_VTR_OP.L2_TRANSLATE_1_2, inner=300, outer=400)
251 self.vtr_test(self.pg1, [Tag(dot1=DOT1AD, vlan=400),
252 Tag(dot1=DOT1Q, vlan=300),
253 Tag(dot1=DOT1Q, vlan=100)])
255 def test_1ad_vtr_translate_2_1ad(self):
256 """ 1AD VTR translate 2 -> 1 1AD test
258 self.pg1.sub_if.set_vtr(L2_VTR_OP.L2_TRANSLATE_2_1, tag=300)
259 self.vtr_test(self.pg1, [Tag(dot1=DOT1AD, vlan=300)])
261 def test_1ad_vtr_translate_2_2ad(self):
262 """ 1AD VTR translate 2 -> 2 1AD test
264 self.pg1.sub_if.set_vtr(
265 L2_VTR_OP.L2_TRANSLATE_2_2, inner=300, outer=400)
266 self.vtr_test(self.pg1, [Tag(dot1=DOT1AD, vlan=400),
267 Tag(dot1=DOT1Q, vlan=300)])
269 def test_1ad_vtr_translate_1_1q(self):
270 """ 1AD VTR translate 1 -> 1 1Q test
272 self.pg1.sub_if.set_vtr(L2_VTR_OP.L2_TRANSLATE_1_1, tag=300, push1q=1)
273 self.vtr_test(self.pg1, [Tag(dot1=DOT1Q, vlan=300),
274 Tag(dot1=DOT1Q, vlan=100)])
276 def test_1ad_vtr_translate_1_2q(self):
277 """ 1AD VTR translate 1 -> 2 1Q test
279 self.pg1.sub_if.set_vtr(
280 L2_VTR_OP.L2_TRANSLATE_1_2, inner=300, outer=400, push1q=1)
281 self.vtr_test(self.pg1, [Tag(dot1=DOT1Q, vlan=400),
282 Tag(dot1=DOT1Q, vlan=300),
283 Tag(dot1=DOT1Q, vlan=100)])
285 def test_1ad_vtr_translate_2_1q(self):
286 """ 1AD VTR translate 2 -> 1 1Q test
288 self.pg1.sub_if.set_vtr(L2_VTR_OP.L2_TRANSLATE_2_1, tag=300, push1q=1)
289 self.vtr_test(self.pg1, [Tag(dot1=DOT1Q, vlan=300)])
291 def test_1ad_vtr_translate_2_2q(self):
292 """ 1AD VTR translate 2 -> 2 1Q test
294 self.pg1.sub_if.set_vtr(
295 L2_VTR_OP.L2_TRANSLATE_2_2, inner=300, outer=400, push1q=1)
296 self.vtr_test(self.pg1, [Tag(dot1=DOT1Q, vlan=400),
297 Tag(dot1=DOT1Q, vlan=300)])
299 def test_1q_vtr_pop_1(self):
300 """ 1Q VTR pop 1 test
302 self.pg2.sub_if.set_vtr(L2_VTR_OP.L2_POP_1)
303 self.vtr_test(self.pg2, [])
305 def test_1q_vtr_push_1(self):
306 """ 1Q VTR push 1 test
308 self.pg2.sub_if.set_vtr(L2_VTR_OP.L2_PUSH_1, tag=300)
309 self.vtr_test(self.pg2, [Tag(dot1=DOT1AD, vlan=300),
310 Tag(dot1=DOT1Q, vlan=200)])
312 def test_1q_vtr_push_2(self):
313 """ 1Q VTR push 2 test
315 self.pg2.sub_if.set_vtr(L2_VTR_OP.L2_PUSH_2, outer=400, inner=300)
316 self.vtr_test(self.pg2, [Tag(dot1=DOT1AD, vlan=400),
317 Tag(dot1=DOT1Q, vlan=300),
318 Tag(dot1=DOT1Q, vlan=200)])
320 def test_1q_vtr_translate_1_1(self):
321 """ 1Q VTR translate 1 -> 1 test
323 self.pg2.sub_if.set_vtr(L2_VTR_OP.L2_TRANSLATE_1_1, tag=300)
324 self.vtr_test(self.pg2, [Tag(dot1=DOT1AD, vlan=300)])
326 def test_1q_vtr_translate_1_2(self):
327 """ 1Q VTR translate 1 -> 2 test
329 self.pg2.sub_if.set_vtr(
330 L2_VTR_OP.L2_TRANSLATE_1_2, inner=300, outer=400)
331 self.vtr_test(self.pg2, [Tag(dot1=DOT1AD, vlan=400),
332 Tag(dot1=DOT1Q, vlan=300)])
335 if __name__ == '__main__':
336 unittest.main(testRunner=VppTestRunner)