in ptf/saiqueue.py [0:0]
def runTest(self):
print("CPU Port Queue Object Test")
attr = sai_thrift_get_switch_attribute(self.client, cpu_port=True)
cpu_port_id = attr["cpu_port"]
print("\tPART1: Port Queue Query Test")
global TEST_PASSED
try:
for port in self.port_list:
attr = sai_thrift_get_port_attribute(self.client,
port,
qos_number_of_queues=True)
num_queues = attr["qos_number_of_queues"]
q_list = sai_thrift_object_list_t(count=num_queues)
attr = sai_thrift_get_port_attribute(self.client,
port,
qos_queue_list=q_list)
for queue in range(0, num_queues):
queue_id = attr["qos_queue_list"].idlist[queue]
q_attr = sai_thrift_get_queue_attribute(
self.client,
queue_id,
port=True,
index=True,
parent_scheduler_node=True)
self.assertTrue(queue == q_attr["index"])
self.assertTrue(port == q_attr["port"])
# It should return port handle,
# because SDK does not support hierarchical QoS
# skip this test, platform dependent
#self.assertTrue(port == q_attr["parent_scheduler_node"])
print("\t\tPART1 completed successfully")
print("\tPART2: Buffer Queue Test")
# Get the queue for the CPU port - for the buffer queue test.
attr = sai_thrift_get_port_attribute(self.client,
cpu_port_id,
qos_number_of_queues=True)
num_queues = attr["qos_number_of_queues"]
q_list = sai_thrift_object_list_t(count=num_queues)
attr = sai_thrift_get_port_attribute(
self.client,
cpu_port_id,
qos_queue_list=q_list)
queue_id = attr["qos_queue_list"].idlist
mode = SAI_BUFFER_POOL_THRESHOLD_MODE_DYNAMIC
buff_pool = sai_thrift_create_buffer_pool(
self.client,
type=SAI_BUFFER_POOL_TYPE_EGRESS,
size=1024,
threshold_mode=mode)
self.assertTrue(buff_pool != 0)
buff_prof = sai_thrift_create_buffer_profile(
self.client,
pool_id=buff_pool,
reserved_buffer_size=1024,
threshold_mode=mode,
shared_dynamic_th=1,
xoff_th=100,
xon_th=10)
self.assertTrue(buff_prof != 0)
attr = sai_thrift_get_queue_attribute(self.client,
queue_id[0],
buffer_profile_id=True)
default_buff_prof = attr["buffer_profile_id"]
status = sai_thrift_set_queue_attribute(
self.client,
queue_id[0],
buffer_profile_id=buff_prof)
self.assertEqual(status, SAI_STATUS_SUCCESS)
# Now the buffer profile is being detached.
status = sai_thrift_set_queue_attribute(
self.client,
queue_id[0],
buffer_profile_id=int(SAI_NULL_OBJECT_ID))
self.assertEqual(status, SAI_STATUS_SUCCESS)
status = sai_thrift_remove_buffer_profile(self.client, buff_prof)
self.assertEqual(status, SAI_STATUS_SUCCESS)
status = sai_thrift_remove_buffer_pool(self.client, buff_pool)
self.assertEqual(status, SAI_STATUS_SUCCESS)
# The queue should be assigned to the default buffer profile.
attr = sai_thrift_get_queue_attribute(self.client,
queue_id[0],
buffer_profile_id=True)
self.assertEqual(attr["buffer_profile_id"], default_buff_prof)
print("\t\tPART2 completed successfully")
print("\tPART3: Scheduler Queue Test")
weight = 4
min_rate = 100
max_rate = 1000
min_burst = 200
max_burst = 800
# Get the queue for the CPU port - for the scheduler queue test.
attr = sai_thrift_get_port_attribute(
self.client,
cpu_port_id,
qos_queue_list=q_list)
queue_id = attr["qos_queue_list"].idlist
sched1 = sai_thrift_create_scheduler(
self.client,
meter_type=SAI_METER_TYPE_PACKETS,
scheduling_type=SAI_SCHEDULING_TYPE_DWRR,
scheduling_weight=weight)
self.assertTrue(sched1 != 0)
sched2 = sai_thrift_create_scheduler(
self.client,
meter_type=SAI_METER_TYPE_PACKETS,
scheduling_type=SAI_SCHEDULING_TYPE_STRICT,
min_bandwidth_rate=min_rate,
max_bandwidth_rate=max_rate)
self.assertTrue(sched2 != 0)
sched3 = sai_thrift_create_scheduler(
self.client,
meter_type=SAI_METER_TYPE_PACKETS,
scheduling_type=SAI_SCHEDULING_TYPE_STRICT,
min_bandwidth_burst_rate=min_burst,
max_bandwidth_burst_rate=max_burst)
self.assertTrue(sched3 != 0)
print("\t\tScheduler profiles created successfully")
status = sai_thrift_set_queue_attribute(
self.client,
queue_id[1],
scheduler_profile_id=sched1)
self.assertEqual(status, SAI_STATUS_SUCCESS)
q_attr = sai_thrift_get_queue_attribute(
self.client,
queue_id[1],
scheduler_profile_id=True)
self.assertEqual(q_attr["scheduler_profile_id"], sched1)
s_attr = sai_thrift_get_scheduler_attribute(
self.client,
sched1,
scheduling_weight=True)
self.assertEqual(s_attr["scheduling_weight"], weight)
print("\t\tScheduling weight validated successfully")
status = sai_thrift_set_queue_attribute(
self.client,
queue_id[2],
scheduler_profile_id=sched2)
self.assertEqual(status, SAI_STATUS_SUCCESS)
q_attr = sai_thrift_get_queue_attribute(
self.client,
queue_id[2],
scheduler_profile_id=True)
self.assertEqual(q_attr["scheduler_profile_id"], sched2)
s_attr = sai_thrift_get_scheduler_attribute(
self.client,
sched2,
min_bandwidth_rate=True,
max_bandwidth_rate=True)
self.assertEqual(s_attr["min_bandwidth_rate"], min_rate)
self.assertEqual(s_attr["max_bandwidth_rate"], max_rate)
print("\t\tBandwidth rates validated successfully")
status = sai_thrift_set_queue_attribute(
self.client,
queue_id[3],
scheduler_profile_id=sched3)
self.assertEqual(status, SAI_STATUS_SUCCESS)
q_attr = sai_thrift_get_queue_attribute(
self.client,
queue_id[3],
scheduler_profile_id=True)
self.assertEqual(q_attr["scheduler_profile_id"], sched3)
s_attr = sai_thrift_get_scheduler_attribute(
self.client,
sched3,
min_bandwidth_burst_rate=True,
max_bandwidth_burst_rate=True)
self.assertEqual(s_attr["min_bandwidth_burst_rate"], min_burst)
self.assertEqual(s_attr["max_bandwidth_burst_rate"], max_burst)
print("\t\tBandwidth burst rates validated successfully")
print("\t\tNow the parameters will be modified")
weight = 3
min_rate = 120
max_rate = 980
min_burst = 300
max_burst = 800
status = sai_thrift_set_scheduler_attribute(
self.client,
sched1,
scheduling_weight=weight)
self.assertEqual(status, SAI_STATUS_SUCCESS)
s_attr = sai_thrift_get_scheduler_attribute(
self.client,
sched1,
scheduling_weight=True)
self.assertEqual(s_attr["scheduling_weight"], weight)
print("\t\tScheduling weight validated successfully")
status = sai_thrift_set_scheduler_attribute(
self.client,
sched2,
min_bandwidth_rate=min_rate)
self.assertEqual(status, SAI_STATUS_SUCCESS)
status = sai_thrift_set_scheduler_attribute(
self.client,
sched2,
max_bandwidth_rate=max_rate)
self.assertEqual(status, SAI_STATUS_SUCCESS)
s_attr = sai_thrift_get_scheduler_attribute(
self.client,
sched2,
min_bandwidth_rate=True,
max_bandwidth_rate=True)
self.assertEqual(s_attr["min_bandwidth_rate"], min_rate)
self.assertEqual(s_attr["max_bandwidth_rate"], max_rate)
print("\t\tBandwidth rates validated successfully")
status = sai_thrift_set_scheduler_attribute(
self.client,
sched3,
min_bandwidth_burst_rate=min_burst)
self.assertEqual(status, SAI_STATUS_SUCCESS)
status = sai_thrift_set_scheduler_attribute(
self.client,
sched3,
max_bandwidth_burst_rate=max_burst)
self.assertEqual(status, SAI_STATUS_SUCCESS)
s_attr = sai_thrift_get_scheduler_attribute(
self.client,
sched3,
min_bandwidth_burst_rate=True,
max_bandwidth_burst_rate=True)
self.assertEqual(s_attr["min_bandwidth_burst_rate"], min_burst)
self.assertEqual(s_attr["max_bandwidth_burst_rate"], max_burst)
print("\t\tBandwidth burst rates validated successfully")
print("\t\tPART3 completed successfully")
print("\tPART4: PFC Priority Queue Test")
# Get the queue for the CPU port.
p_attr = sai_thrift_get_port_attribute(
self.client,
cpu_port_id,
qos_queue_list=q_list)
queue_id = p_attr["qos_queue_list"].idlist
q_attr = sai_thrift_get_queue_attribute(
self.client,
queue_id[4],
index=True)
q_idx = q_attr["index"]
# The PFC Priority -> Queue map table configuration.
pfc_to_queue_map = sai_thrift_qos_map_t(
key=sai_thrift_qos_map_params_t(prio=4),
value=sai_thrift_qos_map_params_t(queue_index=q_idx))
qos_map_list = sai_thrift_qos_map_list_t(
count=1,
maplist=[pfc_to_queue_map])
qos_map = sai_thrift_create_qos_map(
self.client,
type=SAI_QOS_MAP_TYPE_PFC_PRIORITY_TO_QUEUE,
map_to_value_list=qos_map_list)
self.assertTrue(qos_map != 0)
# The queue index modification in the previously created QoS map.
q_attr = sai_thrift_get_queue_attribute(
self.client,
queue_id[3],
index=True)
new_q_idx = q_attr["index"]
new_pfc_to_queue_map = sai_thrift_qos_map_t(
key=sai_thrift_qos_map_params_t(prio=4),
value=sai_thrift_qos_map_params_t(queue_index=new_q_idx))
new_qos_map_list = sai_thrift_qos_map_list_t(
count=1,
maplist=[new_pfc_to_queue_map])
status = sai_thrift_set_qos_map_attribute(
self.client,
qos_map,
map_to_value_list=new_qos_map_list)
self.assertEqual(status, SAI_STATUS_SUCCESS)
# The verification if correct queue is applied to QoS map
map_lists = sai_thrift_qos_map_list_t(count=1)
attr = sai_thrift_get_qos_map_attribute(
self.client,
qos_map,
map_to_value_list=map_lists)
queue_idx = attr["map_to_value_list"].maplist[0].value.queue_index
self.assertEqual(queue_idx, new_q_idx)
print("\t\tPART4 completed successfully")
print("\tTest completed successfully")
TEST_PASSED = True
finally:
if not TEST_PASSED:
print("\tTest FAILED, skip tear down for un-expected operation.")
else:
sai_thrift_remove_qos_map(self.client, qos_map)
status = sai_thrift_set_queue_attribute(
self.client,
queue_id[3],
scheduler_profile_id=0)
self.assertEqual(status, SAI_STATUS_SUCCESS)
status = sai_thrift_set_queue_attribute(
self.client,
queue_id[2],
scheduler_profile_id=0)
self.assertEqual(status, SAI_STATUS_SUCCESS)
status = sai_thrift_set_queue_attribute(
self.client,
queue_id[1],
scheduler_profile_id=0)
self.assertEqual(status, SAI_STATUS_SUCCESS)
sai_thrift_remove_scheduler(self.client, sched3)
sai_thrift_remove_scheduler(self.client, sched2)
sai_thrift_remove_scheduler(self.client, sched1)