5 from framework import VppTestCase, VppTestRunner
6 from vpp_papi_provider import QOS_SOURCE
7 from vpp_ip_route import VppIpRoute, VppRoutePath
9 from scapy.packet import Raw
10 from scapy.layers.l2 import Ether
11 from scapy.layers.inet import IP, UDP
12 from scapy.layers.inet6 import IPv6
13 from scapy.contrib.mpls import MPLS
16 class TestQOS(VppTestCase):
20 super(TestQOS, self).setUp()
22 self.create_pg_interfaces(range(5))
24 for i in self.pg_interfaces:
32 for i in self.pg_interfaces:
36 super(TestQOS, self).tearDown()
38 def test_qos_ip(self):
42 # for table 1 map the n=0xff possible values of input QoS mark,
45 output = [chr(0)] * 256
46 for i in range(0, 255):
47 output[i] = chr(255 - i)
49 rows = [{'outputs': os},
54 self.vapi.qos_egress_map_update(1, rows)
57 # For table 2 (and up) use the value n for everything
59 output = [chr(2)] * 256
61 rows = [{'outputs': os},
66 self.vapi.qos_egress_map_update(2, rows)
68 output = [chr(3)] * 256
70 rows = [{'outputs': os},
75 self.vapi.qos_egress_map_update(3, rows)
77 output = [chr(4)] * 256
79 rows = [{'outputs': os},
83 self.vapi.qos_egress_map_update(4, rows)
84 self.vapi.qos_egress_map_update(5, rows)
85 self.vapi.qos_egress_map_update(6, rows)
86 self.vapi.qos_egress_map_update(7, rows)
88 self.logger.info(self.vapi.cli("sh qos eg map"))
91 # Bind interface pgN to table n
93 self.vapi.qos_mark_enable_disable(self.pg1.sw_if_index,
97 self.vapi.qos_mark_enable_disable(self.pg2.sw_if_index,
101 self.vapi.qos_mark_enable_disable(self.pg3.sw_if_index,
105 self.vapi.qos_mark_enable_disable(self.pg4.sw_if_index,
111 # packets ingress on Pg0
113 p_v4 = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) /
114 IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4, tos=1) /
115 UDP(sport=1234, dport=1234) /
117 p_v6 = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) /
118 IPv6(src=self.pg0.remote_ip6, dst=self.pg1.remote_ip6,
120 UDP(sport=1234, dport=1234) /
124 # Since we have not yet enabled the recording of the input QoS
125 # from the input iP header, the egress packet's ToS will be unchanged
127 rx = self.send_and_expect(self.pg0, p_v4 * 65, self.pg1)
129 self.assertEqual(p[IP].tos, 1)
130 rx = self.send_and_expect(self.pg0, p_v6 * 65, self.pg1)
132 self.assertEqual(p[IPv6].tc, 1)
135 # Enable QoS recrding on IP input for pg0
137 self.vapi.qos_record_enable_disable(self.pg0.sw_if_index,
142 # send the same packets, this time expect the input TOS of 1
143 # to be mapped to pg1's egress value of 254
145 rx = self.send_and_expect(self.pg0, p_v4 * 65, self.pg1)
147 self.assertEqual(p[IP].tos, 254)
148 rx = self.send_and_expect(self.pg0, p_v6 * 65, self.pg1)
150 self.assertEqual(p[IPv6].tc, 254)
153 # different input ToS to test the mapping
156 rx = self.send_and_expect(self.pg0, p_v4 * 65, self.pg1)
158 self.assertEqual(p[IP].tos, 128)
160 rx = self.send_and_expect(self.pg0, p_v6 * 65, self.pg1)
162 self.assertEqual(p[IPv6].tc, 128)
165 rx = self.send_and_expect(self.pg0, p_v4 * 65, self.pg1)
167 self.assertEqual(p[IP].tos, 1)
169 rx = self.send_and_expect(self.pg0, p_v6 * 65, self.pg1)
171 self.assertEqual(p[IPv6].tc, 1)
174 # send packets out the other interfaces to test the maps are
177 p_v4[IP].dst = self.pg2.remote_ip4
178 rx = self.send_and_expect(self.pg0, p_v4 * 65, self.pg2)
180 self.assertEqual(p[IP].tos, 2)
182 p_v4[IP].dst = self.pg3.remote_ip4
183 rx = self.send_and_expect(self.pg0, p_v4 * 65, self.pg3)
185 self.assertEqual(p[IP].tos, 3)
187 p_v6[IPv6].dst = self.pg3.remote_ip6
188 rx = self.send_and_expect(self.pg0, p_v6 * 65, self.pg3)
190 self.assertEqual(p[IPv6].tc, 3)
193 # remove the map on pg2 and pg3, now expect an unchanged IP tos
195 self.vapi.qos_mark_enable_disable(self.pg2.sw_if_index,
199 self.vapi.qos_mark_enable_disable(self.pg3.sw_if_index,
203 self.logger.info(self.vapi.cli("sh int feat pg2"))
205 p_v4[IP].dst = self.pg2.remote_ip4
206 rx = self.send_and_expect(self.pg0, p_v4 * 65, self.pg2)
208 self.assertEqual(p[IP].tos, 254)
210 p_v4[IP].dst = self.pg3.remote_ip4
211 rx = self.send_and_expect(self.pg0, p_v4 * 65, self.pg3)
213 self.assertEqual(p[IP].tos, 254)
216 # still mapping out of pg1
218 p_v4[IP].dst = self.pg1.remote_ip4
219 rx = self.send_and_expect(self.pg0, p_v4 * 65, self.pg1)
221 self.assertEqual(p[IP].tos, 1)
224 # disable the input recording on pg0
226 self.vapi.qos_record_enable_disable(self.pg0.sw_if_index,
231 # back to an unchanged TOS value
233 rx = self.send_and_expect(self.pg0, p_v4 * 65, self.pg1)
235 self.assertEqual(p[IP].tos, 254)
238 # disable the egress map on pg1 and pg4
240 self.vapi.qos_mark_enable_disable(self.pg1.sw_if_index,
244 self.vapi.qos_mark_enable_disable(self.pg4.sw_if_index,
250 # unchanged Tos on pg1
252 rx = self.send_and_expect(self.pg0, p_v4 * 65, self.pg1)
254 self.assertEqual(p[IP].tos, 254)
259 self.vapi.qos_egress_map_delete(1)
260 self.vapi.qos_egress_map_delete(4)
261 self.vapi.qos_egress_map_delete(2)
262 self.vapi.qos_egress_map_delete(3)
263 self.vapi.qos_egress_map_delete(5)
264 self.vapi.qos_egress_map_delete(6)
265 self.vapi.qos_egress_map_delete(7)
267 def test_qos_mpls(self):
268 """ QoS Mark MPLS """
271 # 255 QoS for all input values
273 output = [chr(255)] * 256
275 rows = [{'outputs': os},
280 self.vapi.qos_egress_map_update(1, rows)
283 # a route with 1 MPLS label
285 route_10_0_0_1 = VppIpRoute(self, "10.0.0.1", 32,
286 [VppRoutePath(self.pg1.remote_ip4,
287 self.pg1.sw_if_index,
289 route_10_0_0_1.add_vpp_config()
292 # a route with 3 MPLS labels
294 route_10_0_0_3 = VppIpRoute(self, "10.0.0.3", 32,
295 [VppRoutePath(self.pg1.remote_ip4,
296 self.pg1.sw_if_index,
297 labels=[63, 33, 34])])
298 route_10_0_0_3.add_vpp_config()
301 # enable IP QoS recording on the input Pg0 and MPLS egress marking
304 self.vapi.qos_record_enable_disable(self.pg0.sw_if_index,
307 self.vapi.qos_mark_enable_disable(self.pg1.sw_if_index,
313 # packet that will get one label added and 3 labels added resp.
315 p_1 = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) /
316 IP(src=self.pg0.remote_ip4, dst="10.0.0.1", tos=1) /
317 UDP(sport=1234, dport=1234) /
319 p_3 = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) /
320 IP(src=self.pg0.remote_ip4, dst="10.0.0.3", tos=1) /
321 UDP(sport=1234, dport=1234) /
324 rx = self.send_and_expect(self.pg0, p_1 * 65, self.pg1)
327 # only 3 bits of ToS value in MPLS make sure tos is correct
328 # and the label and EOS bit have not been corrupted
331 self.assertEqual(p[MPLS].cos, 7)
332 self.assertEqual(p[MPLS].label, 32)
333 self.assertEqual(p[MPLS].s, 1)
334 rx = self.send_and_expect(self.pg0, p_3 * 65, self.pg1)
336 self.assertEqual(p[MPLS].cos, 7)
337 self.assertEqual(p[MPLS].label, 63)
338 self.assertEqual(p[MPLS].s, 0)
340 self.assertEqual(h[MPLS].cos, 7)
341 self.assertEqual(h[MPLS].label, 33)
342 self.assertEqual(h[MPLS].s, 0)
344 self.assertEqual(h[MPLS].cos, 7)
345 self.assertEqual(h[MPLS].label, 34)
346 self.assertEqual(h[MPLS].s, 1)
351 self.vapi.qos_record_enable_disable(self.pg0.sw_if_index,
354 self.vapi.qos_mark_enable_disable(self.pg1.sw_if_index,
358 self.vapi.qos_egress_map_delete(1)
361 if __name__ == '__main__':
362 unittest.main(testRunner=VppTestRunner)