From f858b23898114edd64eafb773693ea7700fa4ed0 Mon Sep 17 00:00:00 2001 From: Shuotian Cheng Date: Fri, 24 Aug 2018 10:27:09 -0700 Subject: [PATCH] [test]: Add test_MultipleVlan test to test creation/removal (#593) Refactored test_vlan.py so that each test is independent from each other and intermediate state will be cleaned up afterwards. Add test_MultipleVlan test to cover scenarios with VLAN and VLAN members change. This test would fail on earlier kernel version .e.g 3.16.0-4 when the bridge interface cannot remain UP all the time. This would cause new VLAN failed to be created later on. The current Debian stretch doesn't have this issue and the test shall pass with current implementation. Signed-off-by: Shu0T1an ChenG --- tests/test_vlan.py | 258 ++++++++++++++++++++++++++++++--------------- 1 file changed, 174 insertions(+), 84 deletions(-) diff --git a/tests/test_vlan.py b/tests/test_vlan.py index 76b56b9c1a8a..d9f063f1c9c5 100644 --- a/tests/test_vlan.py +++ b/tests/test_vlan.py @@ -3,91 +3,181 @@ import re import json -def test_VlanMemberCreation(dvs): - - db = swsscommon.DBConnector(4, dvs.redis_sock, 0) - adb = swsscommon.DBConnector(1, dvs.redis_sock, 0) - - - # create vlan in config db - tbl = swsscommon.Table(db, "VLAN") - fvs = swsscommon.FieldValuePairs([("vlanid", "2")]) - tbl.set("Vlan2", fvs) - - time.sleep(1) - - # check vlan in asic db - atbl = swsscommon.Table(adb, "ASIC_STATE:SAI_OBJECT_TYPE_VLAN") - - keys = atbl.getKeys() - assert len(keys) == 2 - - vlan_oid = None - - for k in keys: - if k == dvs.asicdb.default_vlan_id: - continue - - (status, fvs) = atbl.get(k) +class TestVlan(object): + def setup_db(self, dvs): + self.pdb = swsscommon.DBConnector(0, dvs.redis_sock, 0) + self.adb = swsscommon.DBConnector(1, dvs.redis_sock, 0) + self.cdb = swsscommon.DBConnector(4, dvs.redis_sock, 0) + + def create_vlan(self, vlan): + tbl = swsscommon.Table(self.cdb, "VLAN") + fvs = swsscommon.FieldValuePairs([("vlanid", vlan)]) + tbl.set("Vlan" + vlan, fvs) + time.sleep(1) + + def remove_vlan(self, vlan): + tbl = swsscommon.Table(self.cdb, "VLAN") + tbl._del("Vlan" + vlan) + time.sleep(1) + + def create_vlan_member(self, vlan, interface): + tbl = swsscommon.Table(self.cdb, "VLAN_MEMBER") + fvs = swsscommon.FieldValuePairs([("tagging_mode", "untagged")]) + tbl.set("Vlan" + vlan + "|" + interface, fvs) + time.sleep(1) + + def remove_vlan_member(self, vlan, interface): + tbl = swsscommon.Table(self.cdb, "VLAN_MEMBER") + tbl._del("Vlan" + vlan + "|" + interface) + time.sleep(1) + + def test_VlanAddRemove(self, dvs): + self.setup_db(dvs) + + # create vlan + self.create_vlan("2") + + # check asic database + tbl = swsscommon.Table(self.adb, "ASIC_STATE:SAI_OBJECT_TYPE_VLAN") + vlan_entries = [k for k in tbl.getKeys() if k != dvs.asicdb.default_vlan_id] + assert len(vlan_entries) == 1 + vlan_oid = vlan_entries[0] + + (status, fvs) = tbl.get(vlan_oid) assert status == True - - if fvs[0][0] == "SAI_VLAN_ATTR_VLAN_ID": - assert fvs[0][1] == '2' - vlan_oid = k - - assert vlan_oid != None - - # create vlan member in config db - tbl = swsscommon.Table(db, "VLAN_MEMBER") - fvs = swsscommon.FieldValuePairs([("tagging_mode", "untagged")]) - tbl.set("Vlan2|Ethernet0", fvs) - - time.sleep(1) - - # check vlan member in asic db - bridge_port_map = {} - atbl = swsscommon.Table(adb, "ASIC_STATE:SAI_OBJECT_TYPE_BRIDGE_PORT") - keys = atbl.getKeys() - for k in keys: - (status, fvs) = atbl.get(k) + for fv in fvs: + if fv[0] == "SAI_VLAN_ATTR_VLAN_ID": + assert fv[1] == "2" + + # create vlan member + self.create_vlan_member("2", "Ethernet0") + + # check asic database + bridge_port_map = {} + tbl = swsscommon.Table(self.adb, "ASIC_STATE:SAI_OBJECT_TYPE_BRIDGE_PORT") + bridge_port_entries = tbl.getKeys() + for key in bridge_port_entries: + (status, fvs) = tbl.get(key) + assert status == True + for fv in fvs: + if fv[0] == "SAI_BRIDGE_PORT_ATTR_PORT_ID": + bridge_port_map[key] = fv[1] + + tbl = swsscommon.Table(self.adb, "ASIC_STATE:SAI_OBJECT_TYPE_VLAN_MEMBER") + vlan_member_entries = tbl.getKeys() + assert len(vlan_member_entries) == 1 + + (status, fvs) = tbl.get(vlan_member_entries[0]) assert status == True + assert len(fvs) == 3 + for fv in fvs: + if fv[0] == "SAI_VLAN_MEMBER_ATTR_VLAN_TAGGING_MODE": + assert fv[1] == "SAI_VLAN_TAGGING_MODE_UNTAGGED" + elif fv[0] == "SAI_VLAN_MEMBER_ATTR_VLAN_ID": + assert fv[1] == vlan_oid + elif fv[0] == "SAI_VLAN_MEMBER_ATTR_BRIDGE_PORT_ID": + assert dvs.asicdb.portoidmap[bridge_port_map[fv[1]]] == "Ethernet0" + else: + assert False + + # check port pvid + tbl = swsscommon.Table(self.adb, "ASIC_STATE:SAI_OBJECT_TYPE_PORT") + (status, fvs) = tbl.get(dvs.asicdb.portnamemap["Ethernet0"]) + assert status == True + assert "SAI_PORT_ATTR_PORT_VLAN_ID" in [fv[0] for fv in fvs] + for fv in fvs: + if fv[0] == "SAI_PORT_ATTR_PORT_VLAN_ID": + assert fv[1] == "2" + # check host interface vlan tag + tbl = swsscommon.Table(self.adb, "ASIC_STATE:SAI_OBJECT_TYPE_HOSTIF") + (status, fvs) = tbl.get(dvs.asicdb.hostifnamemap["Ethernet0"]) + assert status == True + assert "SAI_HOSTIF_ATTR_VLAN_TAG" in [fv[0] for fv in fvs] for fv in fvs: - if fv[0] == "SAI_BRIDGE_PORT_ATTR_PORT_ID": - bridge_port_map[k] = fv[1] - - atbl = swsscommon.Table(adb, "ASIC_STATE:SAI_OBJECT_TYPE_VLAN_MEMBER") - keys = atbl.getKeys() - assert len(keys) == 1 - - (status, fvs) = atbl.get(keys[0]) - assert status == True - for fv in fvs: - if fv[0] == "SAI_VLAN_MEMBER_ATTR_VLAN_TAGGING_MODE": - assert fv[1] == "SAI_VLAN_TAGGING_MODE_UNTAGGED" - elif fv[0] == "SAI_VLAN_MEMBER_ATTR_VLAN_ID": - assert fv[1] == vlan_oid - elif fv[0] == "SAI_VLAN_MEMBER_ATTR_BRIDGE_PORT_ID": - assert dvs.asicdb.portoidmap[bridge_port_map[fv[1]]] == "Ethernet0" - else: - assert False - - # check pvid of the port - atbl = swsscommon.Table(adb, "ASIC_STATE:SAI_OBJECT_TYPE_PORT") - (status, fvs) = atbl.get(dvs.asicdb.portnamemap["Ethernet0"]) - assert status == True - - assert "SAI_PORT_ATTR_PORT_VLAN_ID" in [fv[0] for fv in fvs] - for fv in fvs: - if fv[0] == "SAI_PORT_ATTR_PORT_VLAN_ID": - assert fv[1] == "2" - - # check vlan tag for the host interface - atbl = swsscommon.Table(adb, "ASIC_STATE:SAI_OBJECT_TYPE_HOSTIF") - (status, fvs) = atbl.get(dvs.asicdb.hostifnamemap["Ethernet0"]) - assert status == True - - assert "SAI_HOSTIF_ATTR_VLAN_TAG" in [fv[0] for fv in fvs] - for fv in fvs: - if fv[0] == "SAI_HOSTIF_ATTR_VLAN_TAG": - assert fv[1] == "SAI_HOSTIF_VLAN_TAG_KEEP" + if fv[0] == "SAI_HOSTIF_ATTR_VLAN_TAG": + assert fv[1] == "SAI_HOSTIF_VLAN_TAG_KEEP" + + # remove vlan member + self.remove_vlan_member("2", "Ethernet0") + + # remvoe vlan + self.remove_vlan("2") + + def test_MultipleVlan(self, dvs): + self.setup_db(dvs) + + # create vlan and vlan members + self.create_vlan("18") + self.create_vlan_member("18", "Ethernet0") + self.create_vlan_member("18", "Ethernet4") + self.create_vlan_member("18", "Ethernet8") + + # check asic database + tbl = swsscommon.Table(self.adb, "ASIC_STATE:SAI_OBJECT_TYPE_VLAN") + vlan_entries = [k for k in tbl.getKeys() if k != dvs.asicdb.default_vlan_id] + assert len(vlan_entries) == 1 + + tbl = swsscommon.Table(self.adb, "ASIC_STATE:SAI_OBJECT_TYPE_VLAN_MEMBER") + vlan_member_entries = tbl.getKeys() + assert len(vlan_member_entries) == 3 + + # remove vlan members + self.remove_vlan_member("18", "Ethernet0") + self.remove_vlan_member("18", "Ethernet4") + self.remove_vlan_member("18", "Ethernet8") + + tbl = swsscommon.Table(self.adb, "ASIC_STATE:SAI_OBJECT_TYPE_VLAN_MEMBER") + vlan_member_entries = tbl.getKeys() + assert len(vlan_member_entries) == 0 + + # create vlan and vlan members + self.create_vlan("188") + self.create_vlan_member("188", "Ethernet20") + self.create_vlan_member("188", "Ethernet24") + self.create_vlan_member("188", "Ethernet28") + + # check asic database + tbl = swsscommon.Table(self.adb, "ASIC_STATE:SAI_OBJECT_TYPE_VLAN") + vlan_entries = [k for k in tbl.getKeys() if k != dvs.asicdb.default_vlan_id] + assert len(vlan_entries) == 2 + + tbl = swsscommon.Table(self.adb, "ASIC_STATE:SAI_OBJECT_TYPE_VLAN_MEMBER") + vlan_member_entries = tbl.getKeys() + assert len(vlan_member_entries) == 3 + + # create vlan members + self.create_vlan_member("18", "Ethernet40") + self.create_vlan_member("18", "Ethernet44") + self.create_vlan_member("18", "Ethernet48") + + tbl = swsscommon.Table(self.adb, "ASIC_STATE:SAI_OBJECT_TYPE_VLAN_MEMBER") + vlan_member_entries = tbl.getKeys() + assert len(vlan_member_entries) == 6 + + # remove vlan members + self.remove_vlan_member("18", "Ethernet40") + self.remove_vlan_member("18", "Ethernet44") + self.remove_vlan_member("18", "Ethernet48") + + tbl = swsscommon.Table(self.adb, "ASIC_STATE:SAI_OBJECT_TYPE_VLAN_MEMBER") + vlan_member_entries = tbl.getKeys() + assert len(vlan_member_entries) == 3 + + # remove vlan members + self.remove_vlan_member("188", "Ethernet20") + self.remove_vlan_member("188", "Ethernet24") + self.remove_vlan_member("188", "Ethernet28") + + tbl = swsscommon.Table(self.adb, "ASIC_STATE:SAI_OBJECT_TYPE_VLAN_MEMBER") + vlan_member_entries = tbl.getKeys() + assert len(vlan_member_entries) == 0 + + # remove vlans + self.remove_vlan("18") + self.remove_vlan("188") + + # check asic database + tbl = swsscommon.Table(self.adb, "ASIC_STATE:SAI_OBJECT_TYPE_VLAN") + vlan_entries = [k for k in tbl.getKeys() if k != dvs.asicdb.default_vlan_id] + assert len(vlan_entries) == 0