in ptf/saiqosmap.py [0:0]
def l3IPv4QosSameDscpToTcMappingManyPortsTest(self):
'''
L3 IPv4 same DSCP to TC mapping for various ingress ports test.
'''
print("l3IPv4QosSameDscpToTcMappingManyPortsTest")
# default DSCP to TC mapping
test_cases = []
test_cases.append({
'tx_port': self.dev_port0,
'dscp': 0,
'tc': 7,
'queue': 7
})
test_cases.append({
'tx_port': self.dev_port0,
'dscp': 1,
'tc': 6,
'queue': 6
})
test_cases.append({
'tx_port': self.dev_port0,
'dscp': 2,
'tc': 5,
'queue': 5
})
test_cases.append({
'tx_port': self.dev_port0,
'dscp': 3,
'tc': 4,
'queue': 4
})
test_cases.append({
'tx_port': self.dev_port0,
'dscp': 4,
'tc': 3,
'queue': 3
})
test_cases.append({
'tx_port': self.dev_port0,
'dscp': 5,
'tc': 2,
'queue': 2
})
test_cases.append({
'tx_port': self.dev_port0,
'dscp': 6,
'tc': 1,
'queue': 1
})
test_cases.append({
'tx_port': self.dev_port0,
'dscp': 7,
'tc': 0,
'queue': 0
})
# fall info default mapping
test_cases.append({
'tx_port': self.dev_port0,
'dscp': 8,
'tc': 0,
'queue': 0
})
# fall info default mapping
test_cases.append({
'tx_port': self.dev_port0,
'dscp': 9,
'tc': 0,
'queue': 0
})
# fall info default mapping
test_cases.append({
'tx_port': self.dev_port0,
'dscp': 10,
'tc': 0,
'queue': 0
})
test_cases.append({
'tx_port': self.dev_port2,
'dscp': 0,
'tc': 7,
'queue': 7
})
test_cases.append({
'tx_port': self.dev_port2,
'dscp': 1,
'tc': 6,
'queue': 6
})
test_cases.append({
'tx_port': self.dev_port2,
'dscp': 2,
'tc': 5,
'queue': 5
})
test_cases.append({
'tx_port': self.dev_port2,
'dscp': 3,
'tc': 4,
'queue': 4
})
test_cases.append({
'tx_port': self.dev_port2,
'dscp': 4,
'tc': 3,
'queue': 3
})
test_cases.append({
'tx_port': self.dev_port2,
'dscp': 5,
'tc': 2,
'queue': 2
})
test_cases.append({
'tx_port': self.dev_port2,
'dscp': 6,
'tc': 1,
'queue': 1
})
test_cases.append({
'tx_port': self.dev_port2,
'dscp': 7,
'tc': 0,
'queue': 0
})
# fall info default mapping
test_cases.append({
'tx_port': self.dev_port2,
'dscp': 8,
'tc': 0,
'queue': 0
})
# fall info default mapping
test_cases.append({
'tx_port': self.dev_port2,
'dscp': 9,
'tc': 0,
'queue': 0
})
# fall info default mapping
test_cases.append({
'tx_port': self.dev_port2,
'dscp': 10,
'tc': 0,
'queue': 0
})
test_cases.append({
'tx_port': self.dev_port3,
'dscp': 0,
'tc': 7,
'queue': 7
})
test_cases.append({
'tx_port': self.dev_port3,
'dscp': 1,
'tc': 6,
'queue': 6
})
test_cases.append({
'tx_port': self.dev_port3,
'dscp': 2,
'tc': 5,
'queue': 5
})
test_cases.append({
'tx_port': self.dev_port3,
'dscp': 3,
'tc': 4,
'queue': 4
})
test_cases.append({
'tx_port': self.dev_port3,
'dscp': 4,
'tc': 3,
'queue': 3
})
test_cases.append({
'tx_port': self.dev_port3,
'dscp': 5,
'tc': 2,
'queue': 2
})
test_cases.append({
'tx_port': self.dev_port3,
'dscp': 6,
'tc': 1,
'queue': 1
})
test_cases.append({
'tx_port': self.dev_port3,
'dscp': 7,
'tc': 0,
'queue': 0
})
# fall info default mapping
test_cases.append({
'tx_port': self.dev_port3,
'dscp': 8,
'tc': 0,
'queue': 0
})
# fall info default mapping
test_cases.append({
'tx_port': self.dev_port3,
'dscp': 9,
'tc': 0,
'queue': 0
})
# fall info default mapping
test_cases.append({
'tx_port': self.dev_port3,
'dscp': 10,
'tc': 0,
'queue': 0
})
ingress_ports = [self.port0, self.port2, self.port3]
egress_ports = [self.port1]
try:
# setup dscp to tc mapping
# Port 0, 2 and 3 qos map
ingress_dscp_list = [0, 1, 2, 3, 4, 5, 6, 7]
ingress_tc_list = [7, 6, 5, 4, 3, 2, 1, 0]
qos_dscp_to_tc_map_id = create_and_verify_qos_map(
self.client, SAI_QOS_MAP_TYPE_DSCP_TO_TC, ingress_dscp_list,
ingress_tc_list)
self.assertTrue(qos_dscp_to_tc_map_id != 0,
"Failed to create qos_map")
# setup Port 1 tc to queue index mapping
p1_ingress_tc_list = [0, 1, 2, 3, 4, 5, 6, 7]
p1_ingress_queue_list = [0, 1, 2, 3, 4, 5, 6, 7]
qos_tc_to_queue_map_id = create_and_verify_qos_map(
self.client, SAI_QOS_MAP_TYPE_TC_TO_QUEUE, p1_ingress_tc_list,
p1_ingress_queue_list)
self.assertTrue(qos_tc_to_queue_map_id != 0,
"Failed to create qos_map")
for port in ingress_ports:
status = sai_thrift_set_port_attribute(
self.client,
port,
qos_dscp_to_tc_map=qos_dscp_to_tc_map_id)
self.assertEqual(status, SAI_STATUS_SUCCESS,
"Failed to set port attribute")
attr = sai_thrift_get_port_attribute(self.client,
port,
qos_dscp_to_tc_map=True)
self.assertEqual(attr['qos_dscp_to_tc_map'],
qos_dscp_to_tc_map_id)
for port in egress_ports:
status = sai_thrift_set_port_attribute(
self.client,
port,
qos_tc_to_queue_map=qos_tc_to_queue_map_id)
self.assertEqual(status, SAI_STATUS_SUCCESS,
"Failed to set port attribute")
attr = sai_thrift_get_port_attribute(self.client,
port,
qos_tc_to_queue_map=True)
self.assertEqual(attr['qos_tc_to_queue_map'],
qos_tc_to_queue_map_id)
for test in test_cases:
ip_tos = test['dscp'] << 2
pkt = simple_tcp_packet(eth_dst=ROUTER_MAC,
eth_src='00:22:22:22:22:22',
ip_dst='10.10.10.11',
ip_src='192.168.0.1',
ip_tos=ip_tos,
ip_id=105,
ip_ttl=64)
exp_pkt = simple_tcp_packet(eth_dst='00:11:22:33:44:04',
eth_src=ROUTER_MAC,
ip_dst='10.10.10.11',
ip_src='192.168.0.1',
ip_tos=ip_tos,
ip_id=105,
ip_ttl=63)
initial_q_cnt = self.getPortQueueIndexStats(
self.port1, test['queue'], ['SAI_QUEUE_STAT_PACKETS'])
print("Sending packet port %d (ip_tos=%d) ->"
" port %d (tc=%d => queue=%d)"
% (test['tx_port'], ip_tos, self.dev_port1, test['tc'],
test['queue']))
send_packet(self, test['tx_port'], pkt)
verify_packet(self, exp_pkt, self.dev_port1)
post_q_cnt = self.getPortQueueIndexStats(
self.port1, test['queue'], ['SAI_QUEUE_STAT_PACKETS'])
self.assertEqual(
initial_q_cnt + 1, post_q_cnt,
'queue {} packets counter {} != {}'.format(
test['queue'], initial_q_cnt + 1, post_q_cnt))
finally:
for port in self.test_ports:
status = sai_thrift_set_port_attribute(
self.client, port, qos_dscp_to_tc_map=int(
SAI_NULL_OBJECT_ID))
self.assertEqual(status, SAI_STATUS_SUCCESS,
"Failed to set port attribute")
status = sai_thrift_set_port_attribute(
self.client, port, qos_tc_to_queue_map=int(
SAI_NULL_OBJECT_ID))
self.assertEqual(status, SAI_STATUS_SUCCESS,
"Failed to set port attribute")
sai_thrift_remove_qos_map(self.client, qos_tc_to_queue_map_id)
sai_thrift_remove_qos_map(self.client, qos_dscp_to_tc_map_id)