lb: remove api boilerplate
[vpp.git] / test / test_vtr.py
1 #!/usr/bin/env python
2
3 import unittest
4 import random
5
6 from scapy.packet import Raw
7 from scapy.layers.l2 import Ether, Dot1Q
8 from scapy.layers.inet import IP, UDP
9
10 from util import Host
11 from framework import VppTestCase, VppTestRunner
12 from vpp_sub_interface import L2_VTR_OP, VppDot1QSubint, VppDot1ADSubint
13 from collections import namedtuple
14
15 Tag = namedtuple('Tag', ['dot1', 'vlan'])
16 DOT1AD = 0x88A8
17 DOT1Q = 0x8100
18
19
20 class TestVtr(VppTestCase):
21     """ VTR Test Case """
22
23     @classmethod
24     def setUpClass(cls):
25         super(TestVtr, cls).setUpClass()
26
27         # Test variables
28         cls.bd_id = 1
29         cls.mac_entries_count = 5
30         cls.Atag = 100
31         cls.Btag = 200
32         cls.dot1ad_sub_id = 20
33
34         try:
35             ifs = range(3)
36             cls.create_pg_interfaces(ifs)
37
38             cls.sub_interfaces = [
39                 VppDot1ADSubint(cls, cls.pg1, cls.dot1ad_sub_id,
40                                 cls.Btag, cls.Atag),
41                 VppDot1QSubint(cls, cls.pg2, cls.Btag)]
42
43             interfaces = list(cls.pg_interfaces)
44             interfaces.extend(cls.sub_interfaces)
45
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,
52                                                     bd_id=cls.bd_id)
53
54             # setup all interfaces
55             for i in interfaces:
56                 i.admin_up()
57
58             # mapping between packet-generator index and lists of test hosts
59             cls.hosts_by_pg_idx = dict()
60
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"))
65
66         except Exception:
67             super(TestVtr, cls).tearDownClass()
68             raise
69
70     @classmethod
71     def tearDownClass(cls):
72         super(TestVtr, cls).tearDownClass()
73
74     def setUp(self):
75         """
76         Clear trace and packet infos before running each test.
77         """
78         super(TestVtr, self).setUp()
79         self.reset_packet_infos()
80
81     def tearDown(self):
82         """
83         Show various debug prints after each test.
84         """
85         super(TestVtr, self).tearDown()
86
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" %
90                                          self.bd_id))
91
92     @classmethod
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]
97             packets = []
98             for j in range(1, count + 1):
99                 host = Host(
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))
103                 hosts.append(host)
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)
110         cls.pg_start()
111
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) /
121              Raw(payload))
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)
127         return p
128
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
137         return packet
138
139     def _remove_tag(self, packet, vlan=None, tag_type=None):
140         if tag_type:
141             self.assertEqual(packet.type, tag_type)
142
143         payload = packet.payload
144         if vlan:
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
151
152     def add_tags(self, packet, tags):
153         for t in reversed(tags):
154             self._add_tag(packet, t.vlan, t.dot1)
155
156     def remove_tags(self, packet, tags):
157         for t in tags:
158             self._remove_tag(packet, t.vlan, t.dot1)
159
160     def vtr_test(self, swif, tags):
161         p = self.create_packet(swif, self.pg0)
162         swif.add_stream(p)
163         self.pg_enable_capture(self.pg_interfaces)
164         self.pg_start()
165         rx = self.pg0.get_capture(1)
166
167         if tags:
168             self.remove_tags(rx[0], tags)
169         self.assertTrue(Dot1Q not in rx[0])
170
171         if not tags:
172             return
173
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)
177         i.admin_up()
178
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)
183         self.pg_start()
184         rx = swif.get_capture(1)
185         swif.sub_if.remove_dot1_layer(rx[0])
186         self.assertTrue(Dot1Q not in rx[0])
187
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()
191
192     def test_1ad_vtr_pop_1(self):
193         """ 1AD VTR pop 1 test
194         """
195         self.pg1.sub_if.set_vtr(L2_VTR_OP.L2_POP_1)
196         self.vtr_test(self.pg1, [Tag(dot1=DOT1Q, vlan=100)])
197
198     def test_1ad_vtr_pop_2(self):
199         """ 1AD VTR pop 2 test
200         """
201         self.pg1.sub_if.set_vtr(L2_VTR_OP.L2_POP_2)
202         self.vtr_test(self.pg1, [])
203
204     def test_1ad_vtr_push_1ad(self):
205         """ 1AD VTR push 1 1AD test
206         """
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)])
211
212     def test_1ad_vtr_push_2ad(self):
213         """ 1AD VTR push 2 1AD test
214         """
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)])
220
221     def test_1ad_vtr_push_1q(self):
222         """ 1AD VTR push 1 1Q test
223         """
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)])
228
229     def test_1ad_vtr_push_2q(self):
230         """ 1AD VTR push 2 1Q test
231         """
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)])
238
239     def test_1ad_vtr_translate_1_1ad(self):
240         """ 1AD VTR translate 1 -> 1 1AD test
241         """
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)])
245
246     def test_1ad_vtr_translate_1_2ad(self):
247         """ 1AD VTR translate 1 -> 2 1AD test
248         """
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)])
254
255     def test_1ad_vtr_translate_2_1ad(self):
256         """ 1AD VTR translate 2 -> 1 1AD test
257         """
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)])
260
261     def test_1ad_vtr_translate_2_2ad(self):
262         """ 1AD VTR translate 2 -> 2 1AD test
263         """
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)])
268
269     def test_1ad_vtr_translate_1_1q(self):
270         """ 1AD VTR translate 1 -> 1 1Q test
271         """
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)])
275
276     def test_1ad_vtr_translate_1_2q(self):
277         """ 1AD VTR translate 1 -> 2 1Q test
278         """
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)])
284
285     def test_1ad_vtr_translate_2_1q(self):
286         """ 1AD VTR translate 2 -> 1 1Q test
287         """
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)])
290
291     def test_1ad_vtr_translate_2_2q(self):
292         """ 1AD VTR translate 2 -> 2 1Q test
293         """
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)])
298
299     def test_1q_vtr_pop_1(self):
300         """ 1Q VTR pop 1 test
301         """
302         self.pg2.sub_if.set_vtr(L2_VTR_OP.L2_POP_1)
303         self.vtr_test(self.pg2, [])
304
305     def test_1q_vtr_push_1(self):
306         """ 1Q VTR push 1 test
307         """
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)])
311
312     def test_1q_vtr_push_2(self):
313         """ 1Q VTR push 2 test
314         """
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)])
319
320     def test_1q_vtr_translate_1_1(self):
321         """ 1Q VTR translate 1 -> 1 test
322         """
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)])
325
326     def test_1q_vtr_translate_1_2(self):
327         """ 1Q VTR translate 1 -> 2 test
328         """
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)])
333
334
335 if __name__ == '__main__':
336     unittest.main(testRunner=VppTestRunner)