in ptf/saiqosmap.py [0:0]
def l2QosMapPCPToTcDefaultMappingTest(self):
''' Test verifies the following:
1. default L2 PCP to TC mapping (no mapping defined)
The verification method:
- setting the TC to QUEUE mapping
- verification of the queue stats
'''
print("l2QosMapPCPToTcDefaultMappingTest")
# default PCP to TC mapping
test_cases = []
# port default_tc Test Case
test_cases.append({
'pcp': 0,
'port_default_tc': 0,
'queue': 0,
'color': 0
})
test_cases.append({
'pcp': 1,
'port_default_tc': 5,
'queue': 5,
'color': 0
})
test_cases.append({
'pcp': 2,
'port_default_tc': 0,
'queue': 0,
'color': 0
})
test_cases.append({
'pcp': 3,
'port_default_tc': 0,
'queue': 0,
'color': 0
})
test_cases.append({
'pcp': 4,
'port_default_tc': 0,
'queue': 0,
'color': 0
})
test_cases.append({
'pcp': 5,
'port_default_tc': 0,
'queue': 0,
'color': 0
})
test_cases.append({
'pcp': 6,
'port_default_tc': 0,
'queue': 0,
'color': 0
})
test_cases.append({
'pcp': 7,
'port_default_tc': 0,
'queue': 0,
'color': 0
})
test_cases.append({
'pcp': 0,
'port_default_tc': 1,
'queue': 1,
'color': 0
})
test_cases.append({
'pcp': 1,
'port_default_tc': 2,
'queue': 2,
'color': 0
})
test_cases.append({
'pcp': 2,
'port_default_tc': 3,
'queue': 3,
'color': 0
})
test_cases.append({
'pcp': 3,
'port_default_tc': 4,
'queue': 4,
'color': 0
})
test_cases.append({
'pcp': 4,
'port_default_tc': 5,
'queue': 5,
'color': 0
})
test_cases.append({
'pcp': 5,
'port_default_tc': 6,
'queue': 6,
'color': 0
})
test_cases.append({
'pcp': 6,
'port_default_tc': 7,
'queue': 7,
'color': 0
})
test_cases.append({
'pcp': 7,
'port_default_tc': 0,
'queue': 0,
'color': 0
})
# Test configured default_tc
# different default_tc per port
# port 0 -> default_tc = 5 ingress
# port 1 -> default_tc = 1 egress
# port 2 -> default_tc = 2 egress
# port 3 -> default_tc = 3 egress
test_cases.append({
'pcp': 0,
'port_default_tc': None,
'queue': 5,
'color': 0
})
test_cases.append({
'pcp': 1,
'port_default_tc': None,
'queue': 5,
'color': 0
})
test_cases.append({
'pcp': 2,
'port_default_tc': None,
'queue': 5,
'color': 0
})
test_cases.append({
'pcp': 3,
'port_default_tc': None,
'queue': 5,
'color': 0
})
test_cases.append({
'pcp': 4,
'port_default_tc': None,
'queue': 5,
'color': 0
})
test_cases.append({
'pcp': 5,
'port_default_tc': None,
'queue': 5,
'color': 0
})
test_cases.append({
'pcp': 6,
'port_default_tc': None,
'queue': 5,
'color': 0
})
test_cases.append({
'pcp': 7,
'port_default_tc': None,
'queue': 5,
'color': 0
})
try:
# setup tc to queue index mapping
ingress_tc_list = [0, 1, 2, 3, 4, 5, 6, 7]
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, ingress_tc_list,
ingress_queue_list)
self.assertTrue(qos_tc_to_queue_map_id != 0,
"Failed to create qos_map")
# assign to port the tc_queue and tc_color_to_pcp qos map
for port, tc_to_queue in [[self.port0, qos_tc_to_queue_map_id],
[self.port1, qos_tc_to_queue_map_id],
[self.port2, qos_tc_to_queue_map_id],
[self.port3, qos_tc_to_queue_map_id]]:
status = sai_thrift_set_port_attribute(
self.client, port, qos_tc_to_queue_map=tc_to_queue)
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'], tc_to_queue)
# verify port default tc
for port in [self.port0, self.port1, self.port2, self.port3]:
attr = sai_thrift_get_port_attribute(self.client,
port,
qos_dot1p_to_tc_map=True,
qos_default_tc=True)
self.assertEqual(attr['qos_dot1p_to_tc_map'], 0)
self.assertEqual(attr['qos_default_tc'], 0)
for test in test_cases:
queue = test['queue']
# setup port default_tc
if test['port_default_tc'] is not None:
# every test port gets the same default_tc =
# test['port_default_tc']
port_default_tc = test['port_default_tc']
port_config = [[self.port0, port_default_tc],
[self.port1, port_default_tc],
[self.port2, port_default_tc],
[self.port3, port_default_tc]]
else:
# every port gets default_tc as its port number
# port 0 -> default tc = 5
# port 1 -> default tc = 1
# port 2 -> default tc = 2
# port 3 -> default tc = 3
port_config = [[self.port0, 5], [self.port1, 1],
[self.port2, 2], [self.port3, 3]]
for port, port_default_tc in port_config:
sai_thrift_set_port_attribute(
self.client, port, qos_default_tc=port_default_tc)
attr = sai_thrift_get_port_attribute(self.client,
port,
qos_default_tc=True)
self.assertEqual(attr['qos_default_tc'], port_default_tc)
p1_initial_q_cnt = self.getPortQueueIndexStats(
self.port1, queue, ['SAI_QUEUE_STAT_PACKETS'])
p2_initial_q_cnt = self.getPortQueueIndexStats(
self.port2, queue, ['SAI_QUEUE_STAT_PACKETS'])
p3_initial_q_cnt = self.getPortQueueIndexStats(
self.port3, queue, ['SAI_QUEUE_STAT_PACKETS'])
if test['port_default_tc'] is not None:
port_default_tc = str(test['port_default_tc'])
else:
port_default_tc = str(1)
ingress_pcp = str(test['pcp'])
print("Sending packet port %d (default_tc=%s, pcp=%s)"
" -> port %d (queue=%d, pcp=%d)"
% (self.dev_port0, port_default_tc,
ingress_pcp, self.dev_port1, queue, 1))
print("\t\t\t\t\tport %d (queue=%d, pcp=%d))"
% (self.dev_port2, queue, 2))
print("\t\t\t\t\tport %d (queue=%d, pcp=%d))"
% (self.dev_port3, queue, 3))
ingress_pcp = test['pcp']
pkt = simple_eth_raw_packet_with_taglist(
pktlen=104,
eth_dst='00:33:33:33:33:00',
eth_src='00:33:33:33:33:11',
dl_taglist_enable=True,
dl_vlan_pcp_list=[ingress_pcp],
dl_vlanid_list=[1])
p1_exp_pkt = simple_eth_raw_packet_with_taglist(
pktlen=104,
eth_dst='00:33:33:33:33:00',
eth_src='00:33:33:33:33:11',
dl_taglist_enable=True,
dl_vlan_pcp_list=[1],
dl_vlanid_list=[1])
p2_exp_pkt = simple_eth_raw_packet_with_taglist(
pktlen=104,
eth_dst='00:33:33:33:33:00',
eth_src='00:33:33:33:33:11',
dl_taglist_enable=True,
dl_vlan_pcp_list=[1],
dl_vlanid_list=[1])
p3_exp_pkt = simple_eth_raw_packet_with_taglist(
pktlen=104,
eth_dst='00:33:33:33:33:00',
eth_src='00:33:33:33:33:11',
dl_taglist_enable=True,
dl_vlan_pcp_list=[3],
dl_vlanid_list=[1])
test_port = self.dev_port0
flood_port_list = [[self.dev_port1], [self.dev_port2],
[self.dev_port3]]
flood_pkt_list = [pkt, pkt, pkt]
send_packet(self, test_port, pkt)
verify_each_packet_on_multiple_port_lists(self, flood_pkt_list,
flood_port_list)
print("\tPacket flooded. ok")
p1_post_q_cnt = self.getPortQueueIndexStats(
self.port1, queue, ['SAI_QUEUE_STAT_PACKETS'])
p2_post_q_cnt = self.getPortQueueIndexStats(
self.port2, queue, ['SAI_QUEUE_STAT_PACKETS'])
p3_post_q_cnt = self.getPortQueueIndexStats(
self.port3, queue, ['SAI_QUEUE_STAT_PACKETS'])
self.assertEqual(
p1_initial_q_cnt + 1, p1_post_q_cnt,
'PORT1 queue {} packets counter {} != {}'.format(
queue, p1_initial_q_cnt + 1, p1_post_q_cnt))
self.assertEqual(
p2_initial_q_cnt + 1, p2_post_q_cnt,
'PORT2 queue {} packets counter {} != {}'.format(
queue, p2_initial_q_cnt + 1, p2_post_q_cnt))
self.assertEqual(
p3_initial_q_cnt + 1, p3_post_q_cnt,
'PORT3 queue {} packets counter {} != {}'.format(
queue, p3_initial_q_cnt + 1, p3_post_q_cnt))
finally:
for port in [self.port0, self.port1, self.port2, self.port3]:
sai_thrift_set_port_attribute(self.client,
port,
qos_tc_to_queue_map=0)
sai_thrift_set_port_attribute(self.client,
port,
qos_default_tc=0)
sai_thrift_remove_qos_map(self.client, qos_tc_to_queue_map_id)