in ptf/saiqosmap.py [0:0]
def l2QosMapDifferentPCPToColorMultiIngresssPortTest(self):
''' L2 traffic test verifies the following:
1. Many PCP to color mapping for various ingress ports
The verification method:
- SAI_QOS_MAP_TYPE_DOT1P_TO_COLOR mapping defined for ingress
and assigned to ports 0, 1, 2, and 3
- SAI_QOS_MAP_TYPE_TC_AND_COLOR_TO_DOT1P mapping defined for
ingress and assigned to ports 0, 1, 2, and 3
- various SAI_QOS_MAP_TYPE_TC_AND_COLOR_TO_DOT1P mappings
defined for egress and assigned to ports 0, 1, 2, and 3
egress pcp = 2 x egress port - 1 + color
'''
print("l2QosMapDifferentPCPToColorMultiIngresssPortTest")
# default PCP to TC mapping
test_cases = []
test_cases.append({
'ingress_port': self.dev_port0,
'pcp': 0,
'p0_pcp': None,
'p1_pcp': 1,
'p2_pcp': 3,
'p3_pcp': 5
})
test_cases.append({
'ingress_port': self.dev_port0,
'pcp': 1,
'p0_pcp': None,
'p1_pcp': 1,
'p2_pcp': 3,
'p3_pcp': 5
})
test_cases.append({
'ingress_port': self.dev_port0,
'pcp': 2,
'p0_pcp': None,
'p1_pcp': 1,
'p2_pcp': 3,
'p3_pcp': 5
})
test_cases.append({
'ingress_port': self.dev_port0,
'pcp': 3,
'p0_pcp': None,
'p1_pcp': 1,
'p2_pcp': 3,
'p3_pcp': 5
})
test_cases.append({
'ingress_port': self.dev_port0,
'pcp': 4,
'p0_pcp': None,
'p1_pcp': 1,
'p2_pcp': 3,
'p3_pcp': 5
})
test_cases.append({
'ingress_port': self.dev_port0,
'pcp': 5,
'p0_pcp': None,
'p1_pcp': 1,
'p2_pcp': 3,
'p3_pcp': 5
})
test_cases.append({
'ingress_port': self.dev_port0,
'pcp': 6,
'p0_pcp': None,
'p1_pcp': 1,
'p2_pcp': 3,
'p3_pcp': 5
})
test_cases.append({
'ingress_port': self.dev_port0,
'pcp': 7,
'p0_pcp': None,
'p1_pcp': 1,
'p2_pcp': 3,
'p3_pcp': 5
})
test_cases.append({
'ingress_port': self.dev_port1,
'pcp': 0,
'p0_pcp': 1,
'p1_pcp': None,
'p2_pcp': 3,
'p3_pcp': 5
})
test_cases.append({
'ingress_port': self.dev_port1,
'pcp': 1,
'p0_pcp': 1,
'p1_pcp': None,
'p2_pcp': 3,
'p3_pcp': 5
})
test_cases.append({
'ingress_port': self.dev_port1,
'pcp': 2,
'p0_pcp': 1,
'p1_pcp': None,
'p2_pcp': 3,
'p3_pcp': 5
})
test_cases.append({
'ingress_port': self.dev_port1,
'pcp': 3,
'p0_pcp': 1,
'p1_pcp': None,
'p2_pcp': 3,
'p3_pcp': 5
})
test_cases.append({
'ingress_port': self.dev_port1,
'pcp': 4,
'p0_pcp': 1,
'p1_pcp': None,
'p2_pcp': 3,
'p3_pcp': 5
})
test_cases.append({
'ingress_port': self.dev_port1,
'pcp': 5,
'p0_pcp': 1,
'p1_pcp': None,
'p2_pcp': 3,
'p3_pcp': 5
})
test_cases.append({
'ingress_port': self.dev_port1,
'pcp': 6,
'p0_pcp': 1,
'p1_pcp': None,
'p2_pcp': 3,
'p3_pcp': 5
})
test_cases.append({
'ingress_port': self.dev_port1,
'pcp': 7,
'p0_pcp': 1,
'p1_pcp': None,
'p2_pcp': 3,
'p3_pcp': 5
})
test_cases.append({
'ingress_port': self.dev_port2,
'pcp': 0,
'p0_pcp': 2,
'p1_pcp': 2,
'p2_pcp': None,
'p3_pcp': 6
})
test_cases.append({
'ingress_port': self.dev_port2,
'pcp': 1,
'p0_pcp': 2,
'p1_pcp': 2,
'p2_pcp': None,
'p3_pcp': 6
})
test_cases.append({
'ingress_port': self.dev_port2,
'pcp': 2,
'p0_pcp': 2,
'p1_pcp': 2,
'p2_pcp': None,
'p3_pcp': 6
})
test_cases.append({
'ingress_port': self.dev_port2,
'pcp': 3,
'p0_pcp': 2,
'p1_pcp': 2,
'p2_pcp': None,
'p3_pcp': 6
})
test_cases.append({
'ingress_port': self.dev_port2,
'pcp': 4,
'p0_pcp': 2,
'p1_pcp': 2,
'p2_pcp': None,
'p3_pcp': 6
})
test_cases.append({
'ingress_port': self.dev_port2,
'pcp': 5,
'p0_pcp': 2,
'p1_pcp': 2,
'p2_pcp': None,
'p3_pcp': 6
})
test_cases.append({
'ingress_port': self.dev_port2,
'pcp': 6,
'p0_pcp': 2,
'p1_pcp': 2,
'p2_pcp': None,
'p3_pcp': 6
})
test_cases.append({
'ingress_port': self.dev_port2,
'pcp': 7,
'p0_pcp': 2,
'p1_pcp': 2,
'p2_pcp': None,
'p3_pcp': 6
})
test_cases.append({
'ingress_port': self.dev_port3,
'pcp': 0,
'p0_pcp': 3,
'p1_pcp': 3,
'p2_pcp': 5,
'p3_pcp': None
})
test_cases.append({
'ingress_port': self.dev_port3,
'pcp': 1,
'p0_pcp': 3,
'p1_pcp': 3,
'p2_pcp': 5,
'p3_pcp': None
})
test_cases.append({
'ingress_port': self.dev_port3,
'pcp': 2,
'p0_pcp': 3,
'p1_pcp': 3,
'p2_pcp': 5,
'p3_pcp': None
})
test_cases.append({
'ingress_port': self.dev_port3,
'pcp': 3,
'p0_pcp': 3,
'p1_pcp': 3,
'p2_pcp': 5,
'p3_pcp': None
})
test_cases.append({
'ingress_port': self.dev_port3,
'pcp': 4,
'p0_pcp': 3,
'p1_pcp': 3,
'p2_pcp': 5,
'p3_pcp': None
})
test_cases.append({
'ingress_port': self.dev_port3,
'pcp': 5,
'p0_pcp': 3,
'p1_pcp': 3,
'p2_pcp': 5,
'p3_pcp': None
})
test_cases.append({
'ingress_port': self.dev_port3,
'pcp': 6,
'p0_pcp': 3,
'p1_pcp': 3,
'p2_pcp': 5,
'p3_pcp': None
})
test_cases.append({
'ingress_port': self.dev_port3,
'pcp': 7,
'p0_pcp': 3,
'p1_pcp': 3,
'p2_pcp': 5,
'p3_pcp': None
})
for port, dot1p_to_tc, dot1p_color, tc_color_to_pcp in [
[self.port0, self.qos_dot1p_to_tc_map_id,
self.p1_qos_dot1p_to_color_map_id,
self.p1_qos_tc_color_to_dot1q_id],
[self.port1, self.qos_dot1p_to_tc_map_id,
self.p1_qos_dot1p_to_color_map_id,
self.p1_qos_tc_color_to_dot1q_id],
[self.port2, self.qos_dot1p_to_tc_map_id,
self.p2_qos_dot1p_to_color_map_id,
self.p2_qos_tc_color_to_dot1q_id],
[self.port3, self.qos_dot1p_to_tc_map_id,
self.p3_qos_dot1p_to_color_map_id,
self.p3_qos_tc_color_to_dot1q_id]]:
# ingress set DOT1P -> TC
status = sai_thrift_set_port_attribute(
self.client, port, qos_dot1p_to_tc_map=dot1p_to_tc)
self.assertEqual(status, SAI_STATUS_SUCCESS,
"Failed to set port attribute")
attr = sai_thrift_get_port_attribute(self.client,
port,
qos_dot1p_to_tc_map=True)
self.assertEqual(attr['qos_dot1p_to_tc_map'], dot1p_to_tc)
# ingress set DOT1P -> COLOR
status = sai_thrift_set_port_attribute(
self.client, port, qos_dot1p_to_color_map=dot1p_color)
self.assertEqual(status, SAI_STATUS_SUCCESS,
"Failed to set port attribute")
attr = sai_thrift_get_port_attribute(
self.client, port, qos_dot1p_to_color_map=True)
self.assertEqual(attr['qos_dot1p_to_color_map'], dot1p_color)
# egress set TC and COLOR -> DOT1Q
status = sai_thrift_set_port_attribute(
self.client,
port,
qos_tc_and_color_to_dot1p_map=tc_color_to_pcp)
self.assertEqual(status, SAI_STATUS_SUCCESS,
"Failed to set port attribute")
attr = sai_thrift_get_port_attribute(
self.client, port, qos_tc_and_color_to_dot1p_map=True)
self.assertEqual(attr['qos_tc_and_color_to_dot1p_map'],
tc_color_to_pcp)
for test in test_cases:
ingress_port = test['ingress_port']
if test['pcp'] is not None:
vlan = 1
ingress_pcp = str(test['pcp'])
else:
vlan = 0
ingress_pcp = 'None'
# define expected egress_pcp per egress port
p0_egress_pcp = test['p0_pcp']
p1_egress_pcp = test['p1_pcp']
p2_egress_pcp = test['p2_pcp']
p3_egress_pcp = test['p3_pcp']
pkt = simple_tcp_packet(eth_dst='00:33:33:33:33:00',
eth_src='00:33:33:33:33:11',
dl_vlan_enable=True,
vlan_vid=vlan,
vlan_pcp=test['pcp'],
pktlen=104)
p0_exp_pkt = simple_tcp_packet(
eth_dst='00:33:33:33:33:00',
eth_src='00:33:33:33:33:11',
dl_vlan_enable=True,
vlan_vid=1,
vlan_pcp=p0_egress_pcp,
pktlen=104)
p1_exp_pkt = simple_tcp_packet(
eth_dst='00:33:33:33:33:00',
eth_src='00:33:33:33:33:11',
dl_vlan_enable=True,
vlan_vid=1,
vlan_pcp=p1_egress_pcp,
pktlen=104)
p2_exp_pkt = simple_tcp_packet(
eth_dst='00:33:33:33:33:00',
eth_src='00:33:33:33:33:11',
dl_vlan_enable=True,
vlan_vid=1,
vlan_pcp=p2_egress_pcp,
pktlen=104)
p3_exp_pkt = simple_tcp_packet(
eth_dst='00:33:33:33:33:00',
eth_src='00:33:33:33:33:11',
dl_vlan_enable=True,
vlan_vid=1,
vlan_pcp=p3_egress_pcp,
pktlen=104)
egres_ports = []
flood_port_list = []
flood_pkt_list = []
pkt_list = [p0_exp_pkt, p1_exp_pkt, p2_exp_pkt, p3_exp_pkt]
egress_pcp_list = [p0_egress_pcp, p1_egress_pcp,
p2_egress_pcp, p3_egress_pcp]
egress_pcp = []
for i, port in enumerate(self.dev_test_port_list):
if port == ingress_port:
continue
flood_port_list.append([port])
flood_pkt_list.append(pkt_list[i])
egres_ports.append(port)
egress_pcp.append(egress_pcp_list[i])
print("Sending eth packet port %d (pcp=%s) -> "
" port %d (pcp=%d)"
% (ingress_port, ingress_pcp, egres_ports[0],
egress_pcp[0]))
print("%47s port %d (egress_pcp=%d))" %
(" ", egres_ports[1], egress_pcp[1]))
print("%47s port %d (egress_pcp=%d))" %
(" ", egres_ports[2], egress_pcp[2]))
send_packet(self, ingress_port, pkt)
verify_each_packet_on_multiple_port_lists(self, flood_pkt_list,
flood_port_list)
print("\tPacket flooded. ok")