test_qos.py revision ead1e536
1#!/usr/bin/env python3
2
3import unittest
4
5from framework import VppTestCase, VppTestRunner
6from vpp_sub_interface import VppDot1QSubint
7from vpp_ip import DpoProto
8from vpp_ip_route import VppIpRoute, VppRoutePath, VppMplsRoute, \
9    VppMplsLabel, VppMplsTable, FibPathProto
10
11import scapy.compat
12from scapy.packet import Raw
13from scapy.layers.l2 import Ether, Dot1Q
14from scapy.layers.inet import IP, UDP
15from scapy.layers.inet6 import IPv6
16from scapy.contrib.mpls import MPLS
17from vpp_papi import VppEnum
18from vpp_qos import VppQosRecord, VppQosEgressMap, VppQosMark, VppQosStore
19
20NUM_PKTS = 67
21
22
23class TestQOS(VppTestCase):
24    """ QOS Test Case """
25
26    # Note: Since the enums aren't created dynamically until after
27    #       the papi client attaches to VPP, we put it in a property to
28    #       ensure it is the value at runtime, not at module load time.
29    @property
30    def QOS_SOURCE(self):
31        return VppEnum.vl_api_qos_source_t
32
33    @classmethod
34    def setUpClass(cls):
35        super(TestQOS, cls).setUpClass()
36
37    @classmethod
38    def tearDownClass(cls):
39        super(TestQOS, cls).tearDownClass()
40
41    def setUp(self):
42        super(TestQOS, self).setUp()
43
44        self.create_pg_interfaces(range(5))
45
46        tbl = VppMplsTable(self, 0)
47        tbl.add_vpp_config()
48
49        for i in self.pg_interfaces:
50            i.admin_up()
51            i.config_ip4()
52            i.resolve_arp()
53            i.config_ip6()
54            i.resolve_ndp()
55            i.enable_mpls()
56
57    def tearDown(self):
58        for i in self.pg_interfaces:
59            i.unconfig_ip4()
60            i.unconfig_ip6()
61            i.disable_mpls()
62
63        super(TestQOS, self).tearDown()
64
65    def test_qos_ip(self):
66        """ QoS Mark/Record/Store IP """
67
68        #
69        # for table 1 map the n=0xff possible values of input QoS mark,
70        # n to 1-n
71        #
72        output = [scapy.compat.chb(0)] * 256
73        for i in range(0, 255):
74            output[i] = scapy.compat.chb(255 - i)
75        os = b''.join(output)
76        rows = [{'outputs': os},
77                {'outputs': os},
78                {'outputs': os},
79                {'outputs': os}]
80
81        qem1 = VppQosEgressMap(self, 1, rows).add_vpp_config()
82
83        #
84        # For table 2 (and up) use the value n for everything
85        #
86        output = [scapy.compat.chb(2)] * 256
87        os = b''.join(output)
88        rows = [{'outputs': os},
89                {'outputs': os},
90                {'outputs': os},
91                {'outputs': os}]
92
93        qem2 = VppQosEgressMap(self, 2, rows).add_vpp_config()
94
95        output = [scapy.compat.chb(3)] * 256
96        os = b''.join(output)
97        rows = [{'outputs': os},
98                {'outputs': os},
99                {'outputs': os},
100                {'outputs': os}]
101
102        qem3 = VppQosEgressMap(self, 3, rows).add_vpp_config()
103
104        output = [scapy.compat.chb(4)] * 256
105        os = b''.join(output)
106        rows = [{'outputs': os},
107                {'outputs': os},
108                {'outputs': os},
109                {'outputs': os}]
110
111        qem4 = VppQosEgressMap(self, 4, rows).add_vpp_config()
112        qem5 = VppQosEgressMap(self, 5, rows).add_vpp_config()
113        qem6 = VppQosEgressMap(self, 6, rows).add_vpp_config()
114        qem7 = VppQosEgressMap(self, 7, rows).add_vpp_config()
115
116        self.assertTrue(qem7.query_vpp_config())
117        self.logger.info(self.vapi.cli("sh qos eg map"))
118
119        #
120        # Bind interface pgN to table n
121        #
122        qm1 = VppQosMark(self, self.pg1, qem1,
123                         self.QOS_SOURCE.QOS_API_SOURCE_IP).add_vpp_config()
124        qm2 = VppQosMark(self, self.pg2, qem2,
125                         self.QOS_SOURCE.QOS_API_SOURCE_IP).add_vpp_config()
126        qm3 = VppQosMark(self, self.pg3, qem3,
127                         self.QOS_SOURCE.QOS_API_SOURCE_IP).add_vpp_config()
128        qm4 = VppQosMark(self, self.pg4, qem4,
129                         self.QOS_SOURCE.QOS_API_SOURCE_IP).add_vpp_config()
130        self.assertTrue(qm3.query_vpp_config())
131
132        self.logger.info(self.vapi.cli("sh qos mark"))
133
134        #
135        # packets ingress on Pg0
136        #
137        p_v4 = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) /
138                IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4, tos=1) /
139                UDP(sport=1234, dport=1234) /
140                Raw(scapy.compat.chb(100) * NUM_PKTS))
141        p_v6 = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) /
142                IPv6(src=self.pg0.remote_ip6, dst=self.pg1.remote_ip6,
143                     tc=1) /
144                UDP(sport=1234, dport=1234) /
145                Raw(scapy.compat.chb(100) * NUM_PKTS))
146
147        #
148        # Since we have not yet enabled the recording of the input QoS
149        # from the input iP header, the egress packet's ToS will be unchanged
150        #
151        rx = self.send_and_expect(self.pg0, p_v4 * NUM_PKTS, self.pg1)
152        for p in rx:
153            self.assertEqual(p[IP].tos, 1)
154        rx = self.send_and_expect(self.pg0, p_v6 * NUM_PKTS, self.pg1)
155        for p in rx:
156            self.assertEqual(p[IPv6].tc, 1)
157
158        #
159        # Enable QoS recording on IP input for pg0
160        #
161        qr1 = VppQosRecord(self, self.pg0,
162                           self.QOS_SOURCE.QOS_API_SOURCE_IP)
163        qr1.add_vpp_config()
164        self.logger.info(self.vapi.cli("sh qos record"))
165
166        #
167        # send the same packets, this time expect the input TOS of 1
168        # to be mapped to pg1's egress value of 254
169        #
170        rx = self.send_and_expect(self.pg0, p_v4 * NUM_PKTS, self.pg1)
171        for p in rx:
172            self.assertEqual(p[IP].tos, 254)
173        rx = self.send_and_expect(self.pg0, p_v6 * NUM_PKTS, self.pg1)
174        for p in rx:
175            self.assertEqual(p[IPv6].tc, 254)
176
177        #
178        # different input ToS to test the mapping
179        #
180        p_v4[IP].tos = 127
181        rx = self.send_and_expect(self.pg0, p_v4 * NUM_PKTS, self.pg1)
182        for p in rx:
183            self.assertEqual(p[IP].tos, 128)
184        p_v6[IPv6].tc = 127
185        rx = self.send_and_expect(self.pg0, p_v6 * NUM_PKTS, self.pg1)
186        for p in rx:
187            self.assertEqual(p[IPv6].tc, 128)
188
189        p_v4[IP].tos = 254
190        rx = self.send_and_expect(self.pg0, p_v4 * NUM_PKTS, self.pg1)
191        for p in rx:
192            self.assertEqual(p[IP].tos, 1)
193        p_v6[IPv6].tc = 254
194        rx = self.send_and_expect(self.pg0, p_v6 * NUM_PKTS, self.pg1)
195        for p in rx:
196            self.assertEqual(p[IPv6].tc, 1)
197
198        #
199        # send packets out the other interfaces to test the maps are
200        # correctly applied
201        #
202        p_v4[IP].dst = self.pg2.remote_ip4
203        rx = self.send_and_expect(self.pg0, p_v4 * NUM_PKTS, self.pg2)
204        for p in rx:
205            self.assertEqual(p[IP].tos, 2)
206
207        p_v4[IP].dst = self.pg3.remote_ip4
208        rx = self.send_and_expect(self.pg0, p_v4 * NUM_PKTS, self.pg3)
209        for p in rx:
210            self.assertEqual(p[IP].tos, 3)
211
212        p_v6[IPv6].dst = self.pg3.remote_ip6
213        rx = self.send_and_expect(self.pg0, p_v6 * NUM_PKTS, self.pg3)
214        for p in rx:
215            self.assertEqual(p[IPv6].tc, 3)
216
217        #
218        # remove the map on pg2 and pg3, now expect an unchanged IP tos
219        #
220        qm2.remove_vpp_config()
221        qm3.remove_vpp_config()
222        self.logger.info(self.vapi.cli("sh qos mark"))
223
224        self.assertFalse(qm3.query_vpp_config())
225        self.logger.info(self.vapi.cli("sh int feat pg2"))
226
227        p_v4[IP].dst = self.pg2.remote_ip4
228        rx = self.send_and_expect(self.pg0, p_v4 * NUM_PKTS, self.pg2)
229        for p in rx:
230            self.assertEqual(p[IP].tos, 254)
231
232        p_v4[IP].dst = self.pg3.remote_ip4
233        rx = self.send_and_expect(self.pg0, p_v4 * NUM_PKTS, self.pg3)
234        for p in rx:
235            self.assertEqual(p[IP].tos, 254)
236
237        #
238        # still mapping out of pg1
239        #
240        p_v4[IP].dst = self.pg1.remote_ip4
241        rx = self.send_and_expect(self.pg0, p_v4 * NUM_PKTS, self.pg1)
242        for p in rx:
243            self.assertEqual(p[IP].tos, 1)
244
245        #
246        # disable the input recording on pg0
247        #
248        self.assertTrue(qr1.query_vpp_config())
249        qr1.remove_vpp_config()
250
251        #
252        # back to an unchanged TOS value
253        #
254        rx = self.send_and_expect(self.pg0, p_v4 * NUM_PKTS, self.pg1)
255        for p in rx:
256            self.assertEqual(p[IP].tos, 254)
257
258        #
259        # enable QoS stroe instead of record
260        #
261        qst1 = VppQosStore(self, self.pg0,
262                           self.QOS_SOURCE.QOS_API_SOURCE_IP,
263                           5).add_vpp_config()
264        self.logger.info(self.vapi.cli("sh qos store"))
265
266        p_v4[IP].dst = self.pg1.remote_ip4
267        rx = self.send_and_expect(self.pg0, p_v4 * NUM_PKTS, self.pg1)
268        for p in rx:
269            self.assertEqual(p[IP].tos, 250)
270
271        #
272        # disable the input storing on pg0
273        #
274        self.assertTrue(qst1.query_vpp_config())
275        qst1.remove_vpp_config()
276
277        #
278        # back to an unchanged TOS value
279        #
280        rx = self.send_and_expect(self.pg0, p_v4 * NUM_PKTS, self.pg1)
281        for p in rx:
282            self.assertEqual(p[IP].tos, 254)
283
284        #
285        # disable the egress map on pg1 and pg4
286        #
287        qm1.remove_vpp_config()
288        qm4.remove_vpp_config()
289
290        #
291        # unchanged Tos on pg1
292        #
293        rx = self.send_and_expect(self.pg0, p_v4 * NUM_PKTS, self.pg1)
294        for p in rx:
295            self.assertEqual(p[IP].tos, 254)
296
297    def test_qos_mpls(self):
298        """ QoS Mark/Record MPLS """
299
300        #
301        # 255 QoS for all input values
302        #
303        from_ext = 7
304        from_ip = 6
305        from_mpls = 5
306        from_vlan = 4
307        output = [scapy.compat.chb(from_ext)] * 256
308        os1 = b''.join(output)
309        output = [scapy.compat.chb(from_vlan)] * 256
310        os2 = b''.join(output)
311        output = [scapy.compat.chb(from_mpls)] * 256
312        os3 = b''.join(output)
313        output = [scapy.compat.chb(from_ip)] * 256
314        os4 = b''.join(output)
315        rows = [{'outputs': os1},
316                {'outputs': os2},
317                {'outputs': os3},
318                {'outputs': os4}]
319
320        qem1 = VppQosEgressMap(self, 1, rows).add_vpp_config()
321
322        #
323        # a route with 1 MPLS label
324        #
325        route_10_0_0_1 = VppIpRoute(self, "10.0.0.1", 32,
326                                    [VppRoutePath(self.pg1.remote_ip4,
327                                                  self.pg1.sw_if_index,
328                                                  labels=[32])])
329        route_10_0_0_1.add_vpp_config()
330
331        #
332        # a route with 3 MPLS labels
333        #
334        route_10_0_0_3 = VppIpRoute(self, "10.0.0.3", 32,
335                                    [VppRoutePath(self.pg1.remote_ip4,
336                                                  self.pg1.sw_if_index,
337                                                  labels=[63, 33, 34])])
338        route_10_0_0_3.add_vpp_config()
339
340        #
341        # enable IP QoS recording on the input Pg0 and MPLS egress marking
342        # on Pg1
343        #
344        qr1 = VppQosRecord(self, self.pg0,
345                           self.QOS_SOURCE.QOS_API_SOURCE_IP).add_vpp_config()
346        qm1 = VppQosMark(self, self.pg1, qem1,
347                         self.QOS_SOURCE.QOS_API_SOURCE_MPLS).add_vpp_config()
348
349        #
350        # packet that will get one label added and 3 labels added resp.
351        #
352        p_1 = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) /
353               IP(src=self.pg0.remote_ip4, dst="10.0.0.1", tos=1) /
354               UDP(sport=1234, dport=1234) /
355               Raw(scapy.compat.chb(100) * NUM_PKTS))
356        p_3 = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) /
357               IP(src=self.pg0.remote_ip4, dst="10.0.0.3", tos=1) /
358               UDP(sport=1234, dport=1234) /
359               Raw(scapy.compat.chb(100) * NUM_PKTS))
360
361        rx = self.send_and_expect(self.pg0, p_1 * NUM_PKTS, self.pg1)
362
363        #
364        # only 3 bits of ToS value in MPLS make sure tos is correct
365        # and the label and EOS bit have not been corrupted
366        #
367        for p in rx:
368            self.assertEqual(p[MPLS].cos, from_ip)
369            self.assertEqual(p[MPLS].label, 32)
370            self.assertEqual(p[MPLS].s, 1)
371        rx = self.send_and_expect(self.pg0, p_3 * NUM_PKTS, self.pg1)
372        for p in rx:
373            self.assertEqual(p[MPLS].cos, from_ip)
374            self.assertEqual(p[MPLS].label, 63)
375            self.assertEqual(p[MPLS].s, 0)
376            h = p[MPLS].payload
377            self.assertEqual(h[MPLS].cos, from_ip)
378            self.assertEqual(h[MPLS].label, 33)
379            self.assertEqual(h[MPLS].s, 0)
380            h = h[MPLS].payload
381            self.assertEqual(h[MPLS].cos, from_ip)
382            self.assertEqual(h[MPLS].label, 34)
383            self.assertEqual(h[MPLS].s, 1)
384
385        #
386        # enable MPLS QoS recording on the input Pg0 and IP egress marking
387        # on Pg1
388        #
389        qr2 = VppQosRecord(
390            self, self.pg0,
391            self.QOS_SOURCE.QOS_API_SOURCE_MPLS).add_vpp_config()
392        qm2 = VppQosMark(
393            self, self.pg1, qem1,
394            self.QOS_SOURCE.QOS_API_SOURCE_IP).add_vpp_config()
395
396        #
397        # MPLS x-connect - COS according to pg1 map
398        #
399        route_32_eos = VppMplsRoute(self, 32, 1,
400                                    [VppRoutePath(self.pg1.remote_ip4,
401                                                  self.pg1.sw_if_index,
402                                                  labels=[VppMplsLabel(33)])])
403        route_32_eos.add_vpp_config()
404
405        p_m1 = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) /
406                MPLS(label=32, cos=3, ttl=2) /
407                IP(src=self.pg0.remote_ip4, dst="10.0.0.1", tos=1) /
408                UDP(sport=1234, dport=1234) /
409                Raw(scapy.compat.chb(100) * NUM_PKTS))
410
411        rx = self.send_and_expect(self.pg0, p_m1 * NUM_PKTS, self.pg1)
412        for p in rx:
413            self.assertEqual(p[MPLS].cos, from_mpls)
414            self.assertEqual(p[MPLS].label, 33)
415            self.assertEqual(p[MPLS].s, 1)
416
417        #
418        # MPLS deag - COS is copied from MPLS to IP
419        #
420        route_33_eos = VppMplsRoute(self, 33, 1,
421                                    [VppRoutePath("0.0.0.0",
422                                                  0xffffffff,
423                                                  nh_table_id=0)])
424        route_33_eos.add_vpp_config()
425
426        route_10_0_0_4 = VppIpRoute(self, "10.0.0.4", 32,
427                                    [VppRoutePath(self.pg1.remote_ip4,
428                                                  self.pg1.sw_if_index)])
429        route_10_0_0_4.add_vpp_config()
430
431        p_m2 = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) /
432                MPLS(label=33, ttl=2, cos=3) /
433                IP(src=self.pg0.remote_ip4, dst="10.0.0.4", tos=1) /
434                UDP(sport=1234, dport=1234) /
435                Raw(scapy.compat.chb(100) * NUM_PKTS))
436
437        rx = self.send_and_expect(self.pg0, p_m2 * NUM_PKTS, self.pg1)
438
439        for p in rx:
440            self.assertEqual(p[IP].tos, from_mpls)
441
442    def test_qos_vlan(self):
443        """QoS mark/record VLAN """
444
445        #
446        # QoS for all input values
447        #
448        output = [scapy.compat.chb(0)] * 256
449        for i in range(0, 255):
450            output[i] = scapy.compat.chb(255 - i)
451        os = b''.join(output)
452        rows = [{'outputs': os},
453                {'outputs': os},
454                {'outputs': os},
455                {'outputs': os}]
456
457        qem1 = VppQosEgressMap(self, 1, rows).add_vpp_config()
458
459        sub_if = VppDot1QSubint(self, self.pg0, 11)
460
461        sub_if.admin_up()
462        sub_if.config_ip4()
463        sub_if.resolve_arp()
464        sub_if.config_ip6()
465        sub_if.resolve_ndp()
466
467        #
468        # enable VLAN QoS recording/marking on the input Pg0 subinterface and
469        #
470        qr_v = VppQosRecord(
471            self, sub_if,
472            self.QOS_SOURCE.QOS_API_SOURCE_VLAN).add_vpp_config()
473        qm_v = VppQosMark(
474            self, sub_if, qem1,
475            self.QOS_SOURCE.QOS_API_SOURCE_VLAN).add_vpp_config()
476
477        #
478        # IP marking/recording on pg1
479        #
480        qr_ip = VppQosRecord(
481            self, self.pg1,
482            self.QOS_SOURCE.QOS_API_SOURCE_IP).add_vpp_config()
483        qm_ip = VppQosMark(
484            self, self.pg1, qem1,
485            self.QOS_SOURCE.QOS_API_SOURCE_IP).add_vpp_config()
486
487        #
488        # a routes to/from sub-interface
489        #
490        route_10_0_0_1 = VppIpRoute(self, "10.0.0.1", 32,
491                                    [VppRoutePath(sub_if.remote_ip4,
492                                                  sub_if.sw_if_index)])
493        route_10_0_0_1.add_vpp_config()
494        route_10_0_0_2 = VppIpRoute(self, "10.0.0.2", 32,
495                                    [VppRoutePath(self.pg1.remote_ip4,
496                                                  self.pg1.sw_if_index)])
497        route_10_0_0_2.add_vpp_config()
498        route_2001_1 = VppIpRoute(self, "2001::1", 128,
499                                  [VppRoutePath(sub_if.remote_ip6,
500                                                sub_if.sw_if_index)])
501        route_2001_1.add_vpp_config()
502        route_2001_2 = VppIpRoute(self, "2001::2", 128,
503                                  [VppRoutePath(self.pg1.remote_ip6,
504                                                self.pg1.sw_if_index)])
505        route_2001_2.add_vpp_config()
506
507        p_v1 = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) /
508                Dot1Q(vlan=11, prio=1) /
509                IP(src="1.1.1.1", dst="10.0.0.2", tos=1) /
510                UDP(sport=1234, dport=1234) /
511                Raw(scapy.compat.chb(100) * NUM_PKTS))
512
513        p_v2 = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) /
514                IP(src="1.1.1.1", dst="10.0.0.1", tos=1) /
515                UDP(sport=1234, dport=1234) /
516                Raw(scapy.compat.chb(100) * NUM_PKTS))
517
518        p_v3 = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) /
519                Dot1Q(vlan=11, prio=1, id=1) /
520                IP(src="1.1.1.1", dst="10.0.0.2", tos=2) /
521                UDP(sport=1234, dport=1234) /
522                Raw(scapy.compat.chb(100) * NUM_PKTS))
523
524        rx = self.send_and_expect(self.pg1, p_v2 * NUM_PKTS, self.pg0)
525
526        for p in rx:
527            self.assertEqual(p[Dot1Q].prio, 7)
528            self.assertEqual(p[Dot1Q].id, 0)
529
530        rx = self.send_and_expect(self.pg0, p_v3 * NUM_PKTS, self.pg1)
531
532        for p in rx:
533            self.assertEqual(p[IP].tos, 252)
534
535        rx = self.send_and_expect(self.pg0, p_v1 * NUM_PKTS, self.pg1)
536
537        for p in rx:
538            self.assertEqual(p[IP].tos, 253)
539
540        p_v1 = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) /
541                Dot1Q(vlan=11, prio=2) /
542                IPv6(src="2001::1", dst="2001::2", tc=1) /
543                UDP(sport=1234, dport=1234) /
544                Raw(scapy.compat.chb(100) * NUM_PKTS))
545
546        p_v2 = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) /
547                IPv6(src="3001::1", dst="2001::1", tc=1) /
548                UDP(sport=1234, dport=1234) /
549                Raw(scapy.compat.chb(100) * NUM_PKTS))
550
551        rx = self.send_and_expect(self.pg1, p_v2 * NUM_PKTS, self.pg0)
552
553        for p in rx:
554            self.assertEqual(p[Dot1Q].prio, 7)
555            self.assertEqual(p[Dot1Q].id, 0)
556
557        rx = self.send_and_expect(self.pg0, p_v1 * NUM_PKTS, self.pg1)
558
559        for p in rx:
560            self.assertEqual(p[IPv6].tc, 251)
561
562        #
563        # cleanup
564        #
565        sub_if.unconfig_ip4()
566        sub_if.unconfig_ip6()
567
568
569if __name__ == '__main__':
570    unittest.main(testRunner=VppTestRunner)
571