summaryrefslogtreecommitdiffstats
path: root/src/vat2/CMakeLists.txt
blob: 9069d8f6b6203852e8b4b152b3980dce1a0900af (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
# Copyright (c) 2020 Cisco and/or its affiliates.
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at:
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

##############################################################################
# vat2
##############################################################################
add_vpp_executable(vat2 ENABLE_EXPORTS NO_INSTALL
  SOURCES
  main.c
  plugin.c
  jsonconvert.c

  DEPENDS api_headers

  LINK_LIBRARIES
  vlibmemoryclient
  svm
  vppinfra
  vppapiclient
  Threads::Threads
  rt m dl crypto
)

#
# Unit test code. Call generator directly to avoid it being installed
#set_source_files_properties(vat2_test.c PROPERTIES
#  COMPILE_FLAGS " -fsanitize=address"
#)

vpp_generate_api_c_header (test/vat2_test.api)
add_vpp_executable(test_vat2 ENABLE_EXPORTS NO_INSTALL
  SOURCES
  test/vat2_test.c
  jsonconvert.c

  DEPENDS api_headers

  LINK_LIBRARIES
  vlibmemoryclient
  svm
  vppinfra
  vppapiclient
  Threads::Threads
  rt m dl crypto
)
#target_link_options(test_vat2 PUBLIC "LINKER:-fsanitize=address")

if("${CMAKE_VERSION}" VERSION_GREATER_EQUAL "3.13" AND "${CMAKE_C_COMPILER_ID}" MATCHES "(Apple)?[Cc]lang")
  set(TARGET_NAME test_vat2)
  set(COV_SOURCES ${CMAKE_SOURCE_DIR}/vat2/jsonconvert.c)

  message("Building with llvm Code Coverage Tools ${TARGET_NAME}")
  target_compile_options(${TARGET_NAME} PRIVATE -fprofile-instr-generate -fcoverage-mapping)
  target_link_options(${TARGET_NAME} PRIVATE -fprofile-instr-generate -fcoverage-mapping)
  target_compile_options(${TARGET_NAME} PRIVATE -fsanitize=address)
  target_link_options(${TARGET_NAME} PRIVATE -fsanitize=address)

  # llvm-cov
  add_custom_target(${TARGET_NAME}-ccov-preprocessing
    COMMAND LLVM_PROFILE_FILE=${TARGET_NAME}.profraw $<TARGET_FILE:${TARGET_NAME}>
    COMMAND llvm-profdata merge -sparse ${TARGET_NAME}.profraw -o ${TARGET_NAME}.profdata
    DEPENDS ${TARGET_NAME})

  add_custom_target(${TARGET_NAME}-ccov-show
    COMMAND llvm-cov show $<TARGET_FILE:${TARGET_NAME}> -instr-profile=${TARGET_NAME}.profdata -show-line-counts-or-regions ${COV_SOURCES}
    DEPENDS ${TARGET_NAME}-ccov-preprocessing)

  add_custom_target(${TARGET_NAME}-ccov-report
    COMMAND llvm-cov report -show-functions $<TARGET_FILE:${TARGET_NAME}> -instr-profile=${TARGET_NAME}.profdata ${COV_SOURCES}
    DEPENDS ${TARGET_NAME}-ccov-preprocessing)

  add_custom_target(${TARGET_NAME}-ccov
    COMMAND llvm-cov show $<TARGET_FILE:${TARGET_NAME}> -instr-profile=${TARGET_NAME}.profdata -show-line-counts-or-regions -output-dir=${CMAKE_RUNTIME_OUTPUT_DIRECTORY}/${TARGET_NAME}-llvm-cov -format="html" ${COV_SOURCES}
    DEPENDS ${TARGET_NAME}-ccov-preprocessing)

  add_custom_command(TARGET ${TARGET_NAME}-ccov POST_BUILD
    COMMAND ;
    COMMENT "Open ${CMAKE_RUNTIME_OUTPUT_DIRECTORY}/${TARGET_NAME}-llvm-cov/index.html in your browser to view the coverage report."
)
endif()



##############################################################################
# vat2 headers
##############################################################################
install(
  FILES
  jsonconvert.h
  vat2_helpers.h
  DESTINATION include/vat2
  COMPONENT vpp-dev
)
a id='n508' href='#n508'>508 509 510 511 512 513 514 515 516 517 518 519 520 521 522 523 524 525 526 527 528 529 530 531 532 533 534 535 536 537 538 539 540 541 542 543 544 545 546 547 548 549 550 551 552 553 554 555 556 557 558 559 560 561 562 563 564 565 566 567 568 569 570 571 572 573 574 575 576 577 578 579 580 581 582 583 584 585 586 587 588
#!/usr/bin/env python

import unittest

from framework import VppTestCase, VppTestRunner
from vpp_papi_provider import QOS_SOURCE
from vpp_sub_interface import VppDot1QSubint
from vpp_ip import DpoProto
from vpp_ip_route import VppIpRoute, VppRoutePath, VppMplsRoute, \
    VppMplsLabel, VppMplsTable

import scapy.compat
from scapy.packet import Raw
from scapy.layers.l2 import Ether, Dot1Q
from scapy.layers.inet import IP, UDP
from scapy.layers.inet6 import IPv6
from scapy.contrib.mpls import MPLS


class TestQOS(VppTestCase):
    """ QOS Test Case """

    @classmethod
    def setUpClass(cls):
        super(TestQOS, cls).setUpClass()

    @classmethod
    def tearDownClass(cls):
        super(TestQOS, cls).tearDownClass()

    def setUp(self):
        super(TestQOS, self).setUp()

        self.create_pg_interfaces(range(5))

        tbl = VppMplsTable(self, 0)
        tbl.add_vpp_config()

        for i in self.pg_interfaces:
            i.admin_up()
            i.config_ip4()
            i.resolve_arp()
            i.config_ip6()
            i.resolve_ndp()
            i.enable_mpls()

    def tearDown(self):
        for i in self.pg_interfaces:
            i.unconfig_ip4()
            i.unconfig_ip6()
            i.disable_mpls()

        super(TestQOS, self).tearDown()

    def test_qos_ip(self):
        """ QoS Mark/Record IP """

        #
        # for table 1 map the n=0xff possible values of input QoS mark,
        # n to 1-n
        #
        output = [scapy.compat.chb(0)] * 256
        for i in range(0, 255):
            output[i] = scapy.compat.chb(255 - i)
        os = b''.join(output)
        rows = [{'outputs': os},
                {'outputs': os},
                {'outputs': os},
                {'outputs': os}]

        self.vapi.qos_egress_map_update(1, rows)

        #
        # For table 2 (and up) use the value n for everything
        #
        output = [scapy.compat.chb(2)] * 256
        os = b''.join(output)
        rows = [{'outputs': os},
                {'outputs': os},
                {'outputs': os},
                {'outputs': os}]

        self.vapi.qos_egress_map_update(2, rows)

        output = [scapy.compat.chb(3)] * 256
        os = b''.join(output)
        rows = [{'outputs': os},
                {'outputs': os},
                {'outputs': os},
                {'outputs': os}]

        self.vapi.qos_egress_map_update(3, rows)

        output = [scapy.compat.chb(4)] * 256
        os = b''.join(output)
        rows = [{'outputs': os},
                {'outputs': os},
                {'outputs': os},
                {'outputs': os}]
        self.vapi.qos_egress_map_update(4, rows)
        self.vapi.qos_egress_map_update(5, rows)
        self.vapi.qos_egress_map_update(6, rows)
        self.vapi.qos_egress_map_update(7, rows)

        self.logger.info(self.vapi.cli("sh qos eg map"))

        #
        # Bind interface pgN to table n
        #
        self.vapi.qos_mark_enable_disable(self.pg1.sw_if_index,
                                          QOS_SOURCE.IP,
                                          1,
                                          1)
        self.vapi.qos_mark_enable_disable(self.pg2.sw_if_index,
                                          QOS_SOURCE.IP,
                                          2,
                                          1)
        self.vapi.qos_mark_enable_disable(self.pg3.sw_if_index,
                                          QOS_SOURCE.IP,
                                          3,
                                          1)
        self.vapi.qos_mark_enable_disable(self.pg4.sw_if_index,
                                          QOS_SOURCE.IP,
                                          4,
                                          1)

        #
        # packets ingress on Pg0
        #
        p_v4 = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) /
                IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4, tos=1) /
                UDP(sport=1234, dport=1234) /
                Raw(scapy.compat.chb(100) * 65))
        p_v6 = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) /
                IPv6(src=self.pg0.remote_ip6, dst=self.pg1.remote_ip6,
                     tc=1) /
                UDP(sport=1234, dport=1234) /
                Raw(scapy.compat.chb(100) * 65))

        #
        # Since we have not yet enabled the recording of the input QoS
        # from the input iP header, the egress packet's ToS will be unchanged
        #
        rx = self.send_and_expect(self.pg0, p_v4 * 65, self.pg1)
        for p in rx:
            self.assertEqual(p[IP].tos, 1)
        rx = self.send_and_expect(self.pg0, p_v6 * 65, self.pg1)
        for p in rx:
            self.assertEqual(p[IPv6].tc, 1)

        #
        # Enable QoS recording on IP input for pg0
        #
        self.vapi.qos_record_enable_disable(self.pg0.sw_if_index,
                                            QOS_SOURCE.IP,
                                            1)

        #
        # send the same packets, this time expect the input TOS of 1
        # to be mapped to pg1's egress value of 254
        #
        rx = self.send_and_expect(self.pg0, p_v4 * 65, self.pg1)
        for p in rx:
            self.assertEqual(p[IP].tos, 254)
        rx = self.send_and_expect(self.pg0, p_v6 * 65, self.pg1)
        for p in rx:
            self.assertEqual(p[IPv6].tc, 254)

        #
        # different input ToS to test the mapping
        #
        p_v4[IP].tos = 127
        rx = self.send_and_expect(self.pg0, p_v4 * 65, self.pg1)
        for p in rx:
            self.assertEqual(p[IP].tos, 128)
        p_v6[IPv6].tc = 127
        rx = self.send_and_expect(self.pg0, p_v6 * 65, self.pg1)
        for p in rx:
            self.assertEqual(p[IPv6].tc, 128)

        p_v4[IP].tos = 254
        rx = self.send_and_expect(self.pg0, p_v4 * 65, self.pg1)
        for p in rx:
            self.assertEqual(p[IP].tos, 1)
        p_v6[IPv6].tc = 254
        rx = self.send_and_expect(self.pg0, p_v6 * 65, self.pg1)
        for p in rx:
            self.assertEqual(p[IPv6].tc, 1)

        #
        # send packets out the other interfaces to test the maps are
        # correctly applied
        #
        p_v4[IP].dst = self.pg2.remote_ip4
        rx = self.send_and_expect(self.pg0, p_v4 * 65, self.pg2)
        for p in rx:
            self.assertEqual(p[IP].tos, 2)

        p_v4[IP].dst = self.pg3.remote_ip4
        rx = self.send_and_expect(self.pg0, p_v4 * 65, self.pg3)
        for p in rx:
            self.assertEqual(p[IP].tos, 3)

        p_v6[IPv6].dst = self.pg3.remote_ip6
        rx = self.send_and_expect(self.pg0, p_v6 * 65, self.pg3)
        for p in rx:
            self.assertEqual(p[IPv6].tc, 3)

        #
        # remove the map on pg2 and pg3, now expect an unchanged IP tos
        #
        self.vapi.qos_mark_enable_disable(self.pg2.sw_if_index,
                                          QOS_SOURCE.IP,
                                          2,
                                          0)
        self.vapi.qos_mark_enable_disable(self.pg3.sw_if_index,
                                          QOS_SOURCE.IP,
                                          3,
                                          0)
        self.logger.info(self.vapi.cli("sh int feat pg2"))

        p_v4[IP].dst = self.pg2.remote_ip4
        rx = self.send_and_expect(self.pg0, p_v4 * 65, self.pg2)
        for p in rx:
            self.assertEqual(p[IP].tos, 254)

        p_v4[IP].dst = self.pg3.remote_ip4
        rx = self.send_and_expect(self.pg0, p_v4 * 65, self.pg3)
        for p in rx:
            self.assertEqual(p[IP].tos, 254)

        #
        # still mapping out of pg1
        #
        p_v4[IP].dst = self.pg1.remote_ip4
        rx = self.send_and_expect(self.pg0, p_v4 * 65, self.pg1)
        for p in rx:
            self.assertEqual(p[IP].tos, 1)

        #
        # disable the input recording on pg0
        #
        self.vapi.qos_record_enable_disable(self.pg0.sw_if_index,
                                            QOS_SOURCE.IP,
                                            0)

        #
        # back to an unchanged TOS value
        #
        rx = self.send_and_expect(self.pg0, p_v4 * 65, self.pg1)
        for p in rx:
            self.assertEqual(p[IP].tos, 254)

        #
        # disable the egress map on pg1 and pg4
        #
        self.vapi.qos_mark_enable_disable(self.pg1.sw_if_index,
                                          QOS_SOURCE.IP,
                                          1,
                                          0)
        self.vapi.qos_mark_enable_disable(self.pg4.sw_if_index,
                                          QOS_SOURCE.IP,
                                          4,
                                          0)

        #
        # unchanged Tos on pg1
        #
        rx = self.send_and_expect(self.pg0, p_v4 * 65, self.pg1)
        for p in rx:
            self.assertEqual(p[IP].tos, 254)

        #
        # clean-up the map
        #
        self.vapi.qos_egress_map_delete(1)
        self.vapi.qos_egress_map_delete(4)
        self.vapi.qos_egress_map_delete(2)
        self.vapi.qos_egress_map_delete(3)
        self.vapi.qos_egress_map_delete(5)
        self.vapi.qos_egress_map_delete(6)
        self.vapi.qos_egress_map_delete(7)

    def test_qos_mpls(self):
        """ QoS Mark/Record MPLS """

        #
        # 255 QoS for all input values
        #
        from_ext = 7
        from_ip = 6
        from_mpls = 5
        from_vlan = 4
        output = [scapy.compat.chb(from_ext)] * 256
        os1 = b''.join(output)
        output = [scapy.compat.chb(from_vlan)] * 256
        os2 = b''.join(output)
        output = [scapy.compat.chb(from_mpls)] * 256
        os3 = b''.join(output)
        output = [scapy.compat.chb(from_ip)] * 256
        os4 = b''.join(output)
        rows = [{'outputs': os1},
                {'outputs': os2},
                {'outputs': os3},
                {'outputs': os4}]

        self.vapi.qos_egress_map_update(1, rows)

        #
        # a route with 1 MPLS label
        #
        route_10_0_0_1 = VppIpRoute(self, "10.0.0.1", 32,
                                    [VppRoutePath(self.pg1.remote_ip4,
                                                  self.pg1.sw_if_index,
                                                  labels=[32])])
        route_10_0_0_1.add_vpp_config()

        #
        # a route with 3 MPLS labels
        #
        route_10_0_0_3 = VppIpRoute(self, "10.0.0.3", 32,
                                    [VppRoutePath(self.pg1.remote_ip4,
                                                  self.pg1.sw_if_index,
                                                  labels=[63, 33, 34])])
        route_10_0_0_3.add_vpp_config()

        #
        # enable IP QoS recording on the input Pg0 and MPLS egress marking
        # on Pg1
        #
        self.vapi.qos_record_enable_disable(self.pg0.sw_if_index,
                                            QOS_SOURCE.IP,
                                            1)
        self.vapi.qos_mark_enable_disable(self.pg1.sw_if_index,
                                          QOS_SOURCE.MPLS,
                                          1,
                                          1)

        #
        # packet that will get one label added and 3 labels added resp.
        #
        p_1 = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) /
               IP(src=self.pg0.remote_ip4, dst="10.0.0.1", tos=1) /
               UDP(sport=1234, dport=1234) /
               Raw(scapy.compat.chb(100) * 65))
        p_3 = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) /
               IP(src=self.pg0.remote_ip4, dst="10.0.0.3", tos=1) /
               UDP(sport=1234, dport=1234) /
               Raw(scapy.compat.chb(100) * 65))

        rx = self.send_and_expect(self.pg0, p_1 * 65, self.pg1)

        #
        # only 3 bits of ToS value in MPLS make sure tos is correct
        # and the label and EOS bit have not been corrupted
        #
        for p in rx:
            self.assertEqual(p[MPLS].cos, from_ip)
            self.assertEqual(p[MPLS].label, 32)
            self.assertEqual(p[MPLS].s, 1)
        rx = self.send_and_expect(self.pg0, p_3 * 65, self.pg1)
        for p in rx:
            self.assertEqual(p[MPLS].cos, from_ip)
            self.assertEqual(p[MPLS].label, 63)
            self.assertEqual(p[MPLS].s, 0)
            h = p[MPLS].payload
            self.assertEqual(h[MPLS].cos, from_ip)
            self.assertEqual(h[MPLS].label, 33)
            self.assertEqual(h[MPLS].s, 0)
            h = h[MPLS].payload
            self.assertEqual(h[MPLS].cos, from_ip)
            self.assertEqual(h[MPLS].label, 34)
            self.assertEqual(h[MPLS].s, 1)

        #
        # enable MPLS QoS recording on the input Pg0 and IP egress marking
        # on Pg1
        #
        self.vapi.qos_record_enable_disable(self.pg0.sw_if_index,
                                            QOS_SOURCE.MPLS,
                                            1)
        self.vapi.qos_mark_enable_disable(self.pg1.sw_if_index,
                                          QOS_SOURCE.IP,
                                          1,
                                          1)

        #
        # MPLS x-connect - COS according to pg1 map
        #
        route_32_eos = VppMplsRoute(self, 32, 1,
                                    [VppRoutePath(self.pg1.remote_ip4,
                                                  self.pg1.sw_if_index,
                                                  labels=[VppMplsLabel(33)])])
        route_32_eos.add_vpp_config()

        p_m1 = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) /
                MPLS(label=32, cos=3, ttl=2) /
                IP(src=self.pg0.remote_ip4, dst="10.0.0.1", tos=1) /
                UDP(sport=1234, dport=1234) /
                Raw(scapy.compat.chb(100) * 65))

        rx = self.send_and_expect(self.pg0, p_m1 * 65, self.pg1)
        for p in rx:
            self.assertEqual(p[MPLS].cos, from_mpls)
            self.assertEqual(p[MPLS].label, 33)
            self.assertEqual(p[MPLS].s, 1)

        #
        # MPLS deag - COS is copied from MPLS to IP
        #
        route_33_eos = VppMplsRoute(self, 33, 1,
                                    [VppRoutePath("0.0.0.0",
                                                  0xffffffff,
                                                  nh_table_id=0)])
        route_33_eos.add_vpp_config()

        route_10_0_0_4 = VppIpRoute(self, "10.0.0.4", 32,
                                    [VppRoutePath(self.pg1.remote_ip4,
                                                  self.pg1.sw_if_index)])
        route_10_0_0_4.add_vpp_config()

        p_m2 = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) /
                MPLS(label=33, ttl=2, cos=3) /
                IP(src=self.pg0.remote_ip4, dst="10.0.0.4", tos=1) /
                UDP(sport=1234, dport=1234) /
                Raw(scapy.compat.chb(100) * 65))

        rx = self.send_and_expect(self.pg0, p_m2 * 65, self.pg1)

        for p in rx:
            self.assertEqual(p[IP].tos, from_mpls)

        #
        # cleanup
        #
        self.vapi.qos_record_enable_disable(self.pg0.sw_if_index,
                                            QOS_SOURCE.IP,
                                            0)
        self.vapi.qos_mark_enable_disable(self.pg1.sw_if_index,
                                          QOS_SOURCE.MPLS,
                                          1,
                                          0)
        self.vapi.qos_record_enable_disable(self.pg0.sw_if_index,
                                            QOS_SOURCE.MPLS,
                                            0)
        self.vapi.qos_mark_enable_disable(self.pg1.sw_if_index,
                                          QOS_SOURCE.IP,
                                          1,
                                          0)
        self.vapi.qos_egress_map_delete(1)

    def test_qos_vlan(self):
        """QoS mark/record VLAN """

        #
        # QoS for all input values
        #
        output = [scapy.compat.chb(0)] * 256
        for i in range(0, 255):
            output[i] = scapy.compat.chb(255 - i)
        os = b''.join(output)
        rows = [{'outputs': os},
                {'outputs': os},
                {'outputs': os},
                {'outputs': os}]

        self.vapi.qos_egress_map_update(1, rows)

        sub_if = VppDot1QSubint(self, self.pg0, 11)

        sub_if.admin_up()
        sub_if.config_ip4()
        sub_if.resolve_arp()
        sub_if.config_ip6()
        sub_if.resolve_ndp()

        #
        # enable VLAN QoS recording/marking on the input Pg0 subinterface and
        #
        self.vapi.qos_record_enable_disable(sub_if.sw_if_index,
                                            QOS_SOURCE.VLAN,
                                            1)
        self.vapi.qos_mark_enable_disable(sub_if.sw_if_index,
                                          QOS_SOURCE.VLAN,
                                          1,
                                          1)

        #
        # IP marking/recording on pg1
        #
        self.vapi.qos_record_enable_disable(self.pg1.sw_if_index,
                                            QOS_SOURCE.IP,
                                            1)
        self.vapi.qos_mark_enable_disable(self.pg1.sw_if_index,
                                          QOS_SOURCE.IP,
                                          1,
                                          1)

        #
        # a routes to/from sub-interface
        #
        route_10_0_0_1 = VppIpRoute(self, "10.0.0.1", 32,
                                    [VppRoutePath(sub_if.remote_ip4,
                                                  sub_if.sw_if_index)])
        route_10_0_0_1.add_vpp_config()
        route_10_0_0_2 = VppIpRoute(self, "10.0.0.2", 32,
                                    [VppRoutePath(self.pg1.remote_ip4,
                                                  self.pg1.sw_if_index)])
        route_10_0_0_2.add_vpp_config()
        route_2001_1 = VppIpRoute(self, "2001::1", 128,
                                  [VppRoutePath(sub_if.remote_ip6,
                                                sub_if.sw_if_index,
                                                proto=DpoProto.DPO_PROTO_IP6)],
                                  is_ip6=1)
        route_2001_1.add_vpp_config()
        route_2001_2 = VppIpRoute(self, "2001::2", 128,
                                  [VppRoutePath(self.pg1.remote_ip6,
                                                self.pg1.sw_if_index,
                                                proto=DpoProto.DPO_PROTO_IP6)],
                                  is_ip6=1)
        route_2001_2.add_vpp_config()

        p_v1 = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) /
                Dot1Q(vlan=11, prio=1) /
                IP(src="1.1.1.1", dst="10.0.0.2", tos=1) /
                UDP(sport=1234, dport=1234) /
                Raw(scapy.compat.chb(100) * 65))

        p_v2 = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) /
                IP(src="1.1.1.1", dst="10.0.0.1", tos=1) /
                UDP(sport=1234, dport=1234) /
                Raw(scapy.compat.chb(100) * 65))

        rx = self.send_and_expect(self.pg1, p_v2 * 65, self.pg0)

        for p in rx:
            self.assertEqual(p[Dot1Q].prio, 6)

        rx = self.send_and_expect(self.pg0, p_v1 * 65, self.pg1)

        for p in rx:
            self.assertEqual(p[IP].tos, 254)

        p_v1 = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) /
                Dot1Q(vlan=11, prio=2) /
                IPv6(src="2001::1", dst="2001::2", tc=1) /
                UDP(sport=1234, dport=1234) /
                Raw(scapy.compat.chb(100) * 65))

        p_v2 = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) /
                IPv6(src="3001::1", dst="2001::1", tc=1) /
                UDP(sport=1234, dport=1234) /
                Raw(scapy.compat.chb(100) * 65))

        rx = self.send_and_expect(self.pg1, p_v2 * 65, self.pg0)

        for p in rx:
            self.assertEqual(p[Dot1Q].prio, 6)

        rx = self.send_and_expect(self.pg0, p_v1 * 65, self.pg1)

        for p in rx:
            self.assertEqual(p[IPv6].tc, 253)

        #
        # cleanup
        #
        sub_if.unconfig_ip4()
        sub_if.unconfig_ip6()

        self.vapi.qos_record_enable_disable(sub_if.sw_if_index,
                                            QOS_SOURCE.VLAN,
                                            0)
        self.vapi.qos_mark_enable_disable(sub_if.sw_if_index,
                                          QOS_SOURCE.VLAN,
                                          1,
                                          0)
        self.vapi.qos_record_enable_disable(self.pg1.sw_if_index,
                                            QOS_SOURCE.IP,
                                            0)
        self.vapi.qos_mark_enable_disable(self.pg1.sw_if_index,
                                          QOS_SOURCE.IP,
                                          1,
                                          0)


if __name__ == '__main__':
    unittest.main(testRunner=VppTestRunner)