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()
72 Clear trace and packet infos before running each test.
74 super(TestVtr, self).setUp()
75 self.reset_packet_infos()
79 Show various debug prints after each test.
81 super(TestVtr, self).tearDown()
83 self.logger.info(self.vapi.ppcli("show l2fib verbose"))
84 self.logger.info(self.vapi.ppcli("show bridge-domain %s detail" %
88 def create_hosts_and_learn(cls, count):
89 for pg_if in cls.pg_interfaces:
90 cls.hosts_by_pg_idx[pg_if.sw_if_index] = []
91 hosts = cls.hosts_by_pg_idx[pg_if.sw_if_index]
93 for j in range(1, count + 1):
95 "00:00:00:ff:%02x:%02x" % (pg_if.sw_if_index, j),
96 "172.17.1%02x.%u" % (pg_if.sw_if_index, j))
97 packet = (Ether(dst="ff:ff:ff:ff:ff:ff", src=host.mac))
99 if hasattr(pg_if, 'sub_if'):
100 packet = pg_if.sub_if.add_dot1_layer(packet)
101 packets.append(packet)
102 pg_if.add_stream(packets)
103 cls.logger.info("Sending broadcast eth frames for MAC learning")
104 cls.pg_enable_capture(cls.pg_interfaces)
107 def create_packet(self, src_if, dst_if, do_dot1=True):
108 packet_sizes = [64, 512, 1518, 9018]
109 dst_host = random.choice(self.hosts_by_pg_idx[dst_if.sw_if_index])
110 src_host = random.choice(self.hosts_by_pg_idx[src_if.sw_if_index])
111 pkt_info = self.create_packet_info(src_if, dst_if)
112 payload = self.info_to_payload(pkt_info)
113 p = (Ether(dst=dst_host.mac, src=src_host.mac) /
114 IP(src=src_host.ip4, dst=dst_host.ip4) /
115 UDP(sport=1234, dport=1234) /
117 pkt_info.data = p.copy()
118 if do_dot1 and hasattr(src_if, 'sub_if'):
119 p = src_if.sub_if.add_dot1_layer(p)
120 size = random.choice(packet_sizes)
121 self.extend_packet(p, size)
124 def _add_tag(self, packet, vlan, tag_type):
125 payload = packet.payload
126 inner_type = packet.type
127 packet.remove_payload()
128 packet.add_payload(Dot1Q(vlan=vlan) / payload)
129 packet.payload.type = inner_type
130 packet.payload.vlan = vlan
131 packet.type = tag_type
134 def _remove_tag(self, packet, vlan=None, tag_type=None):
136 self.assertEqual(packet.type, tag_type)
138 payload = packet.payload
140 self.assertEqual(payload.vlan, vlan)
141 inner_type = payload.type
142 payload = payload.payload
143 packet.remove_payload()
144 packet.add_payload(payload)
145 packet.type = inner_type
147 def add_tags(self, packet, tags):
148 for t in reversed(tags):
149 self._add_tag(packet, t.vlan, t.dot1)
151 def remove_tags(self, packet, tags):
153 self._remove_tag(packet, t.vlan, t.dot1)
155 def vtr_test(self, swif, tags):
156 p = self.create_packet(swif, self.pg0)
158 self.pg_enable_capture(self.pg_interfaces)
160 rx = self.pg0.get_capture(1)
163 self.remove_tags(rx[0], tags)
164 self.assertTrue(Dot1Q not in rx[0])
169 i = VppDot1QSubint(self, self.pg0, tags[0].vlan)
170 self.vapi.sw_interface_set_l2_bridge(rx_sw_if_index=i.sw_if_index,
171 bd_id=self.bd_id, enable=1)
174 p = self.create_packet(self.pg0, swif, do_dot1=False)
175 self.add_tags(p, tags)
176 self.pg0.add_stream(p)
177 self.pg_enable_capture(self.pg_interfaces)
179 rx = swif.get_capture(1)
180 swif.sub_if.remove_dot1_layer(rx[0])
181 self.assertTrue(Dot1Q not in rx[0])
183 self.vapi.sw_interface_set_l2_bridge(rx_sw_if_index=i.sw_if_index,
184 bd_id=self.bd_id, enable=0)
185 i.remove_vpp_config()
187 def test_1ad_vtr_pop_1(self):
188 """ 1AD VTR pop 1 test
190 self.pg1.sub_if.set_vtr(L2_VTR_OP.L2_POP_1)
191 self.vtr_test(self.pg1, [Tag(dot1=DOT1Q, vlan=100)])
193 def test_1ad_vtr_pop_2(self):
194 """ 1AD VTR pop 2 test
196 self.pg1.sub_if.set_vtr(L2_VTR_OP.L2_POP_2)
197 self.vtr_test(self.pg1, [])
199 def test_1ad_vtr_push_1ad(self):
200 """ 1AD VTR push 1 1AD test
202 self.pg1.sub_if.set_vtr(L2_VTR_OP.L2_PUSH_1, tag=300)
203 self.vtr_test(self.pg1, [Tag(dot1=DOT1AD, vlan=300),
204 Tag(dot1=DOT1AD, vlan=200),
205 Tag(dot1=DOT1Q, vlan=100)])
207 def test_1ad_vtr_push_2ad(self):
208 """ 1AD VTR push 2 1AD test
210 self.pg1.sub_if.set_vtr(L2_VTR_OP.L2_PUSH_2, outer=400, inner=300)
211 self.vtr_test(self.pg1, [Tag(dot1=DOT1AD, vlan=400),
212 Tag(dot1=DOT1Q, vlan=300),
213 Tag(dot1=DOT1AD, vlan=200),
214 Tag(dot1=DOT1Q, vlan=100)])
216 def test_1ad_vtr_push_1q(self):
217 """ 1AD VTR push 1 1Q test
219 self.pg1.sub_if.set_vtr(L2_VTR_OP.L2_PUSH_1, tag=300, push1q=1)
220 self.vtr_test(self.pg1, [Tag(dot1=DOT1Q, vlan=300),
221 Tag(dot1=DOT1AD, vlan=200),
222 Tag(dot1=DOT1Q, vlan=100)])
224 def test_1ad_vtr_push_2q(self):
225 """ 1AD VTR push 2 1Q test
227 self.pg1.sub_if.set_vtr(L2_VTR_OP.L2_PUSH_2,
228 outer=400, inner=300, push1q=1)
229 self.vtr_test(self.pg1, [Tag(dot1=DOT1Q, vlan=400),
230 Tag(dot1=DOT1Q, vlan=300),
231 Tag(dot1=DOT1AD, vlan=200),
232 Tag(dot1=DOT1Q, vlan=100)])
234 def test_1ad_vtr_translate_1_1ad(self):
235 """ 1AD VTR translate 1 -> 1 1AD test
237 self.pg1.sub_if.set_vtr(L2_VTR_OP.L2_TRANSLATE_1_1, tag=300)
238 self.vtr_test(self.pg1, [Tag(dot1=DOT1AD, vlan=300),
239 Tag(dot1=DOT1Q, vlan=100)])
241 def test_1ad_vtr_translate_1_2ad(self):
242 """ 1AD VTR translate 1 -> 2 1AD test
244 self.pg1.sub_if.set_vtr(
245 L2_VTR_OP.L2_TRANSLATE_1_2, inner=300, outer=400)
246 self.vtr_test(self.pg1, [Tag(dot1=DOT1AD, vlan=400),
247 Tag(dot1=DOT1Q, vlan=300),
248 Tag(dot1=DOT1Q, vlan=100)])
250 def test_1ad_vtr_translate_2_1ad(self):
251 """ 1AD VTR translate 2 -> 1 1AD test
253 self.pg1.sub_if.set_vtr(L2_VTR_OP.L2_TRANSLATE_2_1, tag=300)
254 self.vtr_test(self.pg1, [Tag(dot1=DOT1AD, vlan=300)])
256 def test_1ad_vtr_translate_2_2ad(self):
257 """ 1AD VTR translate 2 -> 2 1AD test
259 self.pg1.sub_if.set_vtr(
260 L2_VTR_OP.L2_TRANSLATE_2_2, inner=300, outer=400)
261 self.vtr_test(self.pg1, [Tag(dot1=DOT1AD, vlan=400),
262 Tag(dot1=DOT1Q, vlan=300)])
264 def test_1ad_vtr_translate_1_1q(self):
265 """ 1AD VTR translate 1 -> 1 1Q test
267 self.pg1.sub_if.set_vtr(L2_VTR_OP.L2_TRANSLATE_1_1, tag=300, push1q=1)
268 self.vtr_test(self.pg1, [Tag(dot1=DOT1Q, vlan=300),
269 Tag(dot1=DOT1Q, vlan=100)])
271 def test_1ad_vtr_translate_1_2q(self):
272 """ 1AD VTR translate 1 -> 2 1Q test
274 self.pg1.sub_if.set_vtr(
275 L2_VTR_OP.L2_TRANSLATE_1_2, inner=300, outer=400, push1q=1)
276 self.vtr_test(self.pg1, [Tag(dot1=DOT1Q, vlan=400),
277 Tag(dot1=DOT1Q, vlan=300),
278 Tag(dot1=DOT1Q, vlan=100)])
280 def test_1ad_vtr_translate_2_1q(self):
281 """ 1AD VTR translate 2 -> 1 1Q test
283 self.pg1.sub_if.set_vtr(L2_VTR_OP.L2_TRANSLATE_2_1, tag=300, push1q=1)
284 self.vtr_test(self.pg1, [Tag(dot1=DOT1Q, vlan=300)])
286 def test_1ad_vtr_translate_2_2q(self):
287 """ 1AD VTR translate 2 -> 2 1Q test
289 self.pg1.sub_if.set_vtr(
290 L2_VTR_OP.L2_TRANSLATE_2_2, inner=300, outer=400, push1q=1)
291 self.vtr_test(self.pg1, [Tag(dot1=DOT1Q, vlan=400),
292 Tag(dot1=DOT1Q, vlan=300)])
294 def test_1q_vtr_pop_1(self):
295 """ 1Q VTR pop 1 test
297 self.pg2.sub_if.set_vtr(L2_VTR_OP.L2_POP_1)
298 self.vtr_test(self.pg2, [])
300 def test_1q_vtr_push_1(self):
301 """ 1Q VTR push 1 test
303 self.pg2.sub_if.set_vtr(L2_VTR_OP.L2_PUSH_1, tag=300)
304 self.vtr_test(self.pg2, [Tag(dot1=DOT1AD, vlan=300),
305 Tag(dot1=DOT1Q, vlan=200)])
307 def test_1q_vtr_push_2(self):
308 """ 1Q VTR push 2 test
310 self.pg2.sub_if.set_vtr(L2_VTR_OP.L2_PUSH_2, outer=400, inner=300)
311 self.vtr_test(self.pg2, [Tag(dot1=DOT1AD, vlan=400),
312 Tag(dot1=DOT1Q, vlan=300),
313 Tag(dot1=DOT1Q, vlan=200)])
315 def test_1q_vtr_translate_1_1(self):
316 """ 1Q VTR translate 1 -> 1 test
318 self.pg2.sub_if.set_vtr(L2_VTR_OP.L2_TRANSLATE_1_1, tag=300)
319 self.vtr_test(self.pg2, [Tag(dot1=DOT1AD, vlan=300)])
321 def test_1q_vtr_translate_1_2(self):
322 """ 1Q VTR translate 1 -> 2 test
324 self.pg2.sub_if.set_vtr(
325 L2_VTR_OP.L2_TRANSLATE_1_2, inner=300, outer=400)
326 self.vtr_test(self.pg2, [Tag(dot1=DOT1AD, vlan=400),
327 Tag(dot1=DOT1Q, vlan=300)])
330 if __name__ == '__main__':
331 unittest.main(testRunner=VppTestRunner)