in ptf/sailag.py [0:0]
def lagMemberActivateFloodTest(self):
'''
The LAG mamber activate Flood test
'''
print("lagMemberActivateFloodTest()")
try:
l1 = [self.dev_port4, self.dev_port5, self.dev_port6]
max_itrs = 4
def packet_test():
'''
Packet test function that requests max_itrs packets
and verifies the number of packets received per LAG member
Returns:
list: list of packets counts per LAG member
'''
pkt = simple_arp_packet(
arp_op=1,
pktlen=100,
eth_src='00:11:11:11:11:11',
hw_snd='00:11:11:11:11:11',
ip_snd='10.10.10.1',
ip_tgt='10.10.10.2')
tagged_pkt = simple_arp_packet(
arp_op=1,
vlan_vid=10,
pktlen=104,
eth_src='00:11:11:11:11:11',
hw_snd='00:11:11:11:11:11',
ip_snd='10.10.10.1',
ip_tgt='10.10.10.2')
count = [0, 0, 0, 0]
for _ in range(0, max_itrs):
send_packet(self, self.dev_port0, pkt)
rcv_idx = verify_each_packet_on_multiple_port_lists(
self,
[pkt, tagged_pkt],
[l1, [self.dev_port1]])
for v in rcv_idx[0]:
count[v] += 1
return count
mbrs = {
0: self.lag1_member4,
1: self.lag1_member5,
2: self.lag1_member6}
disabled_mbrs = set()
# for each iteration
# - check if packets are recieved on only LAG member
# - check if packets are not recieved on disabled LAG member
count = packet_test()
self.assertEqual(len([i for i in count if i != 0]), 1)
print('Test with 3 mbrs enabled:', count)
index = [i for i in range(len(count)) if count[i] != 0]
status = sai_thrift_set_lag_member_attribute(
self.client,
mbrs[index[0]],
egress_disable=True)
self.assertEqual(status, SAI_STATUS_SUCCESS)
print('Disabled LAG member %d' % (index[0]))
disabled_mbrs.add(index[0])
count = packet_test()
self.assertEqual(len([i for i in count if i != 0]), 1)
print('Test with 2 mbrs enabled:', count)
index = [i for i in range(len(count)) if count[i] != 0]
self.assertNotIn(index[0], disabled_mbrs)
status = sai_thrift_set_lag_member_attribute(
self.client,
mbrs[index[0]],
egress_disable=True)
self.assertEqual(status, SAI_STATUS_SUCCESS)
print('Disabled LAG member %d' % (index[0]))
disabled_mbrs.add(index[0])
count = packet_test()
self.assertEqual(len([i for i in count if i != 0]), 1)
print('Test with 1 mbrs enabled:', count)
index = [i for i in range(len(count)) if count[i] != 0]
self.assertNotIn(index[0], disabled_mbrs)
status = sai_thrift_set_lag_member_attribute(
self.client,
mbrs[index[0]],
egress_disable=True)
self.assertEqual(status, SAI_STATUS_SUCCESS)
finally:
for mbr in mbrs.values():
sai_thrift_set_lag_member_attribute(
self.client,
mbr,
egress_disable=False)