in ptf/saiqosmap.py [0:0]
def l2QosMapSamePCPToColorMultiIngresssPortTest(self):
''' L2 traffic test verifies the following:
1. Many PCP to one color mapping
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("l2QosMapSamePCPToColorMultiIngresssPortTest")
# default PCP to TC mapping
test_cases = []
# priority vlan tagged packed
test_cases.append({
'pcp': None,
'port_default_tc': 0,
'tc': 0,
'color': 0
})
test_cases.append({
'pcp': None,
'port_default_tc': 1,
'tc': 1,
'color': 0
})
test_cases.append({
'pcp': None,
'port_default_tc': 2,
'tc': 2,
'color': 0
})
test_cases.append({
'pcp': None,
'port_default_tc': 3,
'tc': 3,
'color': 0
})
test_cases.append({
'pcp': None,
'port_default_tc': 4,
'tc': 4,
'color': 0
})
test_cases.append({
'pcp': None,
'port_default_tc': 5,
'tc': 5,
'color': 0
})
test_cases.append({
'pcp': None,
'port_default_tc': 6,
'tc': 6,
'color': 0
})
test_cases.append({
'pcp': None,
'port_default_tc': 7,
'tc': 7,
'color': 0
})
test_cases.append({
'pcp': 0,
'port_default_tc': 5,
'tc': 0,
'color': 0
})
test_cases.append({
'pcp': 1,
'port_default_tc': 5,
'tc': 1,
'color': 0
})
test_cases.append({
'pcp': 2,
'port_default_tc': 0,
'tc': 2,
'color': 0
})
test_cases.append({
'pcp': 3,
'port_default_tc': 0,
'tc': 3,
'color': 1
})
test_cases.append({
'pcp': 4,
'port_default_tc': 0,
'tc': 4,
'color': 1
})
test_cases.append({
'pcp': 5,
'port_default_tc': 0,
'tc': 5,
'color': 1
})
test_cases.append({
'pcp': 6,
'port_default_tc': 0,
'tc': 6,
'color': 2
})
test_cases.append({
'pcp': 7,
'port_default_tc': 0,
'tc': 7,
'color': 2
})
for port, dot1p_to_tc, dot1p_color, tc_color in [
[self.port0, self.qos_dot1p_to_tc_map_id,
self.qos_dot1p_to_color_map_id,
self.p1_qos_tc_color_to_dot1q_id],
[self.port1, self.qos_dot1p_to_tc_map_id,
self.qos_dot1p_to_color_map_id,
self.p1_qos_tc_color_to_dot1q_id],
[self.port2, self.qos_dot1p_to_tc_map_id,
self.qos_dot1p_to_color_map_id,
self.p2_qos_tc_color_to_dot1q_id],
[self.port3, self.qos_dot1p_to_tc_map_id,
self.qos_dot1p_to_color_map_id,
self.p3_qos_tc_color_to_dot1q_id]]:
# 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)
# 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)
# set TC and COLOR -> DOT1Q
status = sai_thrift_set_port_attribute(
self.client, port, qos_tc_and_color_to_dot1p_map=tc_color)
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)
test_dev_ports_list = [self.dev_port0, self.dev_port1,
self.dev_port2, self.dev_port3]
for _, ingress_port in enumerate(test_dev_ports_list):
for test in test_cases:
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
egress_tc = test['tc']
egress_color = test['color']
p1_egress_pcp = 1 + test['color']
p2_egress_pcp = 3 + test['color']
p3_egress_pcp = 5 + test['color']
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)
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 = [p1_exp_pkt, p1_exp_pkt, p2_exp_pkt, p3_exp_pkt]
for i, port in enumerate(test_dev_ports_list):
if port == ingress_port:
continue
flood_port_list.append([port])
flood_pkt_list.append(pkt_list[i])
egres_ports.append(port)
print("Sending eth packet port %d (pcp=%s) -> "
"port %d (tc=%d color=%d -> egress_pcp=%d)"
% (ingress_port, ingress_pcp,
egres_ports[0], egress_tc, egress_color,
p1_egress_pcp))
print("%47s port %d (tc=%d color=%d -> egress_pcp=%d))"
% (" ", egres_ports[1], egress_tc, egress_color,
p2_egress_pcp))
print("%47s port %d (tc=%d color=%d -> egress_pcp=%d))"
% (" ", egres_ports[2], egress_tc, egress_color,
p3_egress_pcp))
send_packet(self, ingress_port, pkt)
verify_each_packet_on_multiple_port_lists(
self, flood_pkt_list, flood_port_list)
print("\tPacket flooded. ok")