ptf/saifdb.py (3,145 lines of code) (raw):
# Copyright 2021-present Intel Corporation.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""
Thrift SAI interface L2 FDB tests
"""
from sai_thrift.sai_headers import *
from sai_base_test import *
@group("draft")
class FdbAttributeTest(SaiHelper):
'''
Basic FDB attributes getting and setting test
'''
def setUp(self):
super(FdbAttributeTest, self).setUp()
self.fdb_entry = sai_thrift_fdb_entry_t(
switch_id=self.switch_id,
mac_address="00:11:22:33:44:55",
bv_id=self.vlan10)
sai_thrift_create_fdb_entry(self.client,
self.fdb_entry,
type=SAI_FDB_ENTRY_TYPE_STATIC,
bridge_port_id=self.port0_bp)
self.assertEqual(self.status(), SAI_STATUS_SUCCESS)
def runTest(self):
# get bridge_port_id
attr = sai_thrift_get_fdb_entry_attribute(self.client,
self.fdb_entry,
bridge_port_id=True)
self.assertEqual(attr['bridge_port_id'], self.port0_bp)
# get type
attr = sai_thrift_get_fdb_entry_attribute(
self.client, self.fdb_entry, type=True)
self.assertEqual(attr['type'], SAI_FDB_ENTRY_TYPE_STATIC)
# set packet_action
sai_thrift_set_fdb_entry_attribute(
self.client,
self.fdb_entry,
packet_action=SAI_PACKET_ACTION_FORWARD)
self.assertEqual(self.status(), SAI_STATUS_SUCCESS)
# get packet_action
attr = sai_thrift_get_fdb_entry_attribute(
self.client, self.fdb_entry, packet_action=True)
self.assertEqual(attr['packet_action'], SAI_PACKET_ACTION_FORWARD)
def tearDown(self):
sai_thrift_remove_fdb_entry(self.client, self.fdb_entry)
super(FdbAttributeTest, self).tearDown()
@group("draft")
class FdbNoLearnTest(SaiHelper):
'''
Verify different cases when MAC addresses should not be learned
'''
def setUp(self):
super(FdbNoLearnTest, self).setUp()
vlan_id = 10
self.src_mac = "00:11:11:11:11:11"
self.dst_mac = "00:22:22:22:22:22"
self.pkt = simple_udp_packet(eth_dst=self.dst_mac,
eth_src=self.src_mac,
pktlen=100)
self.tag_pkt = simple_udp_packet(eth_dst=self.dst_mac,
eth_src=self.src_mac,
dl_vlan_enable=True,
vlan_vid=vlan_id,
pktlen=104)
self.chck_pkt = simple_udp_packet(eth_dst=self.src_mac,
eth_src=self.dst_mac,
pktlen=100)
self.tag_chck_pkt = simple_udp_packet(eth_dst=self.src_mac,
eth_src=self.dst_mac,
dl_vlan_enable=True,
vlan_vid=vlan_id,
pktlen=104)
def runTest(self):
self.vlanPortNoLearnTest()
self.vlanLagNoLearnTest()
self.bpPortNoLearnTest()
self.bpLagNoLearnTest()
self.noBpNoLearnTest()
self.removedBpNoLearnTest()
def vlanPortNoLearnTest(self):
'''
Verify if MAC addresses are not learned on port when VLAN learning
is disabled
Send a packet on port0 with SMAC 00:11:11:11:11:11 and we verify it
was broadcasted on other ports in VLAN.
Next send a packet on port1 with DMAC 00:11:11:11:11:11 and verify
it was also broadcasted on other ports in VLAN
'''
print("\nvlanPortNoLearnTest()")
try:
# disable VLAN10 learning
status = sai_thrift_set_vlan_attribute(
self.client, self.vlan10, learn_disable=True)
self.assertEqual(status, SAI_STATUS_SUCCESS)
print("MAC Learning disabled on VLAN")
print("Sending packet on port %d, %s -> %s - will flood" %
(self.dev_port0, self.src_mac, self.dst_mac))
send_packet(self, self.dev_port0, self.pkt)
verify_each_packet_on_multiple_port_lists(
self, [self.tag_pkt, self.pkt],
[[self.dev_port1],
[self.dev_port4, self.dev_port5, self.dev_port6]])
print("Sending packet on port %d with DMAC %s - MAC not learned, "
"will flood" % (self.dev_port1, self.src_mac))
send_packet(self, self.dev_port1, self.tag_chck_pkt)
verify_each_packet_on_multiple_port_lists(
self, [self.chck_pkt, self.chck_pkt],
[[self.dev_port0],
[self.dev_port4, self.dev_port5, self.dev_port6]])
print("Verification complete")
finally:
sai_thrift_flush_fdb_entries(
self.client, entry_type=SAI_FDB_FLUSH_ENTRY_TYPE_ALL)
# restore initial VLAN Learning state
sai_thrift_set_vlan_attribute(
self.client, self.vlan10, learn_disable=False)
def vlanLagNoLearnTest(self):
'''
Verify if MAC addresses are not learned on LAG when VLAN learning
is disabled
Send a packet on LAG1 with SMAC 00:11:11:11:11:11, and verify it was
broadcasted on other ports in VLAN.
Next send a packet on port1 with DMAC 00:11:11:11:11:11 and verify
it was also broadcasted on other ports in VLAN
'''
print("\nvlanLagNoLearnTest()")
try:
# disable VLAN10 learning
status = sai_thrift_set_vlan_attribute(
self.client, self.vlan10, learn_disable=True)
self.assertEqual(status, SAI_STATUS_SUCCESS)
print("MAC Learning disabled on VLAN")
print("Sending packet on LAG port %d, %s -> %s - will flood" %
(self.dev_port5, self.src_mac, self.dst_mac))
send_packet(self, self.dev_port5, self.pkt)
verify_each_packet_on_each_port(self, [self.pkt, self.tag_pkt],
[self.dev_port0, self.dev_port1])
print("Sending packet on port %d with DMAC %s - MAC not learned, "
"will flood" % (self.dev_port1, self.src_mac))
send_packet(self, self.dev_port1, self.tag_chck_pkt)
verify_each_packet_on_multiple_port_lists(
self, [self.chck_pkt, self.chck_pkt],
[[self.dev_port0],
[self.dev_port4, self.dev_port5, self.dev_port6]])
print("Verification complete")
finally:
sai_thrift_flush_fdb_entries(
self.client, entry_type=SAI_FDB_FLUSH_ENTRY_TYPE_ALL)
# restore initial VLAN Learning state
sai_thrift_set_vlan_attribute(
self.client, self.vlan10, learn_disable=False)
def bpPortNoLearnTest(self):
'''
Verify if MAC addresses are not learned on port when bridge port
learning is disabled.
Disable learning on port0, send a packet on port0 with SMAC
00:11:11:11:11:11 and verify it was broadcasted on other ports in VLAN.
Next send a packet on port1 with DMAC 00:11:11:11:11:11 and verify
it was also broadcasted on other ports in VLAN
'''
print("\nbpPortNoLearnTest()")
try:
# disable learning on port0
status = sai_thrift_set_bridge_port_attribute(
self.client,
self.port0_bp,
fdb_learning_mode=SAI_BRIDGE_PORT_FDB_LEARNING_MODE_DISABLE)
self.assertEqual(status, SAI_STATUS_SUCCESS)
print("MAC Learning disabled on port %d" % self.dev_port0)
print("Sending packet on port %d, %s -> %s - will flood" %
(self.dev_port0, self.src_mac, self.dst_mac))
send_packet(self, self.dev_port0, self.pkt)
verify_each_packet_on_multiple_port_lists(
self, [self.tag_pkt, self.pkt],
[[self.dev_port1],
[self.dev_port4, self.dev_port5, self.dev_port6]])
print("Sending packet on port %d with DMAC %s - MAC not learned, "
"will flood" % (self.dev_port1, self.src_mac))
send_packet(self, self.dev_port1, self.tag_chck_pkt)
verify_each_packet_on_multiple_port_lists(
self, [self.chck_pkt, self.chck_pkt],
[[self.dev_port0],
[self.dev_port4, self.dev_port5, self.dev_port6]])
print("Verification complete")
finally:
sai_thrift_flush_fdb_entries(
self.client, entry_type=SAI_FDB_FLUSH_ENTRY_TYPE_ALL)
# restore initial VLAN Learning state
sai_thrift_set_bridge_port_attribute(
self.client, self.port0_bp,
fdb_learning_mode=SAI_BRIDGE_PORT_FDB_LEARNING_MODE_HW)
def bpLagNoLearnTest(self):
'''
Verify if MAC addresses are not learned on port when LAG bridge
port learning is disabled.
Disable learning on LAG1, send a packet on port0 with
SMAC: 00:11:11:11:11:11, and then we verify it was broadcasted on other
ports in VLAN.
Next send a packet on port1 with DMAC 00:11:11:11:11:11 and verify
it was also broadcasted on other ports in VLAN
'''
print("\nbpLagNoLearnTest()")
try:
# disable learning on LAG1
status = sai_thrift_set_bridge_port_attribute(
self.client,
self.lag1_bp,
fdb_learning_mode=SAI_BRIDGE_PORT_FDB_LEARNING_MODE_DISABLE)
self.assertEqual(status, SAI_STATUS_SUCCESS)
print("MAC Learning disabled on LAG1")
print("Sending packet on LAG port %d, %s -> %s - will flood" %
(self.dev_port5, self.src_mac, self.dst_mac))
send_packet(self, self.dev_port5, self.pkt)
verify_each_packet_on_each_port(self, [self.pkt, self.tag_pkt],
[self.dev_port0, self.dev_port1])
print("Sending packet on port %d with DMAC %s - MAC not learned, "
"will flood" % (self.dev_port1, self.src_mac))
send_packet(self, self.dev_port1, self.tag_chck_pkt)
verify_each_packet_on_multiple_port_lists(
self, [self.chck_pkt, self.chck_pkt],
[[self.dev_port0],
[self.dev_port4, self.dev_port5, self.dev_port6]])
print("Verification complete")
finally:
sai_thrift_flush_fdb_entries(
self.client, entry_type=SAI_FDB_FLUSH_ENTRY_TYPE_ALL)
# restore initial VLAN Learning state
sai_thrift_set_bridge_port_attribute(
self.client, self.lag1_bp,
fdb_learning_mode=SAI_BRIDGE_PORT_FDB_LEARNING_MODE_HW)
def noBpNoLearnTest(self):
'''
Verify if MAC is not learned on port if bridge port is not created
on that port
'''
print("\nnoBpNoLearnTest()")
test_port = self.port24
test_dev_port = self.dev_port24
vlan10_ports = [[self.dev_port0], [self.dev_port1],
[self.dev_port4, self.dev_port5, self.dev_port6]]
chck_port = vlan10_ports[0][0]
flood_pkt_list = [self.pkt, self.tag_pkt, self.pkt]
flood_port_list = [[self.dev_port1],
[self.dev_port4, self.dev_port5, self.dev_port6]]
flood_chk_pkt_list = [self.tag_chck_pkt, self.chck_pkt]
vlan_id = 10
try:
# set VLAN on port without bridge port creation
sai_thrift_set_port_attribute(
self.client, test_port, port_vlan_id=vlan_id)
print("Sending packet on port %d - with no bridge port created" %
test_dev_port)
send_packet(self, test_dev_port, self.pkt)
verify_each_packet_on_multiple_port_lists(
self, flood_pkt_list, vlan10_ports)
print("\tOK")
print("Sending packet to check MAC address was not learned on "
"port %d" % test_dev_port)
send_packet(self, chck_port, self.chck_pkt)
verify_each_packet_on_multiple_port_lists(
self, flood_chk_pkt_list, flood_port_list)
print("\tOK")
finally:
sai_thrift_flush_fdb_entries(
self.client, entry_type=SAI_FDB_FLUSH_ENTRY_TYPE_ALL)
sai_thrift_set_port_attribute(
self.client, test_port, port_vlan_id=0)
def removedBpNoLearnTest(self):
'''
Verify if MAC address is not learned on port after bridge port
is removed on that port
'''
print("\nremovedBpNoLearnTest()")
try:
sai_thrift_remove_bridge_port(self.client, self.port0_bp)
print("Sending packet on port %d - with removed bridge port" %
self.dev_port0)
print("Packet will be flooded on vlan 10")
send_packet(self, self.dev_port0, self.pkt)
verify_each_packet_on_multiple_port_lists(
self, [self.tag_pkt, self.pkt],
[[self.dev_port1],
[self.dev_port4, self.dev_port5, self.dev_port6]])
print("\tOK")
print("Sending packet to check MAC address was not learned on "
"port %d" % self.dev_port0)
send_packet(self, self.dev_port1, self.tag_chck_pkt)
verify_each_packet_on_multiple_port_lists(
self, [self.chck_pkt, self.chck_pkt],
[[self.dev_port0],
[self.dev_port4, self.dev_port5, self.dev_port6]])
print("\tOK")
finally:
sai_thrift_flush_fdb_entries(
self.client, entry_type=SAI_FDB_FLUSH_ENTRY_TYPE_ALL)
self.port0_bp = sai_thrift_create_bridge_port(
self.client,
bridge_id=self.default_1q_bridge,
port_id=self.port0,
type=SAI_BRIDGE_PORT_TYPE_PORT,
admin_state=True)
self.assertNotEqual(self.port0_bp, 0)
@group("draft")
class FdbLearnTest(SaiHelper):
'''
Verify different cases of FDB learning
'''
def setUp(self):
super(FdbLearnTest, self).setUp()
self.vlan_id = 10
# add (tagged) LAG2 to VLAN 10
self.vlan10_member3 = sai_thrift_create_vlan_member(
self.client,
vlan_id=self.vlan10,
bridge_port_id=self.lag2_bp,
vlan_tagging_mode=SAI_VLAN_TAGGING_MODE_TAGGED)
self.src_ports = [self.dev_port0, self.dev_port1, self.dev_port4,
self.dev_port5, self.dev_port6, self.dev_port7,
self.dev_port8, self.dev_port9]
self.macs = []
for i in range(1, len(self.src_ports)):
self.macs.append("00:%02d:%02d:%02d:%02d:%02d" % (i, i, i, i, i))
self.utg_lag_ports = [self.dev_port4, self.dev_port5, self.dev_port6]
self.tg_lag_ports = [self.dev_port7, self.dev_port8, self.dev_port9]
self.dst_ports = [[self.dev_port0], [self.dev_port1],
self.utg_lag_ports, self.tg_lag_ports]
def runTest(self):
self.dynamicMacLearnTest()
self.macLearnErrorTest()
def tearDown(self):
# remove LAG2 from VLAN 10
sai_thrift_remove_vlan_member(self.client, self.vlan10_member3)
sai_thrift_flush_fdb_entries(
self.client, entry_type=SAI_FDB_FLUSH_ENTRY_TYPE_ALL)
super(FdbLearnTest, self).tearDown()
def dynamicMacLearnTest(self): # noqa pylint: disable=too-many-branches
'''
Verify if MAC addresses are learned on tagged and untagged VLAN
ports and LAGs. The test checks also if MAC addresses are learned on
newly added VLAN and LAG member.
Send packets with given SMACs and verify they are broadcasted
on other ports.
Next send packets with DMACs == previous SMACs and verify they are
forwarded only to the proper ports
'''
print("\ndynamicMacLearnTest()")
try:
# learning phase
dst_mac = "00:11:22:33:44:55"
for src_port, src_mac in zip(self.src_ports, self.macs):
pkt = simple_udp_packet(eth_dst=dst_mac,
eth_src=src_mac,
pktlen=100)
tag_pkt = simple_udp_packet(eth_dst=dst_mac,
eth_src=src_mac,
dl_vlan_enable=True,
vlan_vid=self.vlan_id,
pktlen=104)
if src_port == self.dev_port1 or src_port in self.tg_lag_ports:
send_pkt = tag_pkt
else:
send_pkt = pkt
# create port list for flooding verification
flood_port_list = [
self.dst_ports[p] for p in range(len(self.dst_ports))
if src_port not in self.dst_ports[p]]
# create packet list for flooding verification
flood_pkt_list = []
tg_lag_set = False
utg_lag_set = False
for dst_port in self.dst_ports:
if src_port in dst_port:
continue
if dst_port == [self.dev_port0]:
flood_pkt_list.append(pkt)
elif dst_port == [self.dev_port1]:
flood_pkt_list.append(tag_pkt)
elif dst_port == self.utg_lag_ports and not utg_lag_set:
flood_pkt_list.append(pkt)
utg_lag_set = True
elif dst_port == self.tg_lag_ports and not tg_lag_set:
flood_pkt_list.append(tag_pkt)
tg_lag_set = True
print("Sending packet on port %d, %s -> %s - will flood" %
(src_port, src_mac, dst_mac))
send_packet(self, src_port, send_pkt)
verify_each_packet_on_multiple_port_lists(self, flood_pkt_list,
flood_port_list)
print("\tLearning complete\n")
# verification phase:
for dst_port, dst_mac in zip(self.src_ports, self.macs):
for src_port, src_mac in zip([self.dev_port0, self.dev_port1],
[self.macs[0], self.macs[1]]):
if src_port == dst_port:
continue
pkt = simple_udp_packet(eth_dst=dst_mac,
eth_src=src_mac,
pktlen=100)
tag_pkt = simple_udp_packet(eth_dst=dst_mac,
eth_src=src_mac,
dl_vlan_enable=True,
vlan_vid=self.vlan_id,
pktlen=104)
if src_port == self.dev_port1 or src_port \
in self.tg_lag_ports:
send_pkt = tag_pkt
else:
send_pkt = pkt
if dst_port == self.dev_port1 or dst_port \
in self.tg_lag_ports:
rcv_pkt = tag_pkt
else:
rcv_pkt = pkt
if dst_port in self.utg_lag_ports:
rcv_port = self.utg_lag_ports
elif dst_port in self.tg_lag_ports:
rcv_port = self.tg_lag_ports
else:
rcv_port = [dst_port]
print("Sending packet on port %d, %s -> %s - forwarding" %
(src_port, src_mac, dst_mac))
send_packet(self, src_port, send_pkt)
verify_packet_any_port(self, rcv_pkt, rcv_port)
print("\tVerification complete")
# verify learning on new VLAN member
new_vlan_member = self.port24
new_vlan_member_dev = self.dev_port24
new_vlan_member_mac = "00:12:34:56:78:90"
new_vlan_member_bp = sai_thrift_create_bridge_port(
self.client,
bridge_id=self.default_1q_bridge,
port_id=new_vlan_member,
type=SAI_BRIDGE_PORT_TYPE_PORT,
admin_state=True)
vlan10_member4 = sai_thrift_create_vlan_member(
self.client,
vlan_id=self.vlan10,
bridge_port_id=new_vlan_member_bp,
vlan_tagging_mode=SAI_VLAN_TAGGING_MODE_UNTAGGED)
sai_thrift_set_port_attribute(self.client,
new_vlan_member,
port_vlan_id=self.vlan_id)
print("\nVerify learning on new VLAN member (port %d)" %
new_vlan_member_dev)
# check if packet is flooded to the new VLAN port
pkt = simple_udp_packet(eth_dst=new_vlan_member_mac,
eth_src=self.macs[0],
pktlen=100)
tag_pkt = simple_udp_packet(eth_dst=new_vlan_member_mac,
eth_src=self.macs[0],
dl_vlan_enable=True,
vlan_vid=self.vlan_id,
pktlen=104)
flood_port_list = [[self.dev_port1], self.utg_lag_ports,
self.tg_lag_ports, [new_vlan_member_dev]]
flood_pkt_list = [tag_pkt, pkt, tag_pkt, pkt]
print("Sending packet on port %d, %s -> %s\nchecking if flood "
"reaches port %d" % (self.dev_port0, self.macs[0],
new_vlan_member_mac,
new_vlan_member_dev))
send_packet(self, self.dev_port0, pkt)
verify_each_packet_on_multiple_port_lists(self, flood_pkt_list,
flood_port_list)
print("\tOK")
# check if MAC is learned on new VLAN member
dst_mac = "00:11:22:33:44:55"
pkt = simple_udp_packet(eth_dst=dst_mac,
eth_src=new_vlan_member_mac,
pktlen=100)
tag_pkt = simple_udp_packet(eth_dst=dst_mac,
eth_src=new_vlan_member_mac,
dl_vlan_enable=True,
vlan_vid=self.vlan_id,
pktlen=104)
flood_port_list = self.dst_ports
flood_pkt_list = [pkt, tag_pkt, pkt, tag_pkt]
print("Sending packet on new VLAN port (%d), %s -> %s - will "
"flood" % (new_vlan_member_dev, new_vlan_member_mac,
dst_mac))
send_packet(self, new_vlan_member_dev, pkt)
verify_each_packet_on_multiple_port_lists(self, flood_pkt_list,
flood_port_list)
print("\tOK")
time.sleep(2)
print("Checking if MAC was learned")
pkt = simple_udp_packet(eth_dst=new_vlan_member_mac,
eth_src=self.macs[0],
pktlen=100)
print("Sending packet on port %d, %s -> %s - forward to port %d" %
(self.dev_port0, self.macs[0], new_vlan_member_mac,
new_vlan_member_dev))
send_packet(self, self.dev_port0, pkt)
verify_packets(self, pkt, [new_vlan_member_dev])
print("\tVerification complete")
# verify learning on new LAG member
new_lag_member = self.port25
new_lag_member_dev = self.dev_port25
new_lag_member_mac = "00:09:87:65:43:21"
lag1_member25 = sai_thrift_create_lag_member(
self.client, lag_id=self.lag1, port_id=new_lag_member)
self.utg_lag_ports.append(new_lag_member_dev)
print("\nVerify learning on new LAG member (port %d in LAG1)" %
new_lag_member_dev)
pkt = simple_udp_packet(eth_dst=dst_mac,
eth_src=new_lag_member_mac,
pktlen=100)
tag_pkt = simple_udp_packet(eth_dst=dst_mac,
eth_src=new_lag_member_mac,
dl_vlan_enable=True,
vlan_vid=self.vlan_id,
pktlen=104)
flood_port_list = [[self.dev_port0], [self.dev_port1],
self.tg_lag_ports, [new_vlan_member_dev]]
flood_pkt_list = [pkt, tag_pkt, tag_pkt, pkt]
print("Sending packet on new LAG member (port %d), %s -> %s - "
"will flood" % (new_lag_member_dev, new_lag_member_mac,
dst_mac))
send_packet(self, new_lag_member_dev, pkt)
verify_each_packet_on_multiple_port_lists(self, flood_pkt_list,
flood_port_list)
print("\tOK")
time.sleep(2)
print("Checking if MAC was learned")
pkt = simple_udp_packet(eth_dst=new_lag_member_mac,
eth_src=self.macs[0],
pktlen=100)
print("Sending packet on port %d, %s -> %s forward to LAG1" %
(self.dev_port0, self.macs[0], new_lag_member_mac))
send_packet(self, self.dev_port0, pkt)
verify_packet_any_port(self, pkt, self.utg_lag_ports)
print("\tVerification complete")
finally:
sai_thrift_flush_fdb_entries(
self.client, entry_type=SAI_FDB_FLUSH_ENTRY_TYPE_ALL)
# remove new LAG member
sai_thrift_remove_lag_member(self.client, lag1_member25)
self.utg_lag_ports.remove(new_lag_member_dev)
# remove new vlan member
sai_thrift_set_port_attribute(self.client, new_vlan_member,
port_vlan_id=0)
sai_thrift_remove_vlan_member(self.client, vlan10_member4)
sai_thrift_remove_bridge_port(self.client, new_vlan_member_bp)
def macLearnErrorTest(self):
'''
Verify if MAC addresses are not learned when different undesirable
conditions occured:
1 - invalid VLAN tag
2 - src_mac is a broadcast address (packet drop)
3 - src_mac is a multicast address (packet drop)
4 - src_mac previously added statically
5 - removed VLAN member
6 - removed LAG member
'''
print("\nmacLearnErrorTest()")
access_port = self.dev_port0 # untagged
ap_mac = self.macs[0]
trunk_port = self.dev_port1 # tagged
tp_mac = self.macs[1]
flood_port_list = [[trunk_port], self.utg_lag_ports, self.tg_lag_ports]
try:
fdb_entry1 = sai_thrift_fdb_entry_t(switch_id=self.switch_id,
mac_address=ap_mac,
bv_id=self.vlan10)
status = sai_thrift_create_fdb_entry(
self.client,
fdb_entry1,
type=SAI_FDB_ENTRY_TYPE_STATIC,
bridge_port_id=self.port0_bp)
self.assertEqual(status, SAI_STATUS_SUCCESS)
fdb_entry2 = sai_thrift_fdb_entry_t(switch_id=self.switch_id,
mac_address=tp_mac,
bv_id=self.vlan10)
status = sai_thrift_create_fdb_entry(
self.client,
fdb_entry2,
type=SAI_FDB_ENTRY_TYPE_STATIC,
bridge_port_id=self.port1_bp)
self.assertEqual(status, SAI_STATUS_SUCCESS)
print("Case 1 - invalid VLAN tag")
inv_vlan_id = 100
lrn_mac = self.macs[2]
inv_vlan_tag_pkt = simple_udp_packet(eth_dst=ap_mac,
eth_src=lrn_mac,
dl_vlan_enable=True,
vlan_vid=inv_vlan_id,
pktlen=104)
chck_inv_vlan_pkt = simple_udp_packet(eth_dst=lrn_mac,
eth_src=ap_mac,
pktlen=100)
chck_inv_vlan_tag_pkt = simple_udp_packet(eth_dst=lrn_mac,
eth_src=ap_mac,
dl_vlan_enable=True,
vlan_vid=self.vlan_id,
pktlen=104)
flood_pkt_list = [chck_inv_vlan_tag_pkt, chck_inv_vlan_pkt,
chck_inv_vlan_tag_pkt]
print("Sending packet with invalid VLAN tag on port %d, %s -> %s" %
(trunk_port, lrn_mac, ap_mac))
send_packet(self, trunk_port, inv_vlan_tag_pkt)
verify_no_other_packets(self)
print("\tPacket dropped")
print("Checking if MAC %s was not learned on port %d" %
(lrn_mac, trunk_port))
print("Sending packet on port %d, %s -> %s - will flood" %
(access_port, ap_mac, lrn_mac))
send_packet(self, access_port, chck_inv_vlan_pkt)
verify_each_packet_on_multiple_port_lists(self, flood_pkt_list,
flood_port_list)
print("\tVerification complete\n")
print("Case 2 - broadcast src_mac address")
bcast_mac = "ff:ff:ff:ff:ff:ff"
bcast_src_tag_pkt = simple_udp_packet(eth_dst=ap_mac,
eth_src=bcast_mac,
dl_vlan_enable=True,
vlan_vid=self.vlan_id,
pktlen=104)
chck_bcast_src_pkt = simple_udp_packet(eth_dst=bcast_mac,
eth_src=ap_mac,
pktlen=100)
chck_bcast_src_tag_pkt = simple_udp_packet(eth_dst=bcast_mac,
eth_src=ap_mac,
dl_vlan_enable=True,
vlan_vid=self.vlan_id,
pktlen=104)
flood_pkt_list = [chck_bcast_src_tag_pkt, chck_bcast_src_pkt,
chck_bcast_src_tag_pkt]
print("Sending packet with invalid src_mac on port %d, %s -> %s" %
(trunk_port, bcast_mac, ap_mac))
send_packet(self, trunk_port, bcast_src_tag_pkt)
verify_no_other_packets(self)
print("\tPacket dropped")
print("Checking if broadcast MAC %s was not learned on port %d" %
(bcast_mac, trunk_port))
print("Sending packet on port %d, %s -> %s - will flood" %
(access_port, ap_mac, bcast_mac))
send_packet(self, access_port, chck_bcast_src_pkt)
verify_each_packet_on_multiple_port_lists(self, flood_pkt_list,
flood_port_list)
print("\tVerification complete\n")
print("Case 3 - multicast src_mac address")
mcast_mac = "01:00:5e:11:22:33"
mcast_src_tag_pkt = simple_udp_packet(eth_dst=ap_mac,
eth_src=mcast_mac,
dl_vlan_enable=True,
vlan_vid=self.vlan_id,
pktlen=104)
chck_mcast_src_pkt = simple_udp_packet(eth_dst=mcast_mac,
eth_src=ap_mac,
pktlen=100)
chck_mcast_src_tag_pkt = simple_udp_packet(eth_dst=mcast_mac,
eth_src=ap_mac,
dl_vlan_enable=True,
vlan_vid=self.vlan_id,
pktlen=104)
flood_pkt_list = [chck_mcast_src_tag_pkt, chck_mcast_src_pkt,
chck_mcast_src_tag_pkt]
print("Sending packet with invalid src_mac on port %d, %s -> %s" %
(trunk_port, mcast_mac, ap_mac))
send_packet(self, trunk_port, mcast_src_tag_pkt)
verify_no_other_packets(self)
print("\tPacket dropped")
print("Checking if multicast MAC %s was not learned on port %d" %
(mcast_mac, trunk_port))
print("Sending packet on port %d, %s -> %s - will flood" %
(access_port, ap_mac, mcast_mac))
send_packet(self, access_port, chck_mcast_src_pkt)
verify_each_packet_on_multiple_port_lists(self, flood_pkt_list,
flood_port_list)
print("\tVerification complete\n")
print("Case 4 - src_mac statically added")
lrn_mac = self.macs[2]
bcast_mac = "ff:ff:ff:ff:ff:ff"
static_src_tag_pkt = simple_udp_packet(eth_dst=bcast_mac,
eth_src=ap_mac,
dl_vlan_enable=True,
vlan_vid=self.vlan_id,
pktlen=104)
static_src_pkt = simple_udp_packet(eth_dst=bcast_mac,
eth_src=ap_mac,
pktlen=100)
chck_static_src_pkt = simple_udp_packet(eth_dst=ap_mac,
eth_src=lrn_mac,
pktlen=100)
flood_pkt_list = [
static_src_pkt, static_src_pkt, static_src_tag_pkt]
flood_port_list = [
[access_port], self.utg_lag_ports, self.tg_lag_ports]
print("Sending packet on port %d, %s -> %s" %
(trunk_port, ap_mac, bcast_mac))
send_packet(self, trunk_port, static_src_tag_pkt)
verify_each_packet_on_multiple_port_lists(self, flood_pkt_list,
flood_port_list)
print("\tPacket flooded")
print("Checking if MAC %s was not learned on port %d" %
(ap_mac, trunk_port))
print("Sending packet on port %d, %s -> %s - forward to port %d" %
(self.utg_lag_ports[1], lrn_mac, ap_mac, access_port))
send_packet(self, self.utg_lag_ports[1], chck_static_src_pkt)
verify_packets(self, chck_static_src_pkt, [access_port])
print("\tVerification complete\n")
print("Case 5 - removed VLAN member")
rm_vlan_member = self.vlan10_member1
rm_vlan_member_dev = self.dev_port1
rm_vlan_member_mac = "00:12:34:56:78:90"
sai_thrift_remove_vlan_member(self.client, rm_vlan_member)
# check if packet is not flooded to the removed VLAN member port
pkt = simple_udp_packet(eth_dst="ff:ff:ff:ff:ff:ff",
eth_src=self.macs[0],
pktlen=100)
tag_pkt = simple_udp_packet(eth_dst="ff:ff:ff:ff:ff:ff",
eth_src=self.macs[0],
dl_vlan_enable=True,
vlan_vid=self.vlan_id,
pktlen=104)
flood_port_list = [self.utg_lag_ports, self.tg_lag_ports]
flood_pkt_list = [pkt, tag_pkt]
print("Sending packet on port %d, %s -> %s\nchecking if flood "
"doesn't reach port %d" % (self.dev_port0, self.macs[0],
"ff:ff:ff:ff:ff:ff",
rm_vlan_member_dev))
send_packet(self, self.dev_port0, pkt)
verify_each_packet_on_multiple_port_lists(self, flood_pkt_list,
flood_port_list)
print("\tOK")
# check if MAC is not learned on removed VLAN member
tag_pkt = simple_udp_packet(eth_dst="ff:ff:ff:ff:ff:ff",
eth_src=rm_vlan_member_mac,
dl_vlan_enable=True,
vlan_vid=self.vlan_id,
pktlen=104)
print("Sending packet on removed VLAN port (%d), %s -> %s - "
"will be dropped" % (rm_vlan_member_dev, rm_vlan_member_mac,
"ff:ff:ff:ff:ff:ff"))
send_packet(self, rm_vlan_member_dev, tag_pkt)
verify_no_other_packets(self)
print("\tOK")
print("Checking if MAC was not learned")
pkt = simple_udp_packet(eth_dst=rm_vlan_member_mac,
eth_src=self.macs[0],
pktlen=100)
tag_pkt = simple_udp_packet(eth_dst=rm_vlan_member_mac,
eth_src=self.macs[0],
dl_vlan_enable=True,
vlan_vid=self.vlan_id,
pktlen=104)
flood_pkt_list = [pkt, tag_pkt]
print("Sending packet on port %d, %s -> %s - will flood within "
"VLAN 10" % (self.dev_port0, self.macs[0],
rm_vlan_member_mac))
send_packet(self, self.dev_port0, pkt)
verify_each_packet_on_multiple_port_lists(self, flood_pkt_list,
flood_port_list)
print("Case 6 - removed LAG member")
rm_lag_member = self.lag1_member5
rm_lag_member_dev = self.dev_port5
rm_lag_member_mac = "00:09:87:65:43:21"
sai_thrift_remove_lag_member(self.client, rm_lag_member)
self.utg_lag_ports.remove(rm_lag_member_dev)
pkt = simple_udp_packet(eth_dst="ff:ff:ff:ff:ff:ff",
eth_src=rm_lag_member_mac,
pktlen=100)
print("Sending packet on removed LAG member (port %d), %s -> %s - "
"will be dropped" % (rm_lag_member_dev, rm_lag_member_mac,
"ff:ff:ff:ff:ff:ff"))
send_packet(self, rm_lag_member_dev, pkt)
verify_no_other_packets(self)
print("\tOK")
print("Checking if MAC was not learned")
pkt = simple_udp_packet(eth_dst=rm_lag_member_mac,
eth_src=self.macs[0],
pktlen=100)
tag_pkt = simple_udp_packet(eth_dst=rm_lag_member_mac,
eth_src=self.macs[0],
dl_vlan_enable=True,
vlan_vid=self.vlan_id,
pktlen=104)
flood_pkt_list = [pkt, tag_pkt]
print("Sending packet on port %d, %s -> %s - will flood within "
"VLAN 10" % (self.dev_port0, self.macs[0],
rm_lag_member_mac))
send_packet(self, self.dev_port0, pkt)
verify_each_packet_on_multiple_port_lists(self, flood_pkt_list,
flood_port_list)
print("\tVerification complete\n")
finally:
sai_thrift_flush_fdb_entries(
self.client, entry_type=SAI_FDB_FLUSH_ENTRY_TYPE_ALL)
# restore removed LAG member
self.lag1_member5 = sai_thrift_create_lag_member(
self.client, lag_id=self.lag1, port_id=self.port5)
self.utg_lag_ports.insert(1, rm_lag_member_dev)
# restore removed vlan member
self.vlan10_member1 = sai_thrift_create_vlan_member(
self.client, vlan_id=self.vlan10, bridge_port_id=self.port1_bp,
vlan_tagging_mode=SAI_VLAN_TAGGING_MODE_TAGGED)
@group("draft")
class FdbStaticMacTest(SaiHelper):
'''
Verify static MAC entries
'''
def setUp(self):
super(FdbStaticMacTest, self).setUp()
self.vlan_id = 10
lag = [self.dev_port4, self.dev_port5, self.dev_port6]
self.ports = [[self.dev_port0], [self.dev_port1], lag]
self.macs = []
for i in range(1, len(self.ports) + 1):
mac = "00:%02d:%02d:%02d:%02d:%02d" % (i, i, i, i, i)
self.macs.append(mac)
# populate FDB
fdb_entry1 = sai_thrift_fdb_entry_t(switch_id=self.switch_id,
mac_address=self.macs[0],
bv_id=self.vlan10)
status = sai_thrift_create_fdb_entry(
self.client,
fdb_entry1,
type=SAI_FDB_ENTRY_TYPE_STATIC,
bridge_port_id=self.port0_bp)
self.assertEqual(status, SAI_STATUS_SUCCESS)
fdb_entry2 = sai_thrift_fdb_entry_t(switch_id=self.switch_id,
mac_address=self.macs[1],
bv_id=self.vlan10)
status = sai_thrift_create_fdb_entry(
self.client,
fdb_entry2,
type=SAI_FDB_ENTRY_TYPE_STATIC,
bridge_port_id=self.port1_bp)
self.assertEqual(status, SAI_STATUS_SUCCESS)
self.fdb_entry3 = sai_thrift_fdb_entry_t(switch_id=self.switch_id,
mac_address=self.macs[2],
bv_id=self.vlan10)
status = sai_thrift_create_fdb_entry(
self.client,
self.fdb_entry3,
type=SAI_FDB_ENTRY_TYPE_STATIC,
bridge_port_id=self.lag1_bp)
self.assertEqual(status, SAI_STATUS_SUCCESS)
print("FDB entries added")
def runTest(self):
self.staticMacForwardTest()
self.selfForwardingTest()
def tearDown(self):
sai_thrift_flush_fdb_entries(
self.client, entry_type=SAI_FDB_FLUSH_ENTRY_TYPE_ALL)
super(FdbStaticMacTest, self).tearDown()
def staticMacForwardTest(self):
'''
Static MAC tests
Send a packet from one port with DMAC installed on other port and
verify packet forwarding
'''
print("\nstaticMacForwardTest()")
try:
for dst_port, dst_mac in zip(self.ports, self.macs):
for src_port, src_mac in zip([self.dev_port0, self.dev_port1],
self.macs[:1]):
if [src_port] == dst_port:
continue
pkt = simple_udp_packet(eth_dst=dst_mac,
eth_src=src_mac,
pktlen=100)
tag_pkt = simple_udp_packet(eth_dst=dst_mac,
eth_src=src_mac,
dl_vlan_enable=True,
vlan_vid=self.vlan_id,
pktlen=104)
send_pkt = tag_pkt if src_port == self.dev_port1 else pkt
rcv_pkt = tag_pkt if dst_port == [self.dev_port1] else pkt
print("Sending packet on port %d, %s -> %s - forwarding" %
(src_port, src_mac, dst_mac))
send_packet(self, src_port, send_pkt)
verify_packet_any_port(self, rcv_pkt, dst_port)
print("\tVerification complete")
except BaseException:
print("Test error occured")
def selfForwardingTest(self):
'''
Verify if packet send with ingress port MAC address is dropped
'''
print("\nselfForwardingTest()")
test_port_dev = self.dev_port0
test_mac = self.macs[0]
pkt = simple_udp_packet(eth_dst=test_mac)
print("Verifying if packet send on ingress port MAC is dropped")
send_packet(self, test_port_dev, pkt)
verify_no_other_packets(self)
print("\tOK")
@group("draft")
class FdbMacMoveTest(SaiHelper):
'''
Verify MAC entries moving
'''
def setUp(self):
super(FdbMacMoveTest, self).setUp()
# change port1 original tagging mode
sai_thrift_remove_vlan_member(self.client, self.vlan10_member1)
self.vlan10_member1 = sai_thrift_create_vlan_member(
self.client,
vlan_id=self.vlan10,
bridge_port_id=self.port1_bp,
vlan_tagging_mode=SAI_VLAN_TAGGING_MODE_UNTAGGED)
sai_thrift_set_port_attribute(self.client, self.port1, port_vlan_id=10)
# add one more access port to VLAN 10
self.port24_bp = sai_thrift_create_bridge_port(
self.client,
bridge_id=self.default_1q_bridge,
port_id=self.port24,
type=SAI_BRIDGE_PORT_TYPE_PORT,
admin_state=True)
self.assertNotEqual(self.port24_bp, 0)
self.vlan10_member3 = sai_thrift_create_vlan_member(
self.client,
vlan_id=self.vlan10,
bridge_port_id=self.port24_bp,
vlan_tagging_mode=SAI_VLAN_TAGGING_MODE_UNTAGGED)
sai_thrift_set_port_attribute(
self.client, self.port24, port_vlan_id=10)
# create additional LAG in VLAN 10
self.lag10 = sai_thrift_create_lag(self.client)
self.assertNotEqual(self.lag10, 0)
self.lag10_bp = sai_thrift_create_bridge_port(
self.client,
bridge_id=self.default_1q_bridge,
port_id=self.lag10,
type=SAI_BRIDGE_PORT_TYPE_PORT,
admin_state=True)
self.assertNotEqual(self.lag10_bp, 0)
self.lag10_member25 = sai_thrift_create_lag_member(
self.client, lag_id=self.lag10, port_id=self.port25)
self.lag10_member26 = sai_thrift_create_lag_member(
self.client, lag_id=self.lag10, port_id=self.port26)
self.lag10_member27 = sai_thrift_create_lag_member(
self.client, lag_id=self.lag10, port_id=self.port27)
self.vlan10_member4 = sai_thrift_create_vlan_member(
self.client,
vlan_id=self.vlan10,
bridge_port_id=self.lag10_bp,
vlan_tagging_mode=SAI_VLAN_TAGGING_MODE_UNTAGGED)
sai_thrift_set_lag_attribute(self.client, self.lag10, port_vlan_id=10)
self.moving_mac = "00:11:22:33:44:55"
self.chck_mac = "00:11:11:11:11:11"
# add static MAC for port used for verification
self.fdb_entry = sai_thrift_fdb_entry_t(switch_id=self.switch_id,
mac_address=self.chck_mac,
bv_id=self.vlan10)
status = sai_thrift_create_fdb_entry(self.client,
self.fdb_entry,
type=SAI_FDB_ENTRY_TYPE_STATIC,
bridge_port_id=self.port24_bp)
self.assertEqual(status, SAI_STATUS_SUCCESS)
def runTest(self):
self.dynamicMacMoveTest()
self.dynamicMacMoveTest(static_entry=True)
self.staticMacMoveTest()
def tearDown(self):
sai_thrift_flush_fdb_entries(
self.client, entry_type=SAI_FDB_FLUSH_ENTRY_TYPE_ALL)
# remove additionally created LAG
sai_thrift_set_lag_attribute(self.client, self.lag1, port_vlan_id=0)
sai_thrift_remove_vlan_member(self.client, self.vlan10_member4)
sai_thrift_remove_lag_member(self.client, self.lag10_member27)
sai_thrift_remove_lag_member(self.client, self.lag10_member26)
sai_thrift_remove_lag_member(self.client, self.lag10_member25)
sai_thrift_remove_bridge_port(self.client, self.lag10_bp)
sai_thrift_remove_lag(self.client, self.lag10)
# remove additional port from VLAN 10
sai_thrift_set_port_attribute(self.client, self.port24, port_vlan_id=0)
sai_thrift_remove_bridge_port(self.client, self.port24_bp)
sai_thrift_remove_vlan_member(self.client, self.vlan10_member3)
# restore port1 original tagging mode
sai_thrift_set_port_attribute(self.client, self.port1, port_vlan_id=0)
sai_thrift_remove_vlan_member(self.client, self.vlan10_member1)
self.vlan10_member1 = sai_thrift_create_vlan_member(
self.client,
vlan_id=self.vlan10,
bridge_port_id=self.port1_bp,
vlan_tagging_mode=SAI_VLAN_TAGGING_MODE_TAGGED)
super(FdbMacMoveTest, self).tearDown()
def dynamicMacMoveTest(self, static_entry=False):
'''
Dynamic MAC move test that verify if after receiving packet with known
SMAC, but from other port that on which it was learned previously,
next packets with such DMACs are forwarded to the new port.
The test represents a chain of moves:
- port -> port
- port -> LAG
- LAG -> LAG
- LAG -> port
Args:
static_entry (bool): Check move for static fdb entry
'''
print("\ndynamicMacMoveTest()")
# a series of ports representing moving chain:
# port->port->LAG->LAG->port
port_chain = [self.dev_port0, self.dev_port1, self.dev_port5,
self.dev_port27, self.dev_port1]
chck_port = self.dev_port24
lag1_ports = [self.dev_port4, self.dev_port5, self.dev_port6]
lag3_ports = [self.dev_port25, self.dev_port26, self.dev_port27]
pkt = simple_udp_packet(eth_dst=self.chck_mac, eth_src=self.moving_mac)
chck_pkt = simple_udp_packet(eth_dst=self.moving_mac,
eth_src=self.chck_mac)
try:
if static_entry:
# inititally add moving MAC to FDB as static
moving_fdb_entry = sai_thrift_fdb_entry_t(
switch_id=self.switch_id,
mac_address=self.moving_mac,
bv_id=self.vlan10)
status = sai_thrift_create_fdb_entry(
self.client,
moving_fdb_entry,
type=SAI_FDB_ENTRY_TYPE_STATIC,
bridge_port_id=self.port0_bp,
allow_mac_move=True)
self.assertEqual(status, SAI_STATUS_SUCCESS)
for src_port in port_chain:
print("Sending packet on port %d, %s -> %s - learn and "
"forward to port %d" % (src_port, self.moving_mac,
self.chck_mac, chck_port))
send_packet(self, src_port, pkt)
verify_packets(self, pkt, [chck_port])
print("MAC learned")
if src_port in lag1_ports:
rcv_port = lag1_ports
print("Sending packet on port %d, %s -> %s - forward to "
"LAG1" % (chck_port, self.chck_mac, self.moving_mac))
elif src_port in lag3_ports:
rcv_port = lag3_ports
print("Sending packet on port %d, %s -> %s - forward to "
"LAG3" % (chck_port, self.chck_mac, self.moving_mac))
else:
rcv_port = [src_port]
print("Sending packet on port %d, %s -> %s - forward to "
"port %d" % (chck_port, self.chck_mac,
self.moving_mac, src_port))
send_packet(self, chck_port, chck_pkt)
verify_packet_any_port(self, chck_pkt, rcv_port)
print("\tVerification complete\n")
finally:
if static_entry:
sai_thrift_remove_fdb_entry(self.client, moving_fdb_entry)
else:
# flush dynamic MACs from FDB
sai_thrift_flush_fdb_entries(
self.client, entry_type=SAI_FDB_FLUSH_ENTRY_TYPE_DYNAMIC)
def staticMacMoveTest(self):
'''
static MAC move tests.
Changee a bridge port id of some static FDB entry and check that
a packet is forwarded to this new port.
'''
print("\nstaticMacMoveTest()")
# a series of ports representing moving chain:
# port->port->LAG->LAG->port
port_chain = [self.dev_port0, self.dev_port1, self.dev_port5,
self.dev_port27, self.dev_port1 ]
bport_chain = [self.port0_bp, self.port1_bp, self.lag1_bp,
self.lag10_bp, self.port1_bp]
chck_port = self.dev_port24
lag1_ports = [self.dev_port4, self.dev_port5, self.dev_port6]
lag3_ports = [self.dev_port25, self.dev_port26, self.dev_port27]
chck_pkt = simple_udp_packet(eth_dst=self.moving_mac,
eth_src=self.chck_mac)
try:
# inititally add moving MAC to FDB
moving_fdb_entry = sai_thrift_fdb_entry_t(
switch_id=self.switch_id,
mac_address=self.moving_mac,
bv_id=self.vlan10)
status = sai_thrift_create_fdb_entry(
self.client,
moving_fdb_entry,
type=SAI_FDB_ENTRY_TYPE_STATIC,
bridge_port_id=self.port0_bp)
self.assertEqual(status, SAI_STATUS_SUCCESS)
for port, bport in zip(port_chain, bport_chain):
status = sai_thrift_set_fdb_entry_attribute(
self.client, moving_fdb_entry, bridge_port_id=bport)
self.assertEqual(status, SAI_STATUS_SUCCESS)
if port in lag1_ports:
rcv_port = lag1_ports
print("Sending packet on port %d, %s -> %s - forward to "
"LAG1" % (chck_port, self.chck_mac, self.moving_mac))
elif port in lag3_ports:
rcv_port = lag3_ports
print("Sending packet on port %d, %s -> %s - forward to "
"LAG3" % (chck_port, self.chck_mac, self.moving_mac))
else:
rcv_port = [port]
print("Sending packet on port %d, %s -> %s - forward to "
"port %d" % (chck_port, self.chck_mac,
self.moving_mac, port))
send_packet(self, chck_port, chck_pkt)
verify_packet_any_port(self, chck_pkt, rcv_port)
print("\tVerification complete\n")
finally:
sai_thrift_remove_fdb_entry(self.client, moving_fdb_entry)
@group("draft")
class FdbFlushTest(SaiHelper):
'''
MAC flush test that checks flushing functionality for static and dynamic
MAC addresses in FDB.
Test includes flushing per vlan, per bridge port, per vlan/bridge port
for static and dynamic MACs, all static MACs, all dynamic MACs and all
MACs in FDB.
'''
def setUp(self):
super(FdbFlushTest, self).setUp()
# change port1 (VLAN 10) original tagging mode
sai_thrift_remove_vlan_member(self.client, self.vlan10_member1)
self.vlan10_member1 = sai_thrift_create_vlan_member(
self.client,
vlan_id=self.vlan10,
bridge_port_id=self.port1_bp,
vlan_tagging_mode=SAI_VLAN_TAGGING_MODE_UNTAGGED)
sai_thrift_set_port_attribute(self.client, self.port1, port_vlan_id=10)
# change port3 (VLAN 20) original tagging mode
sai_thrift_remove_vlan_member(self.client, self.vlan20_member1)
self.vlan20_member1 = sai_thrift_create_vlan_member(
self.client,
vlan_id=self.vlan20,
bridge_port_id=self.port3_bp,
vlan_tagging_mode=SAI_VLAN_TAGGING_MODE_UNTAGGED)
sai_thrift_set_port_attribute(self.client, self.port3, port_vlan_id=20)
# change lag2 (VLAN 20) original tagging mode
sai_thrift_remove_vlan_member(self.client, self.vlan20_member2)
self.vlan20_member2 = sai_thrift_create_vlan_member(
self.client,
vlan_id=self.vlan20,
bridge_port_id=self.lag2_bp,
vlan_tagging_mode=SAI_VLAN_TAGGING_MODE_UNTAGGED)
sai_thrift_set_lag_attribute(self.client, self.lag2, port_vlan_id=20)
self.vlan10_id = 10
self.vlan20_id = 20
self.tp10_stat_mac = None
self.tp10_dyn_mac = None
self.tp20_stat_mac = None
self.tp20_dyn_mac = None
self.vlan10_member3 = None
self.vlan20_member3 = None
self.vlan10_ports = [self.dev_port0, self.dev_port1, self.dev_port4,
self.dev_port5, self.dev_port6]
self.vlan10_ports_bp = [self.port0_bp, self.port1_bp, self.lag1_bp,
self.lag1_bp, self.lag1_bp]
self.vlan10_lag_ports = [
self.dev_port4, self.dev_port5, self.dev_port6]
self.vlan10_stat_macs = []
self.vlan10_dyn_macs = []
for i in range(1, len(self.vlan10_ports) + 1):
mac = "00:10:00:%02d:%02d:%02d" % (i, i, i)
self.vlan10_stat_macs.append(mac)
mac = "00:10:ff:%02d:%02d:%02d" % (i, i, i)
self.vlan10_dyn_macs.append(mac)
self.vlan20_ports = [self.dev_port2, self.dev_port3, self.dev_port7,
self.dev_port8, self.dev_port9]
self.vlan20_ports_bp = [self.port2_bp, self.port3_bp, self.lag2_bp,
self.lag2_bp, self.lag2_bp]
self.vlan20_lag_ports = [
self.dev_port7, self.dev_port8, self.dev_port9]
self.vlan20_stat_macs = []
self.vlan20_dyn_macs = []
for i in range(1, len(self.vlan20_ports) + 1):
mac = "00:20:00:%02d:%02d:%02d" % (i, i, i)
self.vlan20_stat_macs.append(mac)
mac = "00:20:ff:%02d:%02d:%02d" % (i, i, i)
self.vlan20_dyn_macs.append(mac)
# create trunk port for both: VLAN 10 and VLAN 20
self.port24_bp = sai_thrift_create_bridge_port(
self.client,
bridge_id=self.default_1q_bridge,
port_id=self.port24,
type=SAI_BRIDGE_PORT_TYPE_PORT,
admin_state=True)
self.trunk_port_bp = self.port24_bp
self.trunk_dev_port = self.dev_port24
def runTest(self):
self.flushStaticPerVlanTest()
self.flushDynamicPerVlanTest()
self.flushAllPerVlanTest()
self.flushStaticPerPortTest()
self.flushDynamicPerPortTest()
self.flushAllPerPortTest()
self.flushStaticPerLagTest()
self.flushDynamicPerLagTest()
self.flushAllPerLagTest()
self.flushStaticPerVlanAndPortTest()
self.flushDynamicPerVlanAndPortTest()
self.flushAllPerVlanAndPortTest()
self.flushAllStaticTest()
self.flushAllDynamicTest()
self.flushAllMacsTest()
def tearDown(self):
sai_thrift_flush_fdb_entries(
self.client, entry_type=SAI_FDB_FLUSH_ENTRY_TYPE_ALL)
sai_thrift_remove_bridge_port(self.client, self.port24_bp)
# restore port1 (VLAN 10) original tagging mode
sai_thrift_set_port_attribute(self.client, self.port1, port_vlan_id=0)
sai_thrift_remove_vlan_member(self.client, self.vlan10_member1)
self.vlan10_member1 = sai_thrift_create_vlan_member(
self.client,
vlan_id=self.vlan10,
bridge_port_id=self.port1_bp,
vlan_tagging_mode=SAI_VLAN_TAGGING_MODE_TAGGED)
# restore port3 (VLAN 20) original tagging mode
sai_thrift_set_port_attribute(self.client, self.port3, port_vlan_id=0)
sai_thrift_remove_vlan_member(self.client, self.vlan20_member1)
self.vlan20_member1 = sai_thrift_create_vlan_member(
self.client,
vlan_id=self.vlan20,
bridge_port_id=self.port3_bp,
vlan_tagging_mode=SAI_VLAN_TAGGING_MODE_TAGGED)
# restore lag2 (VLAN 20) original tagging mode
sai_thrift_set_lag_attribute(self.client, self.lag2, port_vlan_id=0)
sai_thrift_remove_vlan_member(self.client, self.vlan20_member2)
self.vlan20_member2 = sai_thrift_create_vlan_member(
self.client,
vlan_id=self.vlan20,
bridge_port_id=self.lag2_bp,
vlan_tagging_mode=SAI_VLAN_TAGGING_MODE_TAGGED)
super(FdbFlushTest, self).tearDown()
# additional helper functions
def _prepareFdb(self):
'''
Helper function for preparing FDB with a large number of entries
'''
sai_thrift_flush_fdb_entries(
self.client, entry_type=SAI_FDB_FLUSH_ENTRY_TYPE_ALL)
print("Preparing FDB")
# add static FDB entries
for i in range(len(self.vlan10_ports)):
setattr(
self, 'vlan10_fdb_entry%d' % i,
sai_thrift_fdb_entry_t(
switch_id=self.switch_id,
mac_address=self.vlan10_stat_macs[i],
bv_id=self.vlan10))
sai_thrift_create_fdb_entry(
self.client,
getattr(self, "vlan10_fdb_entry%d" % i),
type=SAI_FDB_ENTRY_TYPE_STATIC,
bridge_port_id=self.vlan10_ports_bp[i])
self.assertEqual(self.status(), SAI_STATUS_SUCCESS)
for i in range(len(self.vlan20_ports)):
setattr(self, 'vlan20_fdb_entry%d' % i,
sai_thrift_fdb_entry_t(
switch_id=self.switch_id,
mac_address=self.vlan20_stat_macs[i],
bv_id=self.vlan20))
sai_thrift_create_fdb_entry(
self.client,
getattr(self, "vlan20_fdb_entry%d" % i),
type=SAI_FDB_ENTRY_TYPE_STATIC,
bridge_port_id=self.vlan20_ports_bp[i])
self.assertEqual(self.status(), SAI_STATUS_SUCCESS)
print("Static MAC enries added")
# learn dynamic MACs
for mac, port in zip(self.vlan10_dyn_macs, self.vlan10_ports):
pkt = simple_udp_packet(eth_dst="ff:ff:ff:ff:ff:ff",
eth_src=mac)
send_packet(self, port, pkt)
for mac, port in zip(self.vlan20_dyn_macs, self.vlan20_ports):
pkt = simple_udp_packet(eth_dst="ff:ff:ff:ff:ff:ff",
eth_src=mac)
send_packet(self, port, pkt)
print("Dynamic MACs learned")
time.sleep(2)
self.dataplane.flush()
print("\tFDB entries prepared")
def _setUpTrunkPort(self):
'''
Helper function for trunk port setting
'''
self.tp10_stat_mac = "00:10:00:66:66:66"
self.tp10_dyn_mac = "00:10:ff:66:66:66"
chck_vlan10_mac = self.vlan10_stat_macs[0]
chck_vlan10_port = self.dev_port0
self.tp20_stat_mac = "00:20:00:66:66:66"
self.tp20_dyn_mac = "00:20:ff:66:66:66"
chck_vlan20_mac = self.vlan20_stat_macs[0]
chck_vlan20_port = self.dev_port2
print("Set up common trunk port for VLAN 10 and VLAN 20")
self.vlan10_member3 = sai_thrift_create_vlan_member(
self.client,
vlan_id=self.vlan10,
bridge_port_id=self.trunk_port_bp,
vlan_tagging_mode=SAI_VLAN_TAGGING_MODE_TAGGED)
self.assertNotEqual(self.vlan10_member3, 0)
self.vlan20_member3 = sai_thrift_create_vlan_member(
self.client,
vlan_id=self.vlan20,
bridge_port_id=self.trunk_port_bp,
vlan_tagging_mode=SAI_VLAN_TAGGING_MODE_TAGGED)
self.assertNotEqual(self.vlan20_member3, 0)
# add static MACs on trunk port
vlan10_tp_fdb_entry = sai_thrift_fdb_entry_t(
switch_id=self.switch_id,
mac_address=self.tp10_stat_mac,
bv_id=self.vlan10)
status = sai_thrift_create_fdb_entry(self.client,
vlan10_tp_fdb_entry,
type=SAI_FDB_ENTRY_TYPE_STATIC,
bridge_port_id=self.trunk_port_bp)
self.assertEqual(status, SAI_STATUS_SUCCESS)
vlan20_tp_fdb_entry = sai_thrift_fdb_entry_t(
switch_id=self.switch_id,
mac_address=self.tp20_stat_mac,
bv_id=self.vlan20)
status = sai_thrift_create_fdb_entry(self.client,
vlan20_tp_fdb_entry,
type=SAI_FDB_ENTRY_TYPE_STATIC,
bridge_port_id=self.trunk_port_bp)
self.assertEqual(status, SAI_STATUS_SUCCESS)
# learn trunk port dynamic MACs
tag_vlan10_pkt = simple_udp_packet(eth_dst="ff:ff:ff:ff:ff:ff",
eth_src=self.tp10_dyn_mac,
dl_vlan_enable=True,
vlan_vid=self.vlan10_id,
pktlen=104)
tag_vlan20_pkt = simple_udp_packet(eth_dst="ff:ff:ff:ff:ff:ff",
eth_src=self.tp20_dyn_mac,
dl_vlan_enable=True,
vlan_vid=self.vlan20_id,
pktlen=104)
send_packet(self, self.trunk_dev_port, tag_vlan10_pkt)
send_packet(self, self.trunk_dev_port, tag_vlan20_pkt)
time.sleep(2)
self.dataplane.flush()
for mac in [self.tp10_stat_mac, self.tp10_dyn_mac]:
chck_vlan10_pkt = simple_udp_packet(eth_dst=mac,
eth_src=chck_vlan10_mac,
pktlen=100)
chck_vlan10_tag_pkt = simple_udp_packet(
eth_dst=mac,
eth_src=chck_vlan10_mac,
dl_vlan_enable=True,
vlan_vid=self.vlan10_id,
pktlen=104)
send_packet(self, chck_vlan10_port, chck_vlan10_pkt)
verify_packets(self, chck_vlan10_tag_pkt,
[self.trunk_dev_port])
for mac in [self.tp20_stat_mac, self.tp20_dyn_mac]:
chck_vlan20_pkt = simple_udp_packet(eth_dst=mac,
eth_src=chck_vlan20_mac,
pktlen=100)
chck_vlan20_tag_pkt = simple_udp_packet(
eth_dst=mac,
eth_src=chck_vlan20_mac,
dl_vlan_enable=True,
vlan_vid=self.vlan20_id,
pktlen=104)
send_packet(self, chck_vlan20_port, chck_vlan20_pkt)
verify_packets(self, chck_vlan20_tag_pkt,
[self.trunk_dev_port])
print("\tTrunk port configured")
def _tearDownTrunkPort(self):
'''
Helper function for trunk port removing
'''
# flush trunk_port MACs
sai_thrift_flush_fdb_entries(self.client,
bridge_port_id=self.trunk_port_bp,
entry_type=SAI_FDB_FLUSH_ENTRY_TYPE_ALL)
sai_thrift_remove_vlan_member(self.client, self.vlan10_member3)
sai_thrift_remove_vlan_member(self.client, self.vlan20_member3)
def _verifyFwd(self, dst_macs, dst_ports, src_macs, src_ports,
lag_ports=None, trunk_port=None, vlan_id=None):
'''
Verify if packets with DMAC from dst_macs list are forwarded to
corresponding ports from dst_ports list.
*Assuming that src and dst ports are access ports.
Args:
dst_macs (list): a list of DMACs
dst_ports (list): a list of ports corresponding to the DMACs
src_macs (list): a list of SMACs corresponding to ports taken for
trying to send packets from
src_ports (list): a list of ports taken for trying to send packets
from
lag_ports (list): opional list if some ports are in lag
trunk_port (int): optional trunk port number (if exists in
dst_ports)
vlan_id (int): optional VLAN number required if trunk_port is given
'''
for dst_mac, dst_port in zip(dst_macs, dst_ports):
for src_mac, src_port in zip(src_macs, src_ports):
if src_port == dst_port:
continue
pkt = simple_udp_packet(eth_dst=dst_mac,
eth_src=src_mac,
pktlen=100)
rcv_pkt = pkt
if dst_port == trunk_port:
tag_pkt = simple_udp_packet(eth_dst=dst_mac,
eth_src=src_mac,
dl_vlan_enable=True,
vlan_vid=vlan_id,
pktlen=104)
rcv_pkt = tag_pkt
rcv_port = lag_ports if dst_port in lag_ports else [dst_port]
send_packet(self, src_port, pkt)
verify_packet_any_port(self, rcv_pkt, rcv_port)
break
def _verifyFlood(self, dst_macs, dst_ports, src_macs, src_ports,
lag_ports=None, trunk_port=None, vlan_id=None):
'''
Verify if packets with DMAC from dst_macs list are flooded to ports
from dst_ports list.
*Assuming that src and dst ports are access ports.
Args:
dst_macs (list): a list of DMACs
dst_ports (list): a list of ports corresponding to the DMACs
src_macs (list): a list of SMACs corresponding to ports taken for
trying to send packets from
src_ports (list): a list of ports taken for trying to send packets
from
lag_ports (list): opional list if some ports are in lag
trunk_port (int): optional trunk port number (if exists in
dst_ports)
vlan_id (int): optional VLAN number required if trunk_port is given
'''
for dst_mac in dst_macs:
for src_mac, src_port in zip(src_macs, src_ports):
if src_mac == dst_mac:
continue
pkt = simple_udp_packet(eth_dst=dst_mac,
eth_src=src_mac,
pktlen=100)
port_list = []
lag_set = False
for port in dst_ports:
if port == src_port:
continue
if port in lag_ports:
if lag_set:
continue
port_list.append(lag_ports)
lag_set = True
else:
port_list.append([port])
pkt_list = [pkt] * len(port_list)
if trunk_port is not None:
port_list.append([trunk_port])
tag_pkt = simple_udp_packet(eth_dst=dst_mac,
eth_src=src_mac,
dl_vlan_enable=True,
vlan_vid=vlan_id,
pktlen=104)
pkt_list.append(tag_pkt)
send_packet(self, src_port, pkt)
verify_each_packet_on_multiple_port_lists(self, pkt_list,
port_list)
break
# tests funcions
def flushStaticPerVlanTest(self):
'''
Verify flushing of static MAC entries by VLAN
'''
print("\nflushStaticPerVlanTest()")
self._prepareFdb()
print("Flush static MACs by VLAN on VLAN 10")
sai_thrift_flush_fdb_entries(
self.client,
bv_id=self.vlan10,
entry_type=SAI_FDB_FLUSH_ENTRY_TYPE_STATIC)
print("Verify flooding on VLAN 10 static FDB entries")
self._verifyFlood(self.vlan10_stat_macs, self.vlan10_ports,
[self.vlan10_dyn_macs[0], self.vlan10_dyn_macs[1]],
[self.dev_port0, self.dev_port1],
self.vlan10_lag_ports)
print("Verify forwarding for all other FDB entires")
# verify VLAN 20 static MACs
self._verifyFwd(self.vlan20_stat_macs, self.vlan20_ports,
[self.vlan20_stat_macs[0], self.vlan20_stat_macs[1]],
[self.dev_port2, self.dev_port3],
self.vlan20_lag_ports)
# verify VLAN 10 dynamic MACs
self._verifyFwd(self.vlan10_dyn_macs, self.vlan10_ports,
[self.vlan10_stat_macs[0], self.vlan10_stat_macs[1]],
[self.dev_port0, self.dev_port1],
self.vlan10_lag_ports)
# verify VLAN 20 dynamic MACs
self._verifyFwd(self.vlan20_dyn_macs, self.vlan20_ports,
[self.vlan20_stat_macs[0], self.vlan20_stat_macs[1]],
[self.dev_port2, self.dev_port3],
self.vlan20_lag_ports)
print("\tVerification complete")
def flushDynamicPerVlanTest(self):
'''
Verify flushing of dynamic MAC entries by VLAN
'''
print("\nflushDynamicPerVlanTest()")
self._prepareFdb()
print("Flush dynamic MACs by VLAN on VLAN 10")
sai_thrift_flush_fdb_entries(
self.client,
bv_id=self.vlan10,
entry_type=SAI_FDB_FLUSH_ENTRY_TYPE_DYNAMIC)
print("Verify flooding on VLAN 10 dynamic FDB entries")
self._verifyFlood(self.vlan10_dyn_macs, self.vlan10_ports,
[self.vlan10_stat_macs[0], self.vlan10_stat_macs[1]],
[self.dev_port0, self.dev_port1],
self.vlan10_lag_ports)
print("Verify forwarding for all other FDB entires")
# verify VLAN 10 static MACs
self._verifyFwd(self.vlan10_stat_macs, self.vlan10_ports,
[self.vlan10_stat_macs[0], self.vlan10_stat_macs[1]],
[self.dev_port0, self.dev_port1],
self.vlan10_lag_ports)
# verify VLAN 20 static MACs
self._verifyFwd(self.vlan20_stat_macs, self.vlan20_ports,
[self.vlan20_stat_macs[0], self.vlan20_stat_macs[1]],
[self.dev_port2, self.dev_port3],
self.vlan20_lag_ports)
# verify VLAN 20 dynamic MACs
self._verifyFwd(self.vlan20_dyn_macs, self.vlan20_ports,
[self.vlan20_stat_macs[0], self.vlan20_stat_macs[1]],
[self.dev_port2, self.dev_port3],
self.vlan20_lag_ports)
print("\tVerification complete")
def flushAllPerVlanTest(self):
'''
Verify flushing of all kinds of MAC entries by VLAN
'''
print("\nflushAllPerVlanTest()")
self._prepareFdb()
print("Flush MACs by VLAN on VLAN 10")
sai_thrift_flush_fdb_entries(self.client,
bv_id=self.vlan10,
entry_type=SAI_FDB_FLUSH_ENTRY_TYPE_ALL)
chck_mac1 = "00:10:aa:11:11:11"
chck_mac2 = "00:10:aa:22:22:22"
print("Verify flooding on VLAN 10 FDB entries")
self._verifyFlood(self.vlan10_stat_macs, self.vlan10_ports,
[chck_mac1, chck_mac2],
[self.dev_port0, self.dev_port1],
self.vlan10_lag_ports)
self._verifyFlood(self.vlan10_dyn_macs, self.vlan10_ports,
[chck_mac1, chck_mac2],
[self.dev_port0, self.dev_port1],
self.vlan10_lag_ports)
print("Verify forwarding for all other FDB entires")
# verify VLAN 20 static MACs
self._verifyFwd(self.vlan20_stat_macs, self.vlan20_ports,
[self.vlan20_stat_macs[0], self.vlan20_stat_macs[1]],
[self.dev_port2, self.dev_port3],
self.vlan20_lag_ports)
# verify VLAN 20 dynamic MACs
self._verifyFwd(self.vlan20_dyn_macs, self.vlan20_ports,
[self.vlan20_stat_macs[0], self.vlan20_stat_macs[1]],
[self.dev_port2, self.dev_port3],
self.vlan20_lag_ports)
print("\tVerification complete")
def flushStaticPerPortTest(self):
'''
Verify flushing of static MAC entries by port
'''
print("\nflushStaticPerPortTest()")
self._prepareFdb()
flushed_port_bp = self.port0_bp
flushed_dev_port = self.vlan10_ports[0]
flushed_mac = self.vlan10_stat_macs[0]
print("Flush static MAC by port on port %d" % flushed_dev_port)
sai_thrift_flush_fdb_entries(
self.client,
bridge_port_id=flushed_port_bp,
entry_type=SAI_FDB_FLUSH_ENTRY_TYPE_STATIC)
print("Verify flooding on flushed port static MAC")
self._verifyFlood([flushed_mac], self.vlan10_ports,
[self.vlan10_dyn_macs[1]], [self.vlan10_ports[1]],
self.vlan10_lag_ports)
not_flushed_ports = [
self.vlan10_ports[i] for i in range(len(self.vlan10_ports))
if self.vlan10_ports[i] != flushed_dev_port]
not_flushed_macs = [
self.vlan10_stat_macs[i] for i in range(len(self.vlan10_stat_macs))
if self.vlan10_stat_macs[i] != flushed_mac]
print("Verify forwarding for all other FDB entires")
# verify VLAN 10 static MACs
self._verifyFwd(not_flushed_macs, not_flushed_ports,
[self.vlan10_dyn_macs[0], self.vlan10_dyn_macs[1]],
[self.dev_port0, self.dev_port1],
self.vlan10_lag_ports)
# verify VLAN 20 static MACs
self._verifyFwd(self.vlan20_stat_macs, self.vlan20_ports,
[self.vlan20_stat_macs[0], self.vlan20_stat_macs[1]],
[self.dev_port2, self.dev_port3],
self.vlan20_lag_ports)
# verify VLAN 10 dynamic MACs
self._verifyFwd(self.vlan10_dyn_macs, self.vlan10_ports,
[self.vlan10_stat_macs[0], self.vlan10_stat_macs[1]],
[self.dev_port0, self.dev_port1],
self.vlan10_lag_ports)
# verify VLAN 20 dynamic MACs
self._verifyFwd(self.vlan20_dyn_macs, self.vlan20_ports,
[self.vlan20_stat_macs[0], self.vlan20_stat_macs[1]],
[self.dev_port2, self.dev_port3],
self.vlan20_lag_ports)
print("\tVerification complete")
def flushDynamicPerPortTest(self):
'''
Verify flushing of dynamic MAC entries by port
'''
print("\nflushDynamicPerPortTest()")
self._prepareFdb()
flushed_port_bp = self.port0_bp
flushed_dev_port = self.vlan10_ports[0]
flushed_mac = self.vlan10_dyn_macs[0]
print("Flush dynamic MAC by port on port %d" % flushed_dev_port)
sai_thrift_flush_fdb_entries(
self.client,
bridge_port_id=flushed_port_bp,
entry_type=SAI_FDB_FLUSH_ENTRY_TYPE_DYNAMIC)
print("Verify flooding on flushed port static MAC")
self._verifyFlood([flushed_mac], self.vlan10_ports,
[self.vlan10_stat_macs[1]], [self.vlan10_ports[1]],
self.vlan10_lag_ports)
not_flushed_ports = [
self.vlan10_ports[i] for i in range(len(self.vlan10_ports))
if self.vlan10_ports[i] != flushed_dev_port]
not_flushed_macs = [
self.vlan10_dyn_macs[i] for i in range(len(self.vlan10_dyn_macs))
if self.vlan10_dyn_macs[i] != flushed_mac]
print("Verify forwarding for all other FDB entires")
# verify VLAN 10 static MACs
self._verifyFwd(self.vlan10_stat_macs, self.vlan10_ports,
[self.vlan10_stat_macs[0], self.vlan10_stat_macs[1]],
[self.dev_port0, self.dev_port1],
self.vlan10_lag_ports)
# verify VLAN 20 static MACs
self._verifyFwd(self.vlan20_stat_macs, self.vlan20_ports,
[self.vlan20_stat_macs[0], self.vlan20_stat_macs[1]],
[self.dev_port2, self.dev_port3],
self.vlan20_lag_ports)
# verify VLAN 10 dynamic MACs
self._verifyFwd(not_flushed_macs, not_flushed_ports,
[self.vlan10_stat_macs[0], self.vlan10_stat_macs[1]],
[self.dev_port0, self.dev_port1],
self.vlan10_lag_ports)
# verify VLAN 20 dynamic MACs
self._verifyFwd(self.vlan20_dyn_macs, self.vlan20_ports,
[self.vlan20_stat_macs[0], self.vlan20_stat_macs[1]],
[self.dev_port2, self.dev_port3],
self.vlan20_lag_ports)
print("\tVerification complete")
def flushAllPerPortTest(self):
'''
Verify flushing of all kinds of MAC entries by port
'''
print("\nflushAllPerPortTest()")
self._prepareFdb()
flushed_port_bp = self.port0_bp
flushed_dev_port = self.vlan10_ports[0]
flushed_stat_mac = self.vlan10_stat_macs[0]
flushed_dyn_mac = self.vlan10_dyn_macs[0]
print("Flush all MACs by port on port %d" % flushed_dev_port)
sai_thrift_flush_fdb_entries(self.client,
bridge_port_id=flushed_port_bp,
entry_type=SAI_FDB_FLUSH_ENTRY_TYPE_ALL)
print("Verify flooding on flushed port MACs")
self._verifyFlood([flushed_stat_mac], self.vlan10_ports,
[self.vlan10_dyn_macs[1]], [self.vlan10_ports[1]],
self.vlan10_lag_ports)
self._verifyFlood([flushed_dyn_mac], self.vlan10_ports,
[self.vlan10_stat_macs[1]], [self.vlan10_ports[1]],
self.vlan10_lag_ports)
not_flushed_ports = [
self.vlan10_ports[i] for i in range(len(self.vlan10_ports))
if self.vlan10_ports[i] != flushed_dev_port]
not_flushed_stat_macs = [
self.vlan10_stat_macs[i] for i in range(len(self.vlan10_stat_macs))
if self.vlan10_stat_macs[i] != flushed_stat_mac]
not_flushed_dyn_macs = [
self.vlan10_dyn_macs[i] for i in range(len(self.vlan10_dyn_macs))
if self.vlan10_dyn_macs[i] != flushed_dyn_mac]
print("Verify forwarding for all other FDB entires")
# verify VLAN 10 static MACs
self._verifyFwd(not_flushed_stat_macs, not_flushed_ports,
[self.vlan10_dyn_macs[0], self.vlan10_dyn_macs[1]],
[self.dev_port0, self.dev_port1],
self.vlan10_lag_ports)
# verify VLAN 20 static MACs
self._verifyFwd(self.vlan20_stat_macs, self.vlan20_ports,
[self.vlan20_stat_macs[0], self.vlan20_stat_macs[1]],
[self.dev_port2, self.dev_port3],
self.vlan20_lag_ports)
# verify VLAN 10 dynamic MACs
self._verifyFwd(not_flushed_dyn_macs, not_flushed_ports,
[self.vlan10_stat_macs[0], self.vlan10_stat_macs[1]],
[self.dev_port0, self.dev_port1],
self.vlan10_lag_ports)
# verify VLAN 20 dynamic MACs
self._verifyFwd(self.vlan20_dyn_macs, self.vlan20_ports,
[self.vlan20_stat_macs[0], self.vlan20_stat_macs[1]],
[self.dev_port2, self.dev_port3],
self.vlan20_lag_ports)
print("\tVerification complete")
def flushStaticPerLagTest(self):
'''
Verify flushing of static MAC entries by LAG
'''
print("\nflushStaticPerLagTest()")
self._prepareFdb()
flushed_lag_bp = self.lag1_bp
flushed_dev_ports = self.vlan10_lag_ports
flushed_macs = [self.vlan10_stat_macs[2], self.vlan10_stat_macs[3],
self.vlan10_stat_macs[4]]
print("Flush static MACs by LAG on LAG1")
sai_thrift_flush_fdb_entries(
self.client,
bridge_port_id=flushed_lag_bp,
entry_type=SAI_FDB_FLUSH_ENTRY_TYPE_STATIC)
print("Verify flooding on flushed LAG static MACs")
self._verifyFlood(flushed_macs, self.vlan10_ports,
[self.vlan10_dyn_macs[1]], [self.vlan10_ports[1]],
self.vlan10_lag_ports)
not_flushed_ports = [
self.vlan10_ports[i] for i in range(len(self.vlan10_ports))
if self.vlan10_ports[i] not in flushed_dev_ports]
not_flushed_macs = [
self.vlan10_stat_macs[i] for i in range(len(self.vlan10_stat_macs))
if self.vlan10_stat_macs[i] not in flushed_macs]
print("Verify forwarding for all other FDB entires")
# verify VLAN 10 static MACs
self._verifyFwd(not_flushed_macs, not_flushed_ports,
[self.vlan10_dyn_macs[0], self.vlan10_dyn_macs[1]],
[self.dev_port0, self.dev_port1],
self.vlan10_lag_ports)
# verify VLAN 20 static MACs
self._verifyFwd(self.vlan20_stat_macs, self.vlan20_ports,
[self.vlan20_stat_macs[0], self.vlan20_stat_macs[1]],
[self.dev_port2, self.dev_port3],
self.vlan20_lag_ports)
# verify VLAN 10 dynamic MACs
self._verifyFwd(self.vlan10_dyn_macs, self.vlan10_ports,
[self.vlan10_stat_macs[0], self.vlan10_stat_macs[1]],
[self.dev_port0, self.dev_port1],
self.vlan10_lag_ports)
# verify VLAN 20 dynamic MACs
self._verifyFwd(self.vlan20_dyn_macs, self.vlan20_ports,
[self.vlan20_stat_macs[0], self.vlan20_stat_macs[1]],
[self.dev_port2, self.dev_port3],
self.vlan20_lag_ports)
print("\tVerification complete")
def flushDynamicPerLagTest(self):
'''
Verify flushing of dynamic MAC entries by LAG
'''
print("\nflushDynamicPerLagTest()")
self._prepareFdb()
flushed_lag_bp = self.lag1_bp
flushed_dev_ports = self.vlan10_lag_ports
flushed_macs = [self.vlan10_dyn_macs[2], self.vlan10_dyn_macs[3],
self.vlan10_dyn_macs[4]]
print("Flush dynamic MACs by LAG on LAG1")
sai_thrift_flush_fdb_entries(
self.client,
bridge_port_id=flushed_lag_bp,
entry_type=SAI_FDB_FLUSH_ENTRY_TYPE_DYNAMIC)
print("Verify flooding on flushed LAG dynamic MACs")
self._verifyFlood(flushed_macs, self.vlan10_ports,
[self.vlan10_stat_macs[1]], [self.vlan10_ports[1]],
self.vlan10_lag_ports)
not_flushed_ports = [
self.vlan10_ports[i] for i in range(len(self.vlan10_ports))
if self.vlan10_ports[i] not in flushed_dev_ports]
not_flushed_macs = [
self.vlan10_dyn_macs[i] for i in range(len(self.vlan10_dyn_macs))
if self.vlan10_dyn_macs[i] not in flushed_macs]
print("Verify forwarding for all other FDB entires")
# verify VLAN 10 static MACs
self._verifyFwd(self.vlan10_stat_macs, self.vlan10_ports,
[self.vlan10_stat_macs[0], self.vlan10_stat_macs[1]],
[self.dev_port0, self.dev_port1],
self.vlan10_lag_ports)
# verify VLAN 20 static MACs
self._verifyFwd(self.vlan20_stat_macs, self.vlan20_ports,
[self.vlan20_stat_macs[0], self.vlan20_stat_macs[1]],
[self.dev_port2, self.dev_port3],
self.vlan20_lag_ports)
# verify VLAN 10 dynamic MACs
self._verifyFwd(not_flushed_macs, not_flushed_ports,
[self.vlan10_stat_macs[0], self.vlan10_stat_macs[1]],
[self.dev_port0, self.dev_port1],
self.vlan10_lag_ports)
# verify VLAN 20 dynamic MACs
self._verifyFwd(self.vlan20_dyn_macs, self.vlan20_ports,
[self.vlan20_stat_macs[0], self.vlan20_stat_macs[1]],
[self.dev_port2, self.dev_port3],
self.vlan20_lag_ports)
print("\tVerification complete")
def flushAllPerLagTest(self):
'''
Verify flushing of all kinds of MAC entries by LAG
'''
print("\nflushAllPerLagTest()")
self._prepareFdb()
flushed_lag_bp = self.lag1_bp
flushed_dev_ports = self.vlan10_lag_ports
flushed_stat_macs = [self.vlan10_stat_macs[2], self.vlan10_stat_macs[3],
self.vlan10_stat_macs[4]]
flushed_dyn_macs = [self.vlan10_dyn_macs[2], self.vlan10_dyn_macs[3],
self.vlan10_dyn_macs[4]]
print("Flush all MACs by LAG on LAG1")
sai_thrift_flush_fdb_entries(
self.client,
bridge_port_id=flushed_lag_bp,
entry_type=SAI_FDB_FLUSH_ENTRY_TYPE_DYNAMIC)
not_flushed_ports = [
self.vlan10_ports[i] for i in range(len(self.vlan10_ports))
if self.vlan10_ports[i] not in flushed_dev_ports]
not_flushed_stat_macs = [
self.vlan10_stat_macs[i] for i in range(len(self.vlan10_stat_macs))
if self.vlan10_stat_macs[i] not in flushed_stat_macs]
not_flushed_dyn_macs = [
self.vlan10_dyn_macs[i] for i in range(len(self.vlan10_dyn_macs))
if self.vlan10_dyn_macs[i] not in flushed_dyn_macs]
print("Verify forwarding for all other FDB entires")
# verify VLAN 10 static MACs
self._verifyFwd(not_flushed_stat_macs, not_flushed_ports,
[self.vlan10_dyn_macs[0], self.vlan10_dyn_macs[1]],
[self.dev_port0, self.dev_port1],
self.vlan10_lag_ports)
# verify VLAN 20 static MACs
self._verifyFwd(self.vlan20_stat_macs, self.vlan20_ports,
[self.vlan20_stat_macs[0], self.vlan20_stat_macs[1]],
[self.dev_port2, self.dev_port3],
self.vlan20_lag_ports)
# verify VLAN 10 dynamic MACs
self._verifyFwd(not_flushed_dyn_macs, not_flushed_ports,
[self.vlan10_stat_macs[0], self.vlan10_stat_macs[1]],
[self.dev_port0, self.dev_port1],
self.vlan10_lag_ports)
# verify VLAN 20 dynamic MACs
self._verifyFwd(self.vlan20_dyn_macs, self.vlan20_ports,
[self.vlan20_stat_macs[0], self.vlan20_stat_macs[1]],
[self.dev_port2, self.dev_port3],
self.vlan20_lag_ports)
print("\tVerification complete")
def flushStaticPerVlanAndPortTest(self):
'''
Verify flushing of static MAC entries by VLAN and port
'''
print("\nflushStaticPerVlanAndPortTest()")
self._prepareFdb()
chck_vlan10_port = self.dev_port0
chck_vlan10_mac = self.vlan10_stat_macs[0]
try:
self._setUpTrunkPort()
print("Flush static MAC by VLAN and port on trunk port")
sai_thrift_flush_fdb_entries(
self.client,
bv_id=self.vlan10,
bridge_port_id=self.trunk_port_bp,
entry_type=SAI_FDB_FLUSH_ENTRY_TYPE_STATIC)
print("Verify flooding on flushed MAC")
self._verifyFlood([self.tp10_stat_mac], self.vlan10_ports,
[chck_vlan10_mac], [chck_vlan10_port],
self.vlan10_lag_ports, self.trunk_dev_port,
self.vlan10_id)
print("Verify forwarding for all other FDB entries")
self.vlan10_ports.append(self.trunk_dev_port)
self.vlan20_ports.append(self.trunk_dev_port)
self.vlan20_stat_macs.append(self.tp20_stat_mac)
self.vlan10_dyn_macs.append(self.tp10_dyn_mac)
self.vlan20_dyn_macs.append(self.tp20_dyn_mac)
# verify VLAN 10 static MACs
self._verifyFwd(
self.vlan10_stat_macs, self.vlan10_ports,
[self.vlan10_stat_macs[0], self.vlan10_stat_macs[1]],
[self.dev_port0, self.dev_port1], self.vlan10_lag_ports,
self.trunk_dev_port, self.vlan10_id)
# verify VLAN 20 static MACs
self._verifyFwd(
self.vlan20_stat_macs, self.vlan20_ports,
[self.vlan20_stat_macs[0], self.vlan20_stat_macs[1]],
[self.dev_port2, self.dev_port3], self.vlan20_lag_ports,
self.trunk_dev_port, self.vlan20_id)
# verify VLAN 10 dynamic MACs
self._verifyFwd(
self.vlan10_dyn_macs, self.vlan10_ports,
[self.vlan10_stat_macs[0], self.vlan10_stat_macs[1]],
[self.dev_port0, self.dev_port1], self.vlan10_lag_ports,
self.trunk_dev_port, self.vlan10_id)
# verify VLAN 20 dynamic MACs
self._verifyFwd(
self.vlan20_dyn_macs, self.vlan20_ports,
[self.vlan20_stat_macs[0], self.vlan20_stat_macs[1]],
[self.dev_port2, self.dev_port3], self.vlan20_lag_ports,
self.trunk_dev_port, self.vlan20_id)
print("\tVerification complete")
finally:
self.vlan10_ports.remove(self.trunk_dev_port)
self.vlan20_ports.remove(self.trunk_dev_port)
self.vlan20_stat_macs.remove(self.tp20_stat_mac)
self.vlan10_dyn_macs.remove(self.tp10_dyn_mac)
self.vlan20_dyn_macs.remove(self.tp20_dyn_mac)
self._tearDownTrunkPort()
def flushDynamicPerVlanAndPortTest(self):
'''
Verify flushing of dynamic MAC entries by VLAN and port
'''
print("\nflushDynamicPerVlanAndPortTest()")
self._prepareFdb()
chck_vlan10_port = self.dev_port0
chck_vlan10_mac = self.vlan10_stat_macs[0]
try:
self._setUpTrunkPort()
print("Flush dynamic MAC by VLAN and port on trunk port")
sai_thrift_flush_fdb_entries(
self.client,
bv_id=self.vlan10,
bridge_port_id=self.trunk_port_bp,
entry_type=SAI_FDB_FLUSH_ENTRY_TYPE_DYNAMIC)
print("Verify flooding on flushed MAC")
self._verifyFlood([self.tp10_dyn_mac], self.vlan10_ports,
[chck_vlan10_mac], [chck_vlan10_port],
self.vlan10_lag_ports, self.trunk_dev_port,
self.vlan10_id)
print("Verify forwarding for all other FDB entries")
self.vlan10_ports.append(self.trunk_dev_port)
self.vlan20_ports.append(self.trunk_dev_port)
self.vlan10_stat_macs.append(self.tp10_stat_mac)
self.vlan20_stat_macs.append(self.tp20_stat_mac)
self.vlan20_dyn_macs.append(self.tp20_dyn_mac)
# verify VLAN 10 static MACs
self._verifyFwd(
self.vlan10_stat_macs, self.vlan10_ports,
[self.vlan10_stat_macs[0], self.vlan10_stat_macs[1]],
[self.dev_port0, self.dev_port1], self.vlan10_lag_ports,
self.trunk_dev_port, self.vlan10_id)
# verify VLAN 20 static MACs
self._verifyFwd(
self.vlan20_stat_macs, self.vlan20_ports,
[self.vlan20_stat_macs[0], self.vlan20_stat_macs[1]],
[self.dev_port2, self.dev_port3], self.vlan20_lag_ports,
self.trunk_dev_port, self.vlan20_id)
# verify VLAN 10 dynamic MACs
self._verifyFwd(
self.vlan10_dyn_macs, self.vlan10_ports,
[self.vlan10_stat_macs[0], self.vlan10_stat_macs[1]],
[self.dev_port0, self.dev_port1], self.vlan10_lag_ports,
self.trunk_dev_port, self.vlan10_id)
# verify VLAN 20 dynamic MACs
self._verifyFwd(
self.vlan20_dyn_macs, self.vlan20_ports,
[self.vlan20_stat_macs[0], self.vlan20_stat_macs[1]],
[self.dev_port2, self.dev_port3], self.vlan20_lag_ports,
self.trunk_dev_port, self.vlan20_id)
print("\tVerification complete")
finally:
self.vlan10_ports.remove(self.trunk_dev_port)
self.vlan20_ports.remove(self.trunk_dev_port)
self.vlan10_stat_macs.remove(self.tp10_stat_mac)
self.vlan20_stat_macs.remove(self.tp20_stat_mac)
self.vlan20_dyn_macs.remove(self.tp20_dyn_mac)
self._tearDownTrunkPort()
def flushAllPerVlanAndPortTest(self):
'''
Verify flushing of all kinds of MAC entries by VLAN and port
'''
print("\nflushAllPerVlanAndPortTest()")
self._prepareFdb()
chck_vlan10_port = self.dev_port0
chck_vlan10_mac = self.vlan10_stat_macs[0]
try:
self._setUpTrunkPort()
print("Flush all MACs by VLAN and port on trunk port")
sai_thrift_flush_fdb_entries(
self.client,
bv_id=self.vlan10,
bridge_port_id=self.trunk_port_bp,
entry_type=SAI_FDB_FLUSH_ENTRY_TYPE_ALL)
print("Verify flooding on flushed MAC")
self._verifyFlood([self.tp10_stat_mac], self.vlan10_ports,
[chck_vlan10_mac], [chck_vlan10_port],
self.vlan10_lag_ports, self.trunk_dev_port,
self.vlan10_id)
self._verifyFlood([self.tp10_dyn_mac], self.vlan10_ports,
[chck_vlan10_mac], [chck_vlan10_port],
self.vlan10_lag_ports, self.trunk_dev_port,
self.vlan10_id)
print("Verify forwarding for all other FDB entries")
self.vlan10_ports.append(self.trunk_dev_port)
self.vlan20_ports.append(self.trunk_dev_port)
self.vlan20_stat_macs.append(self.tp20_stat_mac)
self.vlan20_dyn_macs.append(self.tp20_dyn_mac)
# verify VLAN 10 static MACs
self._verifyFwd(
self.vlan10_stat_macs, self.vlan10_ports,
[self.vlan10_stat_macs[0], self.vlan10_stat_macs[1]],
[self.dev_port0, self.dev_port1], self.vlan10_lag_ports,
self.trunk_dev_port, self.vlan10_id)
# verify VLAN 20 static MACs
self._verifyFwd(
self.vlan20_stat_macs, self.vlan20_ports,
[self.vlan20_stat_macs[0], self.vlan20_stat_macs[1]],
[self.dev_port2, self.dev_port3], self.vlan20_lag_ports,
self.trunk_dev_port, self.vlan20_id)
# verify VLAN 10 dynamic MACs
self._verifyFwd(
self.vlan10_dyn_macs, self.vlan10_ports,
[self.vlan10_stat_macs[0], self.vlan10_stat_macs[1]],
[self.dev_port0, self.dev_port1], self.vlan10_lag_ports,
self.trunk_dev_port, self.vlan10_id)
# verify VLAN 20 dynamic MACs
self._verifyFwd(
self.vlan20_dyn_macs, self.vlan20_ports,
[self.vlan20_stat_macs[0], self.vlan20_stat_macs[1]],
[self.dev_port2, self.dev_port3], self.vlan20_lag_ports,
self.trunk_dev_port, self.vlan20_id)
print("\tVerification complete")
finally:
self.vlan10_ports.remove(self.trunk_dev_port)
self.vlan20_ports.remove(self.trunk_dev_port)
self.vlan20_stat_macs.remove(self.tp20_stat_mac)
self.vlan20_dyn_macs.remove(self.tp20_dyn_mac)
self._tearDownTrunkPort()
def flushAllStaticTest(self):
'''
Verify flushing of all static MAC entries
'''
print("\nflushAllStaticTest()")
self._prepareFdb()
print("Flush all static MACs")
sai_thrift_flush_fdb_entries(
self.client, entry_type=SAI_FDB_FLUSH_ENTRY_TYPE_STATIC)
print("Verify flooding on VLAN 10 and VLAN 20 static FDB entries")
self._verifyFlood(self.vlan10_stat_macs, self.vlan10_ports,
[self.vlan10_dyn_macs[0], self.vlan10_dyn_macs[1]],
[self.dev_port0, self.dev_port1],
self.vlan10_lag_ports)
self._verifyFlood(self.vlan20_stat_macs, self.vlan20_ports,
[self.vlan20_dyn_macs[0], self.vlan20_dyn_macs[1]],
[self.dev_port2, self.dev_port3],
self.vlan20_lag_ports)
print("Verify forwarding for all other FDB entires")
# verify VLAN 10 dynamic MACs
self._verifyFwd(self.vlan10_dyn_macs, self.vlan10_ports,
[self.vlan10_stat_macs[0], self.vlan10_stat_macs[1]],
[self.dev_port0, self.dev_port1],
self.vlan10_lag_ports)
# verify VLAN 20 dynamic MACs
self._verifyFwd(self.vlan20_dyn_macs, self.vlan20_ports,
[self.vlan20_stat_macs[0], self.vlan20_stat_macs[1]],
[self.dev_port2, self.dev_port3],
self.vlan20_lag_ports)
print("\tVerification complete")
def flushAllDynamicTest(self):
'''
Verify flushing of all dynamic MAC entries
'''
print("\nflushAllDynamicTest()")
self._prepareFdb()
print("Flush all dynamic MACs")
sai_thrift_flush_fdb_entries(
self.client, entry_type=SAI_FDB_FLUSH_ENTRY_TYPE_DYNAMIC)
print("Verify flooding on VLAN 10 and VLAN 20 dynamic FDB entries")
self._verifyFlood(self.vlan10_dyn_macs, self.vlan10_ports,
[self.vlan10_stat_macs[0], self.vlan10_stat_macs[1]],
[self.dev_port0, self.dev_port1],
self.vlan10_lag_ports)
self._verifyFlood(self.vlan20_dyn_macs, self.vlan20_ports,
[self.vlan20_stat_macs[0], self.vlan20_stat_macs[1]],
[self.dev_port2, self.dev_port3],
self.vlan20_lag_ports)
print("Verify forwarding for all other FDB entires")
# verify VLAN 10 static MACs
self._verifyFwd(self.vlan10_stat_macs, self.vlan10_ports,
[self.vlan10_dyn_macs[0], self.vlan10_dyn_macs[1]],
[self.dev_port0, self.dev_port1],
self.vlan10_lag_ports)
# verify VLAN 20 static MACs
self._verifyFwd(self.vlan20_stat_macs, self.vlan20_ports,
[self.vlan20_dyn_macs[0], self.vlan20_dyn_macs[1]],
[self.dev_port2, self.dev_port3],
self.vlan20_lag_ports)
print("\tVerification complete")
def flushAllMacsTest(self):
'''
Verify flushing of all kinds of MAC entries
'''
print("\nflushAllMacsTest()")
self._prepareFdb()
chck_vlan10_mac1 = "00:10:aa:11:11:11"
chck_vlan10_mac2 = "00:10:aa:22:22:22"
chck_vlan20_mac1 = "00:20:aa:11:11:11"
chck_vlan20_mac2 = "00:20:aa:22:22:22"
print("Flush all MACs in FDB")
sai_thrift_flush_fdb_entries(self.client,
entry_type=SAI_FDB_FLUSH_ENTRY_TYPE_ALL)
print("Verify flooding for all flushed FDB entires")
self._verifyFlood(self.vlan10_stat_macs, self.vlan10_ports,
[chck_vlan10_mac1, chck_vlan10_mac2],
[self.dev_port0, self.dev_port1],
self.vlan10_lag_ports)
self._verifyFlood(self.vlan10_dyn_macs, self.vlan10_ports,
[chck_vlan10_mac1, chck_vlan10_mac2],
[self.dev_port0, self.dev_port1],
self.vlan10_lag_ports)
self._verifyFlood(self.vlan20_stat_macs, self.vlan20_ports,
[chck_vlan20_mac1, chck_vlan20_mac2],
[self.dev_port2, self.dev_port3],
self.vlan20_lag_ports)
self._verifyFlood(self.vlan20_dyn_macs, self.vlan20_ports,
[chck_vlan20_mac1, chck_vlan20_mac2],
[self.dev_port2, self.dev_port3],
self.vlan20_lag_ports)
print("\tVerification complete")
@group("draft")
class FdbAgeTest(SaiHelper):
'''
Verify FDB entries aging
'''
def setUp(self):
super(FdbAgeTest, self).setUp()
# age time used in tests (in sec)
self.age_time = 10
status = sai_thrift_set_switch_attribute(self.client,
fdb_aging_time=self.age_time)
self.assertEqual(status, SAI_STATUS_SUCCESS)
print("Aging time set to %d sec for all tests" % self.age_time)
sw_attr = sai_thrift_get_switch_attribute(self.client,
fdb_aging_time=True)
self.assertEqual(sw_attr["fdb_aging_time"], self.age_time)
self.vlan_id = 10
# add one more port to VLAN 10
self.port24_bp = sai_thrift_create_bridge_port(
self.client,
bridge_id=self.default_1q_bridge,
port_id=self.port24,
type=SAI_BRIDGE_PORT_TYPE_PORT,
admin_state=True)
self.assertNotEqual(self.port24_bp, 0)
self.vlan10_member3 = sai_thrift_create_vlan_member(
self.client,
vlan_id=self.vlan10,
bridge_port_id=self.port24_bp,
vlan_tagging_mode=SAI_VLAN_TAGGING_MODE_UNTAGGED)
sai_thrift_set_port_attribute(self.client,
self.port24,
port_vlan_id=10)
# add static MAC address for port0 for verification purposes
self.vrf_mac = "00:12:34:56:78:90"
vrf_port_bp = self.port24_bp
self.vrf_port_dev = self.dev_port24
self.fdb_entry = sai_thrift_fdb_entry_t(switch_id=self.switch_id,
mac_address=self.vrf_mac,
bv_id=self.vlan10)
sai_thrift_create_fdb_entry(self.client,
self.fdb_entry,
type=SAI_FDB_ENTRY_TYPE_STATIC,
bridge_port_id=vrf_port_bp)
def runTest(self):
self.macAgingOnPortTest()
self.macAgingOnLagTest()
self.macAgingAfterMoveTest()
self.macMoveAfterAgingTest()
def tearDown(self):
# remove static MAC from FDB
sai_thrift_remove_fdb_entry(self.client, self.fdb_entry)
# remove additional port from VLAN 10
sai_thrift_set_port_attribute(self.client, self.port24, port_vlan_id=0)
sai_thrift_remove_vlan_member(self.client, self.vlan10_member3)
sai_thrift_remove_bridge_port(self.client, self.port24_bp)
# disable aging
status = sai_thrift_set_switch_attribute(self.client, fdb_aging_time=0)
self.assertEqual(status, SAI_STATUS_SUCCESS)
sw_attr = sai_thrift_get_switch_attribute(self.client,
fdb_aging_time=True)
self.assertEqual(sw_attr["fdb_aging_time"], 0)
super(FdbAgeTest, self).tearDown()
def macAgingOnPortTest(self):
'''
FDB aging test verifying if dynamic FDB entry associated with port
is removed after the aging interval.
'''
print("\nmacAgingOnPortTest()")
try:
# learn MAC adress on port1
lrn_mac = "00:01:01:01:01:01"
lrn_port = self.dev_port1
pkt = simple_udp_packet(eth_dst=self.vrf_mac,
eth_src=lrn_mac,
pktlen=100)
tag_pkt = simple_udp_packet(eth_dst=self.vrf_mac,
eth_src=lrn_mac,
dl_vlan_enable=True,
vlan_vid=self.vlan_id,
pktlen=104)
print("Sending packet on port %d, %s -> %s to learn MAC address" %
(lrn_port, lrn_mac, self.vrf_mac))
send_packet(self, lrn_port, tag_pkt)
verify_packets(self, pkt, [self.vrf_port_dev])
time.sleep(2)
print("Verifying if MAC address was learned")
pkt = simple_udp_packet(eth_dst=lrn_mac,
eth_src=self.vrf_mac,
pktlen=100)
tag_pkt = simple_udp_packet(eth_dst=lrn_mac,
eth_src=self.vrf_mac,
dl_vlan_enable=True,
vlan_vid=self.vlan_id,
pktlen=104)
print("Sending packet on port %d, %s -> %s - forwarding" %
(self.vrf_port_dev, self.vrf_mac, lrn_mac))
send_packet(self, self.vrf_port_dev, pkt)
verify_packets(self, tag_pkt, [lrn_port])
print("\tOK")
self.saiWaitFdbAge(self.age_time)
print("Verify if aged MAC address was removed")
flood_port_list = [[self.dev_port0], [self.dev_port1],
[self.dev_port4, self.dev_port5,
self.dev_port6]]
flood_pkt_list = [pkt, tag_pkt, pkt]
print("Sending packet on port %d, %s -> %s - will flood" %
(self.vrf_port_dev, self.vrf_mac, lrn_mac))
send_packet(self, self.vrf_port_dev, pkt)
verify_each_packet_on_multiple_port_lists(self, flood_pkt_list,
flood_port_list)
print("\tVerification complete")
finally:
sai_thrift_flush_fdb_entries(
self.client, entry_type=SAI_FDB_FLUSH_ENTRY_TYPE_DYNAMIC)
def macAgingOnLagTest(self):
'''
FDB aging test verifying if dynamic FDB entry associated with LAG
is removed after the aging interval
'''
print("\nmacAgingOnLagTest()")
try:
# learn MAC adress on LAG1
lrn_mac = "00:01:01:01:01:01"
lrn_port = self.dev_port5
lag_ports = [self.dev_port4, self.dev_port5, self.dev_port6]
pkt = simple_udp_packet(eth_dst=self.vrf_mac,
eth_src=lrn_mac,
pktlen=100)
print("Sending packet on port %d, %s -> %s to learn MAC address "
"on LAG" % (lrn_port, lrn_mac, self.vrf_mac))
send_packet(self, lrn_port, pkt)
verify_packets(self, pkt, [self.vrf_port_dev])
time.sleep(2)
print("Verifying if MAC address was learned")
pkt = simple_udp_packet(eth_dst=lrn_mac,
eth_src=self.vrf_mac,
pktlen=100)
tag_pkt = simple_udp_packet(eth_dst=lrn_mac,
eth_src=self.vrf_mac,
dl_vlan_enable=True,
vlan_vid=self.vlan_id,
pktlen=104)
print("Sending packet on port %d, %s -> %s - forwarding" %
(self.vrf_port_dev, self.vrf_mac, lrn_mac))
send_packet(self, self.vrf_port_dev, pkt)
verify_packet_any_port(self, pkt, lag_ports)
print("\tOK")
self.saiWaitFdbAge(self.age_time)
print("Verify if aged MAC address was removed")
flood_port_list = [[self.dev_port0], [self.dev_port1],
[self.dev_port4, self.dev_port5,
self.dev_port6]]
flood_pkt_list = [pkt, tag_pkt, pkt]
print("Sending packet on port %d, %s -> %s - will flood" %
(self.vrf_port_dev, self.vrf_mac, lrn_mac))
send_packet(self, self.vrf_port_dev, pkt)
verify_each_packet_on_multiple_port_lists(self, flood_pkt_list,
flood_port_list)
print("\tVerification complete")
finally:
sai_thrift_flush_fdb_entries(
self.client, entry_type=SAI_FDB_FLUSH_ENTRY_TYPE_DYNAMIC)
def macAgingAfterMoveTest(self):
'''
FDB aging test verifying if dynamic FDB entry associated with one port
and then moved to another port is removed after the aging interval
counted starting from the moment of moving the address (not the initial
learning time)
'''
print("\nmacAgingAfterMoveTest()")
try:
age_time = 25
status = sai_thrift_set_switch_attribute(self.client,
fdb_aging_time=age_time)
self.assertEqual(status, SAI_STATUS_SUCCESS)
print("Aging time set to %d sec for macAgingAfterMoveTest()" %
age_time)
sw_attr = sai_thrift_get_switch_attribute(self.client,
fdb_aging_time=True)
self.assertEqual(sw_attr["fdb_aging_time"], age_time)
# learn MAC adress on port1
lrn_mac = "00:01:01:01:01:01"
lrn_port = self.dev_port1
mv_port = self.dev_port0
lrn_pkt = simple_udp_packet(eth_dst=self.vrf_mac,
eth_src=lrn_mac,
pktlen=100)
lrn_tag_pkt = simple_udp_packet(eth_dst=self.vrf_mac,
eth_src=lrn_mac,
dl_vlan_enable=True,
vlan_vid=self.vlan_id,
pktlen=104)
print("Sending packet on port %d, %s -> %s to learn MAC address" %
(lrn_port, lrn_mac, self.vrf_mac))
send_packet(self, lrn_port, lrn_tag_pkt)
verify_packets(self, lrn_pkt, [self.vrf_port_dev])
time.sleep(2)
timer_start = time.time()
print("Verifying if MAC address was learned")
pkt = simple_udp_packet(eth_dst=lrn_mac,
eth_src=self.vrf_mac,
pktlen=100)
tag_pkt = simple_udp_packet(eth_dst=lrn_mac,
eth_src=self.vrf_mac,
dl_vlan_enable=True,
vlan_vid=self.vlan_id,
pktlen=104)
print("Sending packet on port %d, %s -> %s - forwarding" %
(self.vrf_port_dev, self.vrf_mac, lrn_mac))
send_packet(self, self.vrf_port_dev, pkt)
verify_packets(self, tag_pkt, [lrn_port])
print("\tOK")
wait_time = 15
old_age_out = age_time - wait_time
print("Waiting for %d seconds before moving MAC" % wait_time)
while (time.time() - timer_start) < wait_time:
time.sleep(1)
print("Moving learned MAC address to port %d" % mv_port)
print("Sending packet on port %d, %s -> %s to move MAC address" %
(mv_port, lrn_mac, self.vrf_mac))
send_packet(self, mv_port, lrn_pkt)
verify_packets(self, lrn_pkt, [self.vrf_port_dev])
time.sleep(1)
timer_start = time.time()
print("Verifying if MAC address was moved")
print("Sending packet on port %d, %s -> %s - forwarding" %
(self.vrf_port_dev, self.vrf_mac, lrn_mac))
send_packet(self, self.vrf_port_dev, pkt)
verify_packets(self, pkt, [mv_port])
print("\tOK")
old_learn_timeout = old_age_out - (time.time() - timer_start)
print("Age Time remaining from initial mac learning %d seconds" %
old_learn_timeout)
print("Verifying if MAC address was not removed after age_time "
"from initial learning")
self.saiWaitFdbAge(old_learn_timeout)
print("Sending packet on port %d, %s -> %s - forwarding" %
(self.vrf_port_dev, self.vrf_mac, lrn_mac))
send_packet(self, self.vrf_port_dev, pkt)
verify_packets(self, pkt, [mv_port])
print("\tOK")
new_learn_timeout = age_time - (time.time() - timer_start)
self.saiWaitFdbAge(new_learn_timeout)
print("Verify if aged MAC address was removed")
flood_port_list = [[self.dev_port0], [self.dev_port1],
[self.dev_port4, self.dev_port5,
self.dev_port6]]
flood_pkt_list = [pkt, tag_pkt, pkt]
print("Sending packet on port %d, %s -> %s - will flood" %
(self.vrf_port_dev, self.vrf_mac, lrn_mac))
send_packet(self, self.vrf_port_dev, pkt)
verify_each_packet_on_multiple_port_lists(self, flood_pkt_list,
flood_port_list)
print("\tVerification complete")
finally:
sai_thrift_flush_fdb_entries(
self.client, entry_type=SAI_FDB_FLUSH_ENTRY_TYPE_DYNAMIC)
sai_thrift_set_switch_attribute(self.client,
fdb_aging_time=self.age_time)
def macMoveAfterAgingTest(self):
'''
FDB aging test verifying if dynamic FDB entry associated with one port
and then aged and moved to another port is removed after the aging
interval counted starting from the moment of moving the address (not
the initial time)
'''
print("\nmacMoveAfterAgingTest()")
try:
# learn MAC adress on port1
lrn_mac = "00:01:01:01:01:01"
lrn_port = self.dev_port1
mv_port = self.dev_port0
lrn_pkt = simple_udp_packet(eth_dst=self.vrf_mac,
eth_src=lrn_mac,
pktlen=100)
lrn_tag_pkt = simple_udp_packet(eth_dst=self.vrf_mac,
eth_src=lrn_mac,
dl_vlan_enable=True,
vlan_vid=self.vlan_id,
pktlen=104)
print("Sending packet on port %d, %s -> %s to learn MAC address" %
(lrn_port, lrn_mac, self.vrf_mac))
send_packet(self, lrn_port, lrn_tag_pkt)
verify_packets(self, lrn_pkt, [self.vrf_port_dev])
time.sleep(2)
self.saiWaitFdbAge(self.age_time)
print("Moving learned MAC address to port %d" % mv_port)
print("Sending packet on port %d, %s -> %s to move MAC address" %
(mv_port, lrn_mac, self.vrf_mac))
send_packet(self, mv_port, lrn_pkt)
verify_packets(self, lrn_pkt, [self.vrf_port_dev])
time.sleep(1)
print("Verifying if MAC address was moved")
pkt = simple_udp_packet(eth_dst=lrn_mac,
eth_src=self.vrf_mac,
pktlen=100)
tag_pkt = simple_udp_packet(eth_dst=lrn_mac,
eth_src=self.vrf_mac,
dl_vlan_enable=True,
vlan_vid=self.vlan_id,
pktlen=104)
print("Sending packet on port %d, %s -> %s - forwarding" %
(self.vrf_port_dev, self.vrf_mac, lrn_mac))
send_packet(self, self.vrf_port_dev, pkt)
verify_packets(self, pkt, [mv_port])
print("\tOK")
self.saiWaitFdbAge(self.age_time)
print("Verify if aged MAC address was removed")
flood_port_list = [[self.dev_port0], [self.dev_port1],
[self.dev_port4, self.dev_port5,
self.dev_port6]]
flood_pkt_list = [pkt, tag_pkt, pkt]
print("Sending packet on port %d, %s -> %s - will flood" %
(self.vrf_port_dev, self.vrf_mac, lrn_mac))
send_packet(self, self.vrf_port_dev, pkt)
verify_each_packet_on_multiple_port_lists(self, flood_pkt_list,
flood_port_list)
print("\tVerification complete")
finally:
sai_thrift_flush_fdb_entries(
self.client, entry_type=SAI_FDB_FLUSH_ENTRY_TYPE_DYNAMIC)
@group("draft")
class FdbMissTest(SaiHelper):
'''
Verify actions after missing FDB entry
'''
def setUp(self):
super(FdbMissTest, self).setUp()
self.vlan_id = 100
# create VLAN 100 with 3 access ports
self.port24_bp = sai_thrift_create_bridge_port(
self.client,
bridge_id=self.default_1q_bridge,
port_id=self.port24,
type=SAI_BRIDGE_PORT_TYPE_PORT,
admin_state=True)
self.assertNotEqual(self.port24_bp, 0)
self.port25_bp = sai_thrift_create_bridge_port(
self.client,
bridge_id=self.default_1q_bridge,
port_id=self.port25,
type=SAI_BRIDGE_PORT_TYPE_PORT,
admin_state=True)
self.assertNotEqual(self.port25_bp, 0)
self.port26_bp = sai_thrift_create_bridge_port(
self.client,
bridge_id=self.default_1q_bridge,
port_id=self.port26,
type=SAI_BRIDGE_PORT_TYPE_PORT,
admin_state=True)
self.assertNotEqual(self.port26_bp, 0)
self.vlan100 = sai_thrift_create_vlan(self.client, vlan_id=100)
self.assertNotEqual(self.vlan100, 0)
self.vlan100_member0 = sai_thrift_create_vlan_member(
self.client,
vlan_id=self.vlan100,
bridge_port_id=self.port24_bp,
vlan_tagging_mode=SAI_VLAN_TAGGING_MODE_UNTAGGED)
self.vlan100_member1 = sai_thrift_create_vlan_member(
self.client,
vlan_id=self.vlan100,
bridge_port_id=self.port25_bp,
vlan_tagging_mode=SAI_VLAN_TAGGING_MODE_UNTAGGED)
self.vlan100_member2 = sai_thrift_create_vlan_member(
self.client,
vlan_id=self.vlan100,
bridge_port_id=self.port26_bp,
vlan_tagging_mode=SAI_VLAN_TAGGING_MODE_UNTAGGED)
sai_thrift_set_port_attribute(
self.client, self.port24, port_vlan_id=100)
sai_thrift_set_port_attribute(
self.client, self.port25, port_vlan_id=100)
sai_thrift_set_port_attribute(
self.client, self.port26, port_vlan_id=100)
# create a trap group
self.trap_group = sai_thrift_create_hostif_trap_group(
self.client, queue=4)
self.arp_req_trap = sai_thrift_create_hostif_trap(
self.client,
packet_action=SAI_PACKET_ACTION_TRAP,
trap_type=SAI_HOSTIF_TRAP_TYPE_ARP_REQUEST,
trap_group=self.trap_group)
self.lldp_trap = sai_thrift_create_hostif_trap(
self.client,
packet_action=SAI_PACKET_ACTION_TRAP,
trap_type=SAI_HOSTIF_TRAP_TYPE_LLDP,
trap_group=self.trap_group)
self.send_port = self.dev_port24
self.flood_ports = [self.dev_port25, self.dev_port26]
self.src_mac = "00:11:11:11:11:11"
self.dst_mac = "00:22:22:22:22:22"
self.mcast_mac = "01:00:5e:11:22:33"
self.bcast_mac = "ff:ff:ff:ff:ff:ff"
self.lldp_mac = "01:80:c2:00:00:0e"
self.ucast_pkt = simple_udp_packet(eth_dst=self.dst_mac,
eth_src=self.src_mac)
self.mcast_pkt = simple_udp_packet(eth_dst=self.mcast_mac,
eth_src=self.src_mac)
self.bcast_pkt = simple_udp_packet(eth_dst=self.bcast_mac,
eth_src=self.src_mac)
self.arp_pkt = simple_arp_packet(arp_op=1, pktlen=100)
self.lldp_pkt = simple_eth_packet(eth_dst=self.lldp_mac,
eth_src=self.src_mac,
pktlen=60,
eth_type=0x88cc)
def runTest(self):
self.unicastMissDropActionTest()
self.unicastMissCopyActionTest()
self.unicastMissTrapActionTest()
self.multicastMissDropActionTest()
self.multicastMissCopyActionTest()
self.multicastMissTrapActionTest()
self.broadcastMissDropActionTest()
self.broadcastMissCopyActionTest()
self.broadcastMissTrapActionTest()
def tearDown(self):
sai_thrift_flush_fdb_entries(
self.client, entry_type=SAI_FDB_FLUSH_ENTRY_TYPE_ALL)
# remove trap group
sai_thrift_remove_hostif_trap(self.client, self.lldp_trap)
sai_thrift_remove_hostif_trap(self.client, self.arp_req_trap)
sai_thrift_remove_hostif_trap_group(self.client, self.trap_group)
# remove VLAN 100
sai_thrift_set_port_attribute(self.client, self.port24, port_vlan_id=0)
sai_thrift_set_port_attribute(self.client, self.port25, port_vlan_id=0)
sai_thrift_set_port_attribute(self.client, self.port26, port_vlan_id=0)
sai_thrift_remove_vlan_member(self.client, self.vlan100_member0)
sai_thrift_remove_vlan_member(self.client, self.vlan100_member1)
sai_thrift_remove_vlan_member(self.client, self.vlan100_member2)
sai_thrift_remove_vlan(self.client, self.vlan100)
sai_thrift_remove_bridge_port(self.client, self.port24_bp)
sai_thrift_remove_bridge_port(self.client, self.port25_bp)
sai_thrift_remove_bridge_port(self.client, self.port26_bp)
super(FdbMissTest, self).tearDown()
def unicastMissDropActionTest(self):
'''
Verify if packets, which destination MAC is not stored in FDB,
are dropped after setting miss packet action to drop
'''
print("\nunicastMissDropActionTest()")
try:
print("Verifying initial switch behavior")
print("Sending packet on port %d, %s -> %s - will flood" %
(self.send_port, self.src_mac, self.dst_mac))
send_packet(self, self.send_port, self.ucast_pkt)
verify_packets(self, self.ucast_pkt, self.flood_ports)
print("\tOK")
print("Setting unicast FDB miss packet action to drop")
status = sai_thrift_set_switch_attribute(
self.client,
fdb_unicast_miss_packet_action=SAI_PACKET_ACTION_DROP)
self.assertEqual(status, SAI_STATUS_SUCCESS)
sw_attr = sai_thrift_get_switch_attribute(
self.client, fdb_unicast_miss_packet_action=True)
self.assertEqual(sw_attr["fdb_unicast_miss_packet_action"],
SAI_PACKET_ACTION_DROP)
print("Sending packet on port %d, %s -> %s - will be dropped" %
(self.send_port, self.src_mac, self.dst_mac))
send_packet(self, self.send_port, self.ucast_pkt)
verify_no_other_packets(self)
print("\tVerification complete")
finally:
# revert default packet acion
status = sai_thrift_set_switch_attribute(
self.client,
fdb_unicast_miss_packet_action=SAI_PACKET_ACTION_FORWARD)
self.assertEqual(status, SAI_STATUS_SUCCESS)
sw_attr = sai_thrift_get_switch_attribute(
self.client, fdb_unicast_miss_packet_action=True)
self.assertEqual(sw_attr["fdb_unicast_miss_packet_action"],
SAI_PACKET_ACTION_FORWARD)
def unicastMissCopyActionTest(self):
'''
Verify if packets, which destination MAC is not stored in FDB,
are copied to CPU after setting miss packet action to copy
'''
print("\nunicastMissCopyActionTest()")
try:
print("Verifying initial switch behavior")
print("Sending packet on port %d, %s -> %s - will flood" %
(self.send_port, self.src_mac, self.dst_mac))
send_packet(self, self.send_port, self.ucast_pkt)
verify_packets(self, self.ucast_pkt, self.flood_ports)
print("\tOK")
print("Setting unicast FDB miss packet action to copy to CPU")
status = sai_thrift_set_switch_attribute(
self.client,
fdb_unicast_miss_packet_action=SAI_PACKET_ACTION_COPY)
self.assertEqual(status, SAI_STATUS_SUCCESS)
sw_attr = sai_thrift_get_switch_attribute(
self.client, fdb_unicast_miss_packet_action=True)
self.assertEqual(sw_attr["fdb_unicast_miss_packet_action"],
SAI_PACKET_ACTION_COPY)
time.sleep(4)
pre_stats = query_counter(
self, sai_thrift_get_queue_stats, self.cpu_queue0)
print("Sending packet on port %d, %s -> %s - will be copied to "
"CPU" % (self.send_port, self.src_mac, self.dst_mac))
send_packet(self, self.send_port, self.ucast_pkt)
verify_packets(self, self.ucast_pkt, self.flood_ports)
time.sleep(4)
post_stats = query_counter(
self, sai_thrift_get_queue_stats, self.cpu_queue0)
self.assertEqual(
post_stats["SAI_QUEUE_STAT_PACKETS"] -
pre_stats["SAI_QUEUE_STAT_PACKETS"],
1)
self.dataplane.flush()
print("\tVerification complete")
finally:
# revert default packet acion
status = sai_thrift_set_switch_attribute(
self.client,
fdb_unicast_miss_packet_action=SAI_PACKET_ACTION_FORWARD)
self.assertEqual(status, SAI_STATUS_SUCCESS)
sw_attr = sai_thrift_get_switch_attribute(
self.client, fdb_unicast_miss_packet_action=True)
self.assertEqual(sw_attr["fdb_unicast_miss_packet_action"],
SAI_PACKET_ACTION_FORWARD)
def unicastMissTrapActionTest(self):
'''
Verify if packets, which destination MAC is not in FDB,
are redirected to CPU after setting action to trap
'''
print("\nunicastMissTrapActionTest()")
try:
print("Verifying initial switch behavior")
print("Sending packet on port %d, %s -> %s - will flood" %
(self.send_port, self.src_mac, self.dst_mac))
send_packet(self, self.send_port, self.ucast_pkt)
verify_packets(self, self.ucast_pkt, self.flood_ports)
print("\tOK")
print("Setting unicast FDB miss packet action to trap")
status = sai_thrift_set_switch_attribute(
self.client,
fdb_unicast_miss_packet_action=SAI_PACKET_ACTION_TRAP)
self.assertEqual(status, SAI_STATUS_SUCCESS)
sw_attr = sai_thrift_get_switch_attribute(
self.client, fdb_unicast_miss_packet_action=True)
self.assertEqual(sw_attr["fdb_unicast_miss_packet_action"],
SAI_PACKET_ACTION_TRAP)
time.sleep(4)
pre_stats = query_counter(
self, sai_thrift_get_queue_stats, self.cpu_queue0)
print("Sending packet on port %d, %s -> %s - will be redirected "
"to CPU" % (self.send_port, self.src_mac, self.dst_mac))
send_packet(self, self.send_port, self.ucast_pkt)
time.sleep(4)
post_stats = query_counter(
self, sai_thrift_get_queue_stats, self.cpu_queue0)
self.assertEqual(
post_stats["SAI_QUEUE_STAT_PACKETS"] -
pre_stats["SAI_QUEUE_STAT_PACKETS"],
1)
self.dataplane.flush()
print("\tVerification complete")
finally:
# revert default packet acion
status = sai_thrift_set_switch_attribute(
self.client,
fdb_unicast_miss_packet_action=SAI_PACKET_ACTION_FORWARD)
self.assertEqual(status, SAI_STATUS_SUCCESS)
sw_attr = sai_thrift_get_switch_attribute(
self.client, fdb_unicast_miss_packet_action=True)
self.assertEqual(sw_attr["fdb_unicast_miss_packet_action"],
SAI_PACKET_ACTION_FORWARD)
def multicastMissDropActionTest(self):
'''
Verify if multicast packets are dropped after setting miss packet
action to drop.
Verify also if LLDP packets are still redirected to CPU
'''
print("\nmulticastMissDropActionTest()")
try:
print("Verifying initial switch behavior")
print("Sending multicast packet on port %d, %s -> %s - will flood"
% (self.send_port, self.src_mac, self.mcast_mac))
send_packet(self, self.send_port, self.mcast_pkt)
verify_packets(self, self.mcast_pkt, self.flood_ports)
print("\tOK")
print("Setting multicast FDB miss packet action to drop")
status = sai_thrift_set_switch_attribute(
self.client,
fdb_multicast_miss_packet_action=SAI_PACKET_ACTION_DROP)
self.assertEqual(status, SAI_STATUS_SUCCESS)
sw_attr = sai_thrift_get_switch_attribute(
self.client, fdb_multicast_miss_packet_action=True)
self.assertEqual(sw_attr["fdb_multicast_miss_packet_action"],
SAI_PACKET_ACTION_DROP)
print("Sending multicast packet on port %d, %s -> %s - will be "
"dropped" % (self.send_port, self.src_mac, self.mcast_mac))
send_packet(self, self.send_port, self.mcast_pkt)
verify_no_other_packets(self)
print("\tOK")
time.sleep(4)
pre_stats = query_counter(
self, sai_thrift_get_queue_stats, self.cpu_queue4)
print("Checking if LLDP packes are still forwarded to CPU")
print("Sending multicast LLDP packet on port %d, %s -> %s - will "
"be redirected to CPU" % (self.send_port, self.src_mac,
self.lldp_mac))
send_packet(self, self.send_port, self.lldp_pkt)
time.sleep(4)
post_stats = query_counter(
self, sai_thrift_get_queue_stats, self.cpu_queue4)
self.assertEqual(
post_stats["SAI_QUEUE_STAT_PACKETS"] -
pre_stats["SAI_QUEUE_STAT_PACKETS"],
1)
print("\nVerificaion complete")
finally:
self.dataplane.flush()
# reveret default multicast packet action
status = sai_thrift_set_switch_attribute(
self.client,
fdb_multicast_miss_packet_action=SAI_PACKET_ACTION_FORWARD)
self.assertEqual(status, SAI_STATUS_SUCCESS)
sw_attr = sai_thrift_get_switch_attribute(
self.client, fdb_multicast_miss_packet_action=True)
self.assertEqual(sw_attr["fdb_multicast_miss_packet_action"],
SAI_PACKET_ACTION_FORWARD)
def multicastMissCopyActionTest(self):
'''
Verify if multicast packets are copied to CPU after setting miss packet action to copy.
Verify also if LLDP packets are still redirected to CPU
'''
print("\nmulticastMissCopyActionTest()")
try:
print("Verifying initial switch behavior")
print("Sending multicast packet on port %d, %s -> %s - will flood"
% (self.send_port, self.src_mac, self.mcast_mac))
send_packet(self, self.send_port, self.mcast_pkt)
verify_packets(self, self.mcast_pkt, self.flood_ports)
print("\tOK")
print("Setting multicast FDB miss packet action to copy to CPU")
status = sai_thrift_set_switch_attribute(
self.client,
fdb_multicast_miss_packet_action=SAI_PACKET_ACTION_COPY)
self.assertEqual(status, SAI_STATUS_SUCCESS)
sw_attr = sai_thrift_get_switch_attribute(
self.client, fdb_multicast_miss_packet_action=True)
self.assertEqual(sw_attr["fdb_multicast_miss_packet_action"],
SAI_PACKET_ACTION_COPY)
time.sleep(4)
pre_stats = query_counter(
self, sai_thrift_get_queue_stats, self.cpu_queue0)
print("Sending multicast packet on port %d, %s -> %s - will be "
"copied to CPU" % (self.send_port, self.src_mac,
self.mcast_mac))
send_packet(self, self.send_port, self.mcast_pkt)
verify_packets(self, self.mcast_pkt, self.flood_ports)
time.sleep(4)
post_stats = query_counter(
self, sai_thrift_get_queue_stats, self.cpu_queue0)
self.assertEqual(
post_stats["SAI_QUEUE_STAT_PACKETS"] -
pre_stats["SAI_QUEUE_STAT_PACKETS"],
1)
print("\tOK")
time.sleep(4)
pre_stats = query_counter(
self, sai_thrift_get_queue_stats, self.cpu_queue4)
print("Checking if LLDP packes are still forwarded to CPU")
print("Sending multicast LLDP packet on port %d, %s -> %s - will "
"be redirected to CPU" % (self.send_port, self.src_mac,
self.lldp_mac))
send_packet(self, self.send_port, self.lldp_pkt)
time.sleep(4)
post_stats = query_counter(
self, sai_thrift_get_queue_stats, self.cpu_queue4)
self.assertEqual(
post_stats["SAI_QUEUE_STAT_PACKETS"] -
pre_stats["SAI_QUEUE_STAT_PACKETS"],
1)
print("\nVerificaion complete")
finally:
self.dataplane.flush()
# reveret default multicast packet action
status = sai_thrift_set_switch_attribute(
self.client,
fdb_multicast_miss_packet_action=SAI_PACKET_ACTION_FORWARD)
self.assertEqual(status, SAI_STATUS_SUCCESS)
sw_attr = sai_thrift_get_switch_attribute(
self.client, fdb_multicast_miss_packet_action=True)
self.assertEqual(sw_attr["fdb_multicast_miss_packet_action"],
SAI_PACKET_ACTION_FORWARD)
def multicastMissTrapActionTest(self):
'''
Verify if multicast packets are redirected to CPU after setting miss
packet action to trap.
Verify also if LLDP packets are still redirected to CPU.
'''
print("\nmulticastMissTrapActionTest()")
try:
print("Verifying initial switch behavior")
print("Sending multicast packet on port %d, %s -> %s - will flood"
% (self.send_port, self.src_mac, self.mcast_mac))
send_packet(self, self.send_port, self.mcast_pkt)
verify_packets(self, self.mcast_pkt, self.flood_ports)
print("\tOK")
print("Setting multicast FDB miss packet action to trap")
status = sai_thrift_set_switch_attribute(
self.client,
fdb_multicast_miss_packet_action=SAI_PACKET_ACTION_TRAP)
self.assertEqual(status, SAI_STATUS_SUCCESS)
sw_attr = sai_thrift_get_switch_attribute(
self.client, fdb_multicast_miss_packet_action=True)
self.assertEqual(sw_attr["fdb_multicast_miss_packet_action"],
SAI_PACKET_ACTION_TRAP)
time.sleep(4)
pre_stats = query_counter(
self, sai_thrift_get_queue_stats, self.cpu_queue0)
print("Sending multicast packet on port %d, %s -> %s - will be "
"redirected to CPU" % (self.send_port, self.src_mac,
self.mcast_mac))
send_packet(self, self.send_port, self.mcast_pkt)
time.sleep(4)
post_stats = query_counter(
self, sai_thrift_get_queue_stats, self.cpu_queue0)
self.assertEqual(
post_stats["SAI_QUEUE_STAT_PACKETS"] -
pre_stats["SAI_QUEUE_STAT_PACKETS"],
1)
print("\tOK")
time.sleep(4)
pre_stats = query_counter(
self, sai_thrift_get_queue_stats, self.cpu_queue4)
print("Checking if LLDP packes are still forwarded to CPU")
print("Sending multicast LLDP packet on port %d, %s -> %s - will "
"be redirected to CPU" % (self.send_port, self.src_mac,
self.lldp_mac))
send_packet(self, self.send_port, self.lldp_pkt)
time.sleep(4)
post_stats = query_counter(
self, sai_thrift_get_queue_stats, self.cpu_queue4)
self.assertEqual(
post_stats["SAI_QUEUE_STAT_PACKETS"] -
pre_stats["SAI_QUEUE_STAT_PACKETS"],
1)
print("\nVerificaion complete")
finally:
self.dataplane.flush()
# reveret default multicast packet action
status = sai_thrift_set_switch_attribute(
self.client,
fdb_multicast_miss_packet_action=SAI_PACKET_ACTION_FORWARD)
self.assertEqual(status, SAI_STATUS_SUCCESS)
sw_attr = sai_thrift_get_switch_attribute(
self.client, fdb_multicast_miss_packet_action=True)
self.assertEqual(sw_attr["fdb_multicast_miss_packet_action"],
SAI_PACKET_ACTION_FORWARD)
def broadcastMissDropActionTest(self):
'''
Verify if broadcast packets are dropped after setting miss packet
action to drop.
Verify also if ARP packets are still redirected to CPU.
'''
print("\nbroadcastMissDropActionTest()")
try:
print("Verifying initial switch behavior")
print("Sending broadcast packet on port %d, %s -> %s - will flood"
% (self.send_port, self.src_mac, self.bcast_mac))
send_packet(self, self.send_port, self.bcast_pkt)
verify_packets(self, self.bcast_pkt, self.flood_ports)
print("\tOK")
print("Setting broadcast FDB miss packet action to drop")
status = sai_thrift_set_switch_attribute(
self.client,
fdb_broadcast_miss_packet_action=SAI_PACKET_ACTION_DROP)
self.assertEqual(status, SAI_STATUS_SUCCESS)
sw_attr = sai_thrift_get_switch_attribute(
self.client, fdb_broadcast_miss_packet_action=True)
self.assertEqual(sw_attr["fdb_broadcast_miss_packet_action"],
SAI_PACKET_ACTION_DROP)
print("Sending broadcast packet on port %d, %s -> %s - will be "
"dropped" % (self.send_port, self.src_mac, self.bcast_mac))
send_packet(self, self.send_port, self.bcast_pkt)
verify_no_other_packets(self)
print("\tOK")
time.sleep(4)
pre_stats = query_counter(
self, sai_thrift_get_queue_stats, self.cpu_queue4)
print("Checking if ARP packes are still forwarded to CPU")
print("Sending ARP packet on port %d - will be redirected to CPU" %
(self.send_port))
send_packet(self, self.send_port, self.arp_pkt)
time.sleep(4)
post_stats = query_counter(
self, sai_thrift_get_queue_stats, self.cpu_queue4)
self.assertEqual(
post_stats["SAI_QUEUE_STAT_PACKETS"] -
pre_stats["SAI_QUEUE_STAT_PACKETS"],
1)
print("\nVerificaion complete")
finally:
self.dataplane.flush()
# revert default broadcast packet action
status = sai_thrift_set_switch_attribute(
self.client,
fdb_broadcast_miss_packet_action=SAI_PACKET_ACTION_FORWARD)
self.assertEqual(status, SAI_STATUS_SUCCESS)
sw_attr = sai_thrift_get_switch_attribute(
self.client, fdb_broadcast_miss_packet_action=True)
self.assertEqual(sw_attr["fdb_broadcast_miss_packet_action"],
SAI_PACKET_ACTION_FORWARD)
def broadcastMissCopyActionTest(self):
'''
Verify if broadcast packets are copied to CPU after setting miss packet action to copy.
Verify also if ARP packets are still redirected to CPU.
'''
print("\nbroadcastMissCopyActionTest()")
try:
print("Verifying initial switch behavior")
print("Sending broadcast packet on port %d, %s -> %s - will flood"
% (self.send_port, self.src_mac, self.bcast_mac))
send_packet(self, self.send_port, self.bcast_pkt)
verify_packets(self, self.bcast_pkt, self.flood_ports)
print("\tOK")
print("Setting broadcast FDB miss packet action to copy")
status = sai_thrift_set_switch_attribute(
self.client,
fdb_broadcast_miss_packet_action=SAI_PACKET_ACTION_COPY)
self.assertEqual(status, SAI_STATUS_SUCCESS)
sw_attr = sai_thrift_get_switch_attribute(
self.client, fdb_broadcast_miss_packet_action=True)
self.assertEqual(sw_attr["fdb_broadcast_miss_packet_action"],
SAI_PACKET_ACTION_COPY)
time.sleep(4)
pre_stats = query_counter(
self, sai_thrift_get_queue_stats, self.cpu_queue0)
print("Sending broadcast packet on port %d, %s -> %s - will be "
"copied to CPU" % (self.send_port, self.src_mac,
self.bcast_mac))
send_packet(self, self.send_port, self.bcast_pkt)
verify_packets(self, self.bcast_pkt, self.flood_ports)
time.sleep(4)
post_stats = query_counter(
self, sai_thrift_get_queue_stats, self.cpu_queue0)
self.assertEqual(
post_stats["SAI_QUEUE_STAT_PACKETS"] -
pre_stats["SAI_QUEUE_STAT_PACKETS"],
1)
print("\tOK")
time.sleep(4)
pre_stats = query_counter(
self, sai_thrift_get_queue_stats, self.cpu_queue4)
print("Checking if ARP packes are still forwarded to CPU")
print("Sending ARP packet on port %d - will be redirected to CPU" %
(self.send_port))
send_packet(self, self.send_port, self.arp_pkt)
time.sleep(4)
post_stats = query_counter(
self, sai_thrift_get_queue_stats, self.cpu_queue4)
self.assertEqual(
post_stats["SAI_QUEUE_STAT_PACKETS"] -
pre_stats["SAI_QUEUE_STAT_PACKETS"],
1)
print("\nVerificaion complete")
finally:
self.dataplane.flush()
# reveret default multicast packet action
status = sai_thrift_set_switch_attribute(
self.client,
fdb_broadcast_miss_packet_action=SAI_PACKET_ACTION_FORWARD)
self.assertEqual(status, SAI_STATUS_SUCCESS)
sw_attr = sai_thrift_get_switch_attribute(
self.client, fdb_broadcast_miss_packet_action=True)
self.assertEqual(sw_attr["fdb_broadcast_miss_packet_action"],
SAI_PACKET_ACTION_FORWARD)
def broadcastMissTrapActionTest(self):
'''
Verify if broacast packets are redirected to CPU after setting miss
packet action to trap.
Verify also if ARP packets are still redirected to CPU.
'''
print("\nbroadcastMissTrapActionTest()")
try:
print("Verifying initial switch behavior")
print("Sending broadcast packet on port %d, %s -> %s - will flood"
% (self.send_port, self.src_mac, self.bcast_mac))
send_packet(self, self.send_port, self.bcast_pkt)
verify_packets(self, self.bcast_pkt, self.flood_ports)
print("\tOK")
print("Setting broadcast FDB miss packet action to trap")
status = sai_thrift_set_switch_attribute(
self.client,
fdb_broadcast_miss_packet_action=SAI_PACKET_ACTION_TRAP)
self.assertEqual(status, SAI_STATUS_SUCCESS)
sw_attr = sai_thrift_get_switch_attribute(
self.client, fdb_broadcast_miss_packet_action=True)
self.assertEqual(sw_attr["fdb_broadcast_miss_packet_action"],
SAI_PACKET_ACTION_TRAP)
time.sleep(4)
pre_stats = query_counter(
self, sai_thrift_get_queue_stats, self.cpu_queue0)
print("Sending broadcast packet on port %d, %s -> %s - will be "
"redirected to CPU" % (self.send_port, self.src_mac,
self.bcast_mac))
send_packet(self, self.send_port, self.bcast_pkt)
time.sleep(4)
post_stats = query_counter(
self, sai_thrift_get_queue_stats, self.cpu_queue0)
self.assertEqual(
post_stats["SAI_QUEUE_STAT_PACKETS"] -
pre_stats["SAI_QUEUE_STAT_PACKETS"],
1)
print("\tOK")
time.sleep(4)
pre_stats = query_counter(
self, sai_thrift_get_queue_stats, self.cpu_queue4)
print("Checking if ARP packes are still forwarded to CPU")
print("Sending ARP packet on port %d - will be redirected to CPU" %
(self.send_port))
send_packet(self, self.send_port, self.arp_pkt)
time.sleep(4)
post_stats = query_counter(
self, sai_thrift_get_queue_stats, self.cpu_queue4)
self.assertEqual(
post_stats["SAI_QUEUE_STAT_PACKETS"] -
pre_stats["SAI_QUEUE_STAT_PACKETS"],
1)
print("\nVerificaion complete")
finally:
self.dataplane.flush()
# revert default broadcast packet action
status = sai_thrift_set_switch_attribute(
self.client,
fdb_broadcast_miss_packet_action=SAI_PACKET_ACTION_FORWARD)
self.assertEqual(status, SAI_STATUS_SUCCESS)
sw_attr = sai_thrift_get_switch_attribute(
self.client, fdb_broadcast_miss_packet_action=True)
self.assertEqual(sw_attr["fdb_broadcast_miss_packet_action"],
SAI_PACKET_ACTION_FORWARD)
@group("draft")
class FdbEventTest(SaiHelper):
'''
Verify correctness of FDB atributes values after events like:
learning, aging, moving, flushing, deleting
'''
def setUp(self):
super(FdbEventTest, self).setUp()
self.src_mac = "00:11:11:11:11:11"
self.dst_mac = "00:22:22:22:22:22"
vlan_id = 10
self.pkt = simple_udp_packet(eth_dst=self.dst_mac,
eth_src=self.src_mac,
pktlen=100)
self.tag_pkt = simple_udp_packet(eth_dst=self.dst_mac,
eth_src=self.src_mac,
dl_vlan_enable=True,
vlan_vid=vlan_id,
pktlen=104)
self.src_port = self.dev_port0
self.flood_ports = [[self.dev_port1],
[self.dev_port4, self.dev_port5, self.dev_port6]]
self.flood_pkts = [self.tag_pkt, self.pkt]
# expected FDB entry
self.mac_entry = sai_thrift_fdb_entry_t(switch_id=self.switch_id,
mac_address=self.src_mac,
bv_id=self.vlan10)
def runTest(self):
self.macLearnEventTest()
self.macAgeEvenTest()
self.macMoveEventTest()
self.macFlushEventTest()
self.macDeleteEventTest()
def tearDown(self):
sai_thrift_flush_fdb_entries(
self.client, entry_type=SAI_FDB_FLUSH_ENTRY_TYPE_ALL)
super(FdbEventTest, self).tearDown()
def macLearnEventTest(self):
'''
Verify MAC learning event for FDB entry
'''
print("\nmacLearnEventTest()")
try:
print("Sending packet on port %d to learn MAC address, %s -> %s - "
"will flood" % (self.src_port, self.src_mac, self.dst_mac))
send_packet(self, self.src_port, self.pkt)
verify_each_packet_on_multiple_port_lists(
self, self.flood_pkts, self.flood_ports)
print("\tOK")
time.sleep(2)
print("Verifying FDB attributes")
fdb_attr = sai_thrift_get_fdb_entry_attribute(self.client,
self.mac_entry,
bridge_port_id=True,
packet_action=True,
type=True)
self.assertEqual(fdb_attr["bridge_port_id"], self.port0_bp)
print("\tbridge_port_id\tOK")
self.assertEqual(fdb_attr["packet_action"],
SAI_PACKET_ACTION_FORWARD)
print("\tpacket_action\tOK")
self.assertEqual(fdb_attr["type"], SAI_FDB_ENTRY_TYPE_DYNAMIC)
print("\ttype\t\tOK")
print("\tVerification complete")
finally:
sai_thrift_flush_fdb_entries(
self.client, entry_type=SAI_FDB_FLUSH_ENTRY_TYPE_ALL)
def macAgeEvenTest(self):
'''
Verify MAC aging event for FDB entry
'''
print("\nmacAgeEvenTest()")
age_time = 10
try:
status = sai_thrift_set_switch_attribute(self.client,
fdb_aging_time=age_time)
self.assertEqual(status, SAI_STATUS_SUCCESS)
print("Aging time set to %d sec for all tests" % age_time)
print("Sending packet on port %d to learn MAC address, %s -> %s - "
"will flood" % (self.src_port, self.src_mac, self.dst_mac))
send_packet(self, self.src_port, self.pkt)
verify_each_packet_on_multiple_port_lists(self, self.flood_pkts,
self.flood_ports)
print("\tOK")
time.sleep(2)
fdb_attr = sai_thrift_get_fdb_entry_attribute(self.client,
self.mac_entry,
bridge_port_id=True,
packet_action=True,
type=True)
self.assertEqual(fdb_attr["bridge_port_id"], self.port0_bp)
self.assertEqual(fdb_attr["packet_action"],
SAI_PACKET_ACTION_FORWARD)
self.assertEqual(fdb_attr["type"], SAI_FDB_ENTRY_TYPE_DYNAMIC)
print("Waiting until aging interval is gone")
self.saiWaitFdbAge(age_time)
print("Verifying FDB attributes")
fdb_attr = sai_thrift_get_fdb_entry_attribute(self.client,
self.mac_entry,
bridge_port_id=True,
packet_action=True,
type=True)
self.assertEqual(self.status(), SAI_STATUS_ITEM_NOT_FOUND)
print("\tEntry\t\tclear")
self.assertEqual(fdb_attr, None)
print("\tAttrs\t\tclear")
print("\tVerification complete")
finally:
sai_thrift_flush_fdb_entries(
self.client, entry_type=SAI_FDB_FLUSH_ENTRY_TYPE_ALL)
sai_thrift_set_switch_attribute(self.client, fdb_aging_time=0)
def macMoveEventTest(self):
'''
Verify MAC moving event for FDB entry
'''
print("\nmacMoveEventTest()")
mv_port = self.dev_port4
flood_ports = [[self.dev_port0], [self.dev_port1]]
flood_pkts = [self.pkt, self.tag_pkt]
try:
print("Sending packet on port %d to learn MAC address, %s -> %s - "
"will flood" % (self.src_port, self.src_mac, self.dst_mac))
send_packet(self, self.src_port, self.pkt)
verify_each_packet_on_multiple_port_lists(
self, self.flood_pkts, self.flood_ports)
print("\tOK")
time.sleep(2)
fdb_attr = sai_thrift_get_fdb_entry_attribute(self.client,
self.mac_entry,
bridge_port_id=True,
packet_action=True,
type=True)
self.assertEqual(fdb_attr["bridge_port_id"], self.port0_bp)
self.assertEqual(fdb_attr["packet_action"],
SAI_PACKET_ACTION_FORWARD)
self.assertEqual(fdb_attr["type"], SAI_FDB_ENTRY_TYPE_DYNAMIC)
print("Sending packet on port %d to move MAC address, %s -> %s - "
"will flood" % (mv_port, self.src_mac, self.dst_mac))
send_packet(self, mv_port, self.pkt)
verify_each_packet_on_multiple_port_lists(
self, flood_pkts, flood_ports)
print("\tOK")
time.sleep(2)
print("Verifying FDB attributes")
fdb_attr = sai_thrift_get_fdb_entry_attribute(self.client,
self.mac_entry,
bridge_port_id=True,
packet_action=True,
type=True)
self.assertEqual(fdb_attr["bridge_port_id"], self.lag1_bp)
print("\tbridge_port_id\tOK")
self.assertEqual(fdb_attr["packet_action"],
SAI_PACKET_ACTION_FORWARD)
print("\tpacket_action\tOK")
self.assertEqual(fdb_attr["type"], SAI_FDB_ENTRY_TYPE_DYNAMIC)
print("\ttype\t\tOK")
print("\tVerification complete")
finally:
sai_thrift_flush_fdb_entries(
self.client, entry_type=SAI_FDB_FLUSH_ENTRY_TYPE_ALL)
def macFlushEventTest(self):
'''
Verify MAC flushing event for FDB entry
'''
print("\nmacFlushEventTest()")
try:
print("Sending packet on port %d to learn MAC address, %s -> %s - "
"will flood" % (self.src_port, self.src_mac, self.dst_mac))
send_packet(self, self.src_port, self.pkt)
verify_each_packet_on_multiple_port_lists(
self, self.flood_pkts, self.flood_ports)
print("\tOK")
time.sleep(2)
fdb_attr = sai_thrift_get_fdb_entry_attribute(self.client,
self.mac_entry,
bridge_port_id=True,
packet_action=True,
type=True)
self.assertEqual(fdb_attr["bridge_port_id"], self.port0_bp)
self.assertEqual(fdb_attr["packet_action"],
SAI_PACKET_ACTION_FORWARD)
self.assertEqual(fdb_attr["type"], SAI_FDB_ENTRY_TYPE_DYNAMIC)
print("Flush FDB entry")
sai_thrift_flush_fdb_entries(self.client,
bridge_port_id=self.port0_bp,
entry_type=SAI_FDB_ENTRY_TYPE_DYNAMIC)
print("Verifying FDB attributes")
fdb_attr = sai_thrift_get_fdb_entry_attribute(self.client,
self.mac_entry,
bridge_port_id=True,
packet_action=True,
type=True)
self.assertEqual(self.status(), SAI_STATUS_ITEM_NOT_FOUND)
print("\tEntry\t\tclear")
self.assertEqual(fdb_attr, None)
print("\tAttrs\t\tclear")
print("\tVerification complete")
finally:
sai_thrift_flush_fdb_entries(
self.client, entry_type=SAI_FDB_FLUSH_ENTRY_TYPE_ALL)
def macDeleteEventTest(self):
'''
Verify MAC deletion event for FDB entry
'''
print("\nmacDeleteEventTest()")
try:
print("Sending packet on port %d to learn MAC address, %s -> %s - "
"will flood" % (self.src_port, self.src_mac, self.dst_mac))
send_packet(self, self.src_port, self.pkt)
verify_each_packet_on_multiple_port_lists(
self, self.flood_pkts, self.flood_ports)
print("\tOK")
time.sleep(2)
fdb_attr = sai_thrift_get_fdb_entry_attribute(self.client,
self.mac_entry,
bridge_port_id=True,
packet_action=True,
type=True)
self.assertEqual(fdb_attr["bridge_port_id"], self.port0_bp)
self.assertEqual(fdb_attr["packet_action"],
SAI_PACKET_ACTION_FORWARD)
self.assertEqual(fdb_attr["type"], SAI_FDB_ENTRY_TYPE_DYNAMIC)
print("Delete FDB entry")
sai_thrift_remove_fdb_entry(self.client, self.mac_entry)
print("Verifying FDB attributes")
fdb_attr = sai_thrift_get_fdb_entry_attribute(self.client,
self.mac_entry,
bridge_port_id=True,
packet_action=True,
type=True)
self.assertEqual(self.status(), SAI_STATUS_ITEM_NOT_FOUND)
print("\tEntry\t\tclear")
self.assertEqual(fdb_attr, None)
print("\tAttrs\t\tclear")
print("\tVerification complete")
finally:
sai_thrift_flush_fdb_entries(
self.client, entry_type=SAI_FDB_FLUSH_ENTRY_TYPE_ALL)