diff options
Diffstat (limited to 'test/test_qos.py')
| -rw-r--r-- | test/test_qos.py | 345 |
1 files changed, 190 insertions, 155 deletions
diff --git a/test/test_qos.py b/test/test_qos.py index 02941a163b4..53597858238 100644 --- a/test/test_qos.py +++ b/test/test_qos.py @@ -2,11 +2,16 @@ import unittest -from framework import VppTestCase, VppTestRunner +from framework import VppTestCase +from asfframework import VppTestRunner from vpp_sub_interface import VppDot1QSubint -from vpp_ip import DpoProto -from vpp_ip_route import VppIpRoute, VppRoutePath, VppMplsRoute, \ - VppMplsLabel, VppMplsTable, FibPathProto +from vpp_ip_route import ( + VppIpRoute, + VppRoutePath, + VppMplsRoute, + VppMplsLabel, + VppMplsTable, +) import scapy.compat from scapy.packet import Raw @@ -21,7 +26,7 @@ NUM_PKTS = 67 class TestQOS(VppTestCase): - """ QOS Test Case """ + """QOS Test Case""" # Note: Since the enums aren't created dynamically until after # the papi client attaches to VPP, we put it in a property to @@ -63,7 +68,7 @@ class TestQOS(VppTestCase): super(TestQOS, self).tearDown() def test_qos_ip(self): - """ QoS Mark/Record/Store IP """ + """QoS Mark/Record/Store IP""" # # for table 1 map the n=0xff possible values of input QoS mark, @@ -72,11 +77,8 @@ class TestQOS(VppTestCase): 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}] + os = b"".join(output) + rows = [{"outputs": os}, {"outputs": os}, {"outputs": os}, {"outputs": os}] qem1 = VppQosEgressMap(self, 1, rows).add_vpp_config() @@ -84,29 +86,20 @@ class TestQOS(VppTestCase): # 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}] + os = b"".join(output) + rows = [{"outputs": os}, {"outputs": os}, {"outputs": os}, {"outputs": os}] qem2 = VppQosEgressMap(self, 2, rows).add_vpp_config() output = [scapy.compat.chb(3)] * 256 - os = b''.join(output) - rows = [{'outputs': os}, - {'outputs': os}, - {'outputs': os}, - {'outputs': os}] + os = b"".join(output) + rows = [{"outputs": os}, {"outputs": os}, {"outputs": os}, {"outputs": os}] qem3 = VppQosEgressMap(self, 3, rows).add_vpp_config() output = [scapy.compat.chb(4)] * 256 - os = b''.join(output) - rows = [{'outputs': os}, - {'outputs': os}, - {'outputs': os}, - {'outputs': os}] + os = b"".join(output) + rows = [{"outputs": os}, {"outputs": os}, {"outputs": os}, {"outputs": os}] qem4 = VppQosEgressMap(self, 4, rows).add_vpp_config() qem5 = VppQosEgressMap(self, 5, rows).add_vpp_config() @@ -119,14 +112,18 @@ class TestQOS(VppTestCase): # # Bind interface pgN to table n # - qm1 = VppQosMark(self, self.pg1, qem1, - self.QOS_SOURCE.QOS_API_SOURCE_IP).add_vpp_config() - qm2 = VppQosMark(self, self.pg2, qem2, - self.QOS_SOURCE.QOS_API_SOURCE_IP).add_vpp_config() - qm3 = VppQosMark(self, self.pg3, qem3, - self.QOS_SOURCE.QOS_API_SOURCE_IP).add_vpp_config() - qm4 = VppQosMark(self, self.pg4, qem4, - self.QOS_SOURCE.QOS_API_SOURCE_IP).add_vpp_config() + qm1 = VppQosMark( + self, self.pg1, qem1, self.QOS_SOURCE.QOS_API_SOURCE_IP + ).add_vpp_config() + qm2 = VppQosMark( + self, self.pg2, qem2, self.QOS_SOURCE.QOS_API_SOURCE_IP + ).add_vpp_config() + qm3 = VppQosMark( + self, self.pg3, qem3, self.QOS_SOURCE.QOS_API_SOURCE_IP + ).add_vpp_config() + qm4 = VppQosMark( + self, self.pg4, qem4, self.QOS_SOURCE.QOS_API_SOURCE_IP + ).add_vpp_config() self.assertTrue(qm3.query_vpp_config()) self.logger.info(self.vapi.cli("sh qos mark")) @@ -134,15 +131,18 @@ class TestQOS(VppTestCase): # # 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) * NUM_PKTS)) - 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) * NUM_PKTS)) + 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) * NUM_PKTS) + ) + 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) * NUM_PKTS) + ) # # Since we have not yet enabled the recording of the input QoS @@ -158,8 +158,7 @@ class TestQOS(VppTestCase): # # Enable QoS recording on IP input for pg0 # - qr1 = VppQosRecord(self, self.pg0, - self.QOS_SOURCE.QOS_API_SOURCE_IP) + qr1 = VppQosRecord(self, self.pg0, self.QOS_SOURCE.QOS_API_SOURCE_IP) qr1.add_vpp_config() self.logger.info(self.vapi.cli("sh qos record")) @@ -258,9 +257,9 @@ class TestQOS(VppTestCase): # # enable QoS stroe instead of record # - qst1 = VppQosStore(self, self.pg0, - self.QOS_SOURCE.QOS_API_SOURCE_IP, - 5).add_vpp_config() + qst1 = VppQosStore( + self, self.pg0, self.QOS_SOURCE.QOS_API_SOURCE_IP, 5 + ).add_vpp_config() self.logger.info(self.vapi.cli("sh qos store")) p_v4[IP].dst = self.pg1.remote_ip4 @@ -295,7 +294,7 @@ class TestQOS(VppTestCase): self.assertEqual(p[IP].tos, 254) def test_qos_mpls(self): - """ QoS Mark/Record MPLS """ + """QoS Mark/Record MPLS""" # # 255 QoS for all input values @@ -305,58 +304,69 @@ class TestQOS(VppTestCase): from_mpls = 5 from_vlan = 4 output = [scapy.compat.chb(from_ext)] * 256 - os1 = b''.join(output) + os1 = b"".join(output) output = [scapy.compat.chb(from_vlan)] * 256 - os2 = b''.join(output) + os2 = b"".join(output) output = [scapy.compat.chb(from_mpls)] * 256 - os3 = b''.join(output) + os3 = b"".join(output) output = [scapy.compat.chb(from_ip)] * 256 - os4 = b''.join(output) - rows = [{'outputs': os1}, - {'outputs': os2}, - {'outputs': os3}, - {'outputs': os4}] + os4 = b"".join(output) + rows = [{"outputs": os1}, {"outputs": os2}, {"outputs": os3}, {"outputs": os4}] qem1 = VppQosEgressMap(self, 1, rows).add_vpp_config() # # 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 = 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 = 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 # - qr1 = VppQosRecord(self, self.pg0, - self.QOS_SOURCE.QOS_API_SOURCE_IP).add_vpp_config() - qm1 = VppQosMark(self, self.pg1, qem1, - self.QOS_SOURCE.QOS_API_SOURCE_MPLS).add_vpp_config() + qr1 = VppQosRecord( + self, self.pg0, self.QOS_SOURCE.QOS_API_SOURCE_IP + ).add_vpp_config() + qm1 = VppQosMark( + self, self.pg1, qem1, self.QOS_SOURCE.QOS_API_SOURCE_MPLS + ).add_vpp_config() # # 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) * NUM_PKTS)) - 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) * NUM_PKTS)) + 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) * NUM_PKTS) + ) + 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) * NUM_PKTS) + ) rx = self.send_and_expect(self.pg0, p_1 * NUM_PKTS, self.pg1) @@ -387,26 +397,34 @@ class TestQOS(VppTestCase): # on Pg1 # qr2 = VppQosRecord( - self, self.pg0, - self.QOS_SOURCE.QOS_API_SOURCE_MPLS).add_vpp_config() + self, self.pg0, self.QOS_SOURCE.QOS_API_SOURCE_MPLS + ).add_vpp_config() qm2 = VppQosMark( - self, self.pg1, qem1, - self.QOS_SOURCE.QOS_API_SOURCE_IP).add_vpp_config() + self, self.pg1, qem1, self.QOS_SOURCE.QOS_API_SOURCE_IP + ).add_vpp_config() # # 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 = 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) * NUM_PKTS)) + 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) * NUM_PKTS) + ) rx = self.send_and_expect(self.pg0, p_m1 * NUM_PKTS, self.pg1) for p in rx: @@ -417,22 +435,26 @@ class TestQOS(VppTestCase): # # 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 = 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 = 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) * NUM_PKTS)) + 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) * NUM_PKTS) + ) rx = self.send_and_expect(self.pg0, p_m2 * NUM_PKTS, self.pg1) @@ -440,7 +462,7 @@ class TestQOS(VppTestCase): self.assertEqual(p[IP].tos, from_mpls) def test_qos_vlan(self): - """QoS mark/record VLAN """ + """QoS mark/record VLAN""" # # QoS for all input values @@ -448,11 +470,8 @@ class TestQOS(VppTestCase): 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}] + os = b"".join(output) + rows = [{"outputs": os}, {"outputs": os}, {"outputs": os}, {"outputs": os}] qem1 = VppQosEgressMap(self, 1, rows).add_vpp_config() @@ -468,58 +487,70 @@ class TestQOS(VppTestCase): # enable VLAN QoS recording/marking on the input Pg0 subinterface and # qr_v = VppQosRecord( - self, sub_if, - self.QOS_SOURCE.QOS_API_SOURCE_VLAN).add_vpp_config() + self, sub_if, self.QOS_SOURCE.QOS_API_SOURCE_VLAN + ).add_vpp_config() qm_v = VppQosMark( - self, sub_if, qem1, - self.QOS_SOURCE.QOS_API_SOURCE_VLAN).add_vpp_config() + self, sub_if, qem1, self.QOS_SOURCE.QOS_API_SOURCE_VLAN + ).add_vpp_config() # # IP marking/recording on pg1 # qr_ip = VppQosRecord( - self, self.pg1, - self.QOS_SOURCE.QOS_API_SOURCE_IP).add_vpp_config() + self, self.pg1, self.QOS_SOURCE.QOS_API_SOURCE_IP + ).add_vpp_config() qm_ip = VppQosMark( - self, self.pg1, qem1, - self.QOS_SOURCE.QOS_API_SOURCE_IP).add_vpp_config() + self, self.pg1, qem1, self.QOS_SOURCE.QOS_API_SOURCE_IP + ).add_vpp_config() # # 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 = 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 = 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)]) + route_2001_1 = VppIpRoute( + self, "2001::1", 128, [VppRoutePath(sub_if.remote_ip6, sub_if.sw_if_index)] + ) route_2001_1.add_vpp_config() - route_2001_2 = VppIpRoute(self, "2001::2", 128, - [VppRoutePath(self.pg1.remote_ip6, - self.pg1.sw_if_index)]) + route_2001_2 = VppIpRoute( + self, + "2001::2", + 128, + [VppRoutePath(self.pg1.remote_ip6, self.pg1.sw_if_index)], + ) 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) * NUM_PKTS)) - - 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) * NUM_PKTS)) - - p_v3 = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) / - Dot1Q(vlan=11, prio=1, id=1) / - IP(src="1.1.1.1", dst="10.0.0.2", tos=2) / - UDP(sport=1234, dport=1234) / - Raw(scapy.compat.chb(100) * NUM_PKTS)) + 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) * NUM_PKTS) + ) + + 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) * NUM_PKTS) + ) + + p_v3 = ( + Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) + / Dot1Q(vlan=11, prio=1, id=1) + / IP(src="1.1.1.1", dst="10.0.0.2", tos=2) + / UDP(sport=1234, dport=1234) + / Raw(scapy.compat.chb(100) * NUM_PKTS) + ) rx = self.send_and_expect(self.pg1, p_v2 * NUM_PKTS, self.pg0) @@ -537,16 +568,20 @@ class TestQOS(VppTestCase): for p in rx: self.assertEqual(p[IP].tos, 253) - 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) * NUM_PKTS)) - - 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) * NUM_PKTS)) + 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) * NUM_PKTS) + ) + + 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) * NUM_PKTS) + ) rx = self.send_and_expect(self.pg1, p_v2 * NUM_PKTS, self.pg0) @@ -566,5 +601,5 @@ class TestQOS(VppTestCase): sub_if.unconfig_ip6() -if __name__ == '__main__': +if __name__ == "__main__": unittest.main(testRunner=VppTestRunner) |