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 self.logger.info(self.vapi.ppcli("show l2fib verbose"))
88 self.logger.info(self.vapi.ppcli("show bridge-domain %s detail" %
92 def create_hosts_and_learn(cls, count):
93 for pg_if in cls.pg_interfaces:
94 cls.hosts_by_pg_idx[pg_if.sw_if_index] = []
95 hosts = cls.hosts_by_pg_idx[pg_if.sw_if_index]
97 for j in range(1, count + 1):
99 "00:00:00:ff:%02x:%02x" % (pg_if.sw_if_index, j),
100 "172.17.1%02x.%u" % (pg_if.sw_if_index, j))
101 packet = (Ether(dst="ff:ff:ff:ff:ff:ff", src=host.mac))
103 if hasattr(pg_if, 'sub_if'):
104 packet = pg_if.sub_if.add_dot1_layer(packet)
105 packets.append(packet)
106 pg_if.add_stream(packets)
107 cls.logger.info("Sending broadcast eth frames for MAC learning")
108 cls.pg_enable_capture(cls.pg_interfaces)
111 def create_packet(self, src_if, dst_if, do_dot1=True):
112 packet_sizes = [64, 512, 1518, 9018]
113 dst_host = random.choice(self.hosts_by_pg_idx[dst_if.sw_if_index])
114 src_host = random.choice(self.hosts_by_pg_idx[src_if.sw_if_index])
115 pkt_info = self.create_packet_info(src_if, dst_if)
116 payload = self.info_to_payload(pkt_info)
117 p = (Ether(dst=dst_host.mac, src=src_host.mac) /
118 IP(src=src_host.ip4, dst=dst_host.ip4) /
119 UDP(sport=1234, dport=1234) /
121 pkt_info.data = p.copy()
122 if do_dot1 and hasattr(src_if, 'sub_if'):
123 p = src_if.sub_if.add_dot1_layer(p)
124 size = random.choice(packet_sizes)
125 self.extend_packet(p, size)
128 def _add_tag(self, packet, vlan, tag_type):
129 payload = packet.payload
130 inner_type = packet.type
131 packet.remove_payload()
132 packet.add_payload(Dot1Q(vlan=vlan) / payload)
133 packet.payload.type = inner_type
134 packet.payload.vlan = vlan
135 packet.type = tag_type
138 def _remove_tag(self, packet, vlan=None, tag_type=None):
140 self.assertEqual(packet.type, tag_type)
142 payload = packet.payload
144 self.assertEqual(payload.vlan, vlan)
145 inner_type = payload.type
146 payload = payload.payload
147 packet.remove_payload()
148 packet.add_payload(payload)
149 packet.type = inner_type
151 def add_tags(self, packet, tags):
152 for t in reversed(tags):
153 self._add_tag(packet, t.vlan, t.dot1)
155 def remove_tags(self, packet, tags):
157 self._remove_tag(packet, t.vlan, t.dot1)
159 def vtr_test(self, swif, tags):
160 p = self.create_packet(swif, self.pg0)
162 self.pg_enable_capture(self.pg_interfaces)
164 rx = self.pg0.get_capture(1)
167 self.remove_tags(rx[0], tags)
168 self.assertTrue(Dot1Q not in rx[0])
173 i = VppDot1QSubint(self, self.pg0, tags[0].vlan)
174 self.vapi.sw_interface_set_l2_bridge(rx_sw_if_index=i.sw_if_index,
175 bd_id=self.bd_id, enable=1)
178 p = self.create_packet(self.pg0, swif, do_dot1=False)
179 self.add_tags(p, tags)
180 self.pg0.add_stream(p)
181 self.pg_enable_capture(self.pg_interfaces)
183 rx = swif.get_capture(1)
184 swif.sub_if.remove_dot1_layer(rx[0])
185 self.assertTrue(Dot1Q not in rx[0])
187 self.vapi.sw_interface_set_l2_bridge(rx_sw_if_index=i.sw_if_index,
188 bd_id=self.bd_id, enable=0)
189 i.remove_vpp_config()
191 def test_1ad_vtr_pop_1(self):
192 """ 1AD VTR pop 1 test
194 self.pg1.sub_if.set_vtr(L2_VTR_OP.L2_POP_1)
195 self.vtr_test(self.pg1, [Tag(dot1=DOT1Q, vlan=100)])
197 def test_1ad_vtr_pop_2(self):
198 """ 1AD VTR pop 2 test
200 self.pg1.sub_if.set_vtr(L2_VTR_OP.L2_POP_2)
201 self.vtr_test(self.pg1, [])
203 def test_1ad_vtr_push_1ad(self):
204 """ 1AD VTR push 1 1AD test
206 self.pg1.sub_if.set_vtr(L2_VTR_OP.L2_PUSH_1, tag=300)
207 self.vtr_test(self.pg1, [Tag(dot1=DOT1AD, vlan=300),
208 Tag(dot1=DOT1AD, vlan=200),
209 Tag(dot1=DOT1Q, vlan=100)])
211 def test_1ad_vtr_push_2ad(self):
212 """ 1AD VTR push 2 1AD test
214 self.pg1.sub_if.set_vtr(L2_VTR_OP.L2_PUSH_2, outer=400, inner=300)
215 self.vtr_test(self.pg1, [Tag(dot1=DOT1AD, vlan=400),
216 Tag(dot1=DOT1Q, vlan=300),
217 Tag(dot1=DOT1AD, vlan=200),
218 Tag(dot1=DOT1Q, vlan=100)])
220 def test_1ad_vtr_push_1q(self):
221 """ 1AD VTR push 1 1Q test
223 self.pg1.sub_if.set_vtr(L2_VTR_OP.L2_PUSH_1, tag=300, push1q=1)
224 self.vtr_test(self.pg1, [Tag(dot1=DOT1Q, vlan=300),
225 Tag(dot1=DOT1AD, vlan=200),
226 Tag(dot1=DOT1Q, vlan=100)])
228 def test_1ad_vtr_push_2q(self):
229 """ 1AD VTR push 2 1Q test
231 self.pg1.sub_if.set_vtr(L2_VTR_OP.L2_PUSH_2,
232 outer=400, inner=300, push1q=1)
233 self.vtr_test(self.pg1, [Tag(dot1=DOT1Q, vlan=400),
234 Tag(dot1=DOT1Q, vlan=300),
235 Tag(dot1=DOT1AD, vlan=200),
236 Tag(dot1=DOT1Q, vlan=100)])
238 def test_1ad_vtr_translate_1_1ad(self):
239 """ 1AD VTR translate 1 -> 1 1AD test
241 self.pg1.sub_if.set_vtr(L2_VTR_OP.L2_TRANSLATE_1_1, tag=300)
242 self.vtr_test(self.pg1, [Tag(dot1=DOT1AD, vlan=300),
243 Tag(dot1=DOT1Q, vlan=100)])
245 def test_1ad_vtr_translate_1_2ad(self):
246 """ 1AD VTR translate 1 -> 2 1AD test
248 self.pg1.sub_if.set_vtr(
249 L2_VTR_OP.L2_TRANSLATE_1_2, inner=300, outer=400)
250 self.vtr_test(self.pg1, [Tag(dot1=DOT1AD, vlan=400),
251 Tag(dot1=DOT1Q, vlan=300),
252 Tag(dot1=DOT1Q, vlan=100)])
254 def test_1ad_vtr_translate_2_1ad(self):
255 """ 1AD VTR translate 2 -> 1 1AD test
257 self.pg1.sub_if.set_vtr(L2_VTR_OP.L2_TRANSLATE_2_1, tag=300)
258 self.vtr_test(self.pg1, [Tag(dot1=DOT1AD, vlan=300)])
260 def test_1ad_vtr_translate_2_2ad(self):
261 """ 1AD VTR translate 2 -> 2 1AD test
263 self.pg1.sub_if.set_vtr(
264 L2_VTR_OP.L2_TRANSLATE_2_2, inner=300, outer=400)
265 self.vtr_test(self.pg1, [Tag(dot1=DOT1AD, vlan=400),
266 Tag(dot1=DOT1Q, vlan=300)])
268 def test_1ad_vtr_translate_1_1q(self):
269 """ 1AD VTR translate 1 -> 1 1Q test
271 self.pg1.sub_if.set_vtr(L2_VTR_OP.L2_TRANSLATE_1_1, tag=300, push1q=1)
272 self.vtr_test(self.pg1, [Tag(dot1=DOT1Q, vlan=300),
273 Tag(dot1=DOT1Q, vlan=100)])
275 def test_1ad_vtr_translate_1_2q(self):
276 """ 1AD VTR translate 1 -> 2 1Q test
278 self.pg1.sub_if.set_vtr(
279 L2_VTR_OP.L2_TRANSLATE_1_2, inner=300, outer=400, push1q=1)
280 self.vtr_test(self.pg1, [Tag(dot1=DOT1Q, vlan=400),
281 Tag(dot1=DOT1Q, vlan=300),
282 Tag(dot1=DOT1Q, vlan=100)])
284 def test_1ad_vtr_translate_2_1q(self):
285 """ 1AD VTR translate 2 -> 1 1Q test
287 self.pg1.sub_if.set_vtr(L2_VTR_OP.L2_TRANSLATE_2_1, tag=300, push1q=1)
288 self.vtr_test(self.pg1, [Tag(dot1=DOT1Q, vlan=300)])
290 def test_1ad_vtr_translate_2_2q(self):
291 """ 1AD VTR translate 2 -> 2 1Q test
293 self.pg1.sub_if.set_vtr(
294 L2_VTR_OP.L2_TRANSLATE_2_2, inner=300, outer=400, push1q=1)
295 self.vtr_test(self.pg1, [Tag(dot1=DOT1Q, vlan=400),
296 Tag(dot1=DOT1Q, vlan=300)])
298 def test_1q_vtr_pop_1(self):
299 """ 1Q VTR pop 1 test
301 self.pg2.sub_if.set_vtr(L2_VTR_OP.L2_POP_1)
302 self.vtr_test(self.pg2, [])
304 def test_1q_vtr_push_1(self):
305 """ 1Q VTR push 1 test
307 self.pg2.sub_if.set_vtr(L2_VTR_OP.L2_PUSH_1, tag=300)
308 self.vtr_test(self.pg2, [Tag(dot1=DOT1AD, vlan=300),
309 Tag(dot1=DOT1Q, vlan=200)])
311 def test_1q_vtr_push_2(self):
312 """ 1Q VTR push 2 test
314 self.pg2.sub_if.set_vtr(L2_VTR_OP.L2_PUSH_2, outer=400, inner=300)
315 self.vtr_test(self.pg2, [Tag(dot1=DOT1AD, vlan=400),
316 Tag(dot1=DOT1Q, vlan=300),
317 Tag(dot1=DOT1Q, vlan=200)])
319 def test_1q_vtr_translate_1_1(self):
320 """ 1Q VTR translate 1 -> 1 test
322 self.pg2.sub_if.set_vtr(L2_VTR_OP.L2_TRANSLATE_1_1, tag=300)
323 self.vtr_test(self.pg2, [Tag(dot1=DOT1AD, vlan=300)])
325 def test_1q_vtr_translate_1_2(self):
326 """ 1Q VTR translate 1 -> 2 test
328 self.pg2.sub_if.set_vtr(
329 L2_VTR_OP.L2_TRANSLATE_1_2, inner=300, outer=400)
330 self.vtr_test(self.pg2, [Tag(dot1=DOT1AD, vlan=400),
331 Tag(dot1=DOT1Q, vlan=300)])
334 if __name__ == '__main__':
335 unittest.main(testRunner=VppTestRunner)