DPB VLAN system test case and pulled in VLAN related DVSLIB code (#46)

* DPB VLAN system test case and pulled in VLAN related DVSLIB code * Address code-review comments Co-authored-by: Vasant <[email protected]>
zhenggen-xu · Jun 9, 2020 · f0686b9 · f0686b9
1 parent 1ee0deb
commit f0686b9
Show file tree

Hide file tree

Showing 4 changed files with 157 additions and 0 deletions.
diff --git a/tests/conftest.py b/tests/conftest.py
@@ -15,6 +15,8 @@
 
 from dvslib import dvs_database as dvs_db
 from dvslib import dvs_acl
+from dvslib import dvs_vlan
+from dvslib import dvs_lag
 
 def ensure_system(cmd):
     rc = os.WEXITSTATUS(os.system(cmd))
@@ -1044,13 +1046,27 @@ def testlog(request, dvs):
     yield testlog
     dvs.runcmd("logger === finish test %s ===" % request.node.name)
 
+
+################# DVSLIB module manager fixtures #############################
 @pytest.yield_fixture(scope="class")
 def dvs_acl_manager(request, dvs):
     request.cls.dvs_acl = dvs_acl.DVSAcl(dvs.get_asic_db(),
                                          dvs.get_config_db(),
                                          dvs.get_state_db(),
                                          dvs.get_counters_db())
 
+@pytest.yield_fixture(scope="class")
+def dvs_lag_manager(request, dvs):
+    request.cls.dvs_lag = dvs_lag.DVSLag(dvs.get_config_db())
+
+@pytest.yield_fixture(scope="class")
+def dvs_vlan_manager(request, dvs):
+    request.cls.dvs_vlan = dvs_vlan.DVSVlan(dvs.get_asic_db(),
+                                            dvs.get_config_db(),
+                                            dvs.get_state_db(),
+                                            dvs.get_counters_db(),
+                                            dvs.get_app_db())
+
 ##################### DPB fixtures ###########################################
 @pytest.yield_fixture(scope="module")
 def create_dpb_config_file(dvs):

diff --git a/tests/dvslib/dvs_lag.py b/tests/dvslib/dvs_lag.py
@@ -0,0 +1,21 @@
+class DVSLag(object):
+    def __init__(self, cdb):
+        self.config_db = cdb
+
+    def create_port_channel(self, lag_id, admin_status="up", mtu="1500"):
+        lag = "PortChannel{}".format(lag_id)
+        lag_entry = {"admin_status": admin_status, "mtu": mtu}
+        self.config_db.create_entry("PORTCHANNEL", lag, lag_entry)
+
+    def remove_port_channel(self, lag_id):
+        lag = "PortChannel{}".format(lag_id)
+        self.config_db.delete_entry("PORTCHANNEL", lag)
+
+    def create_port_channel_member(self, lag_id, interface):
+        member = "PortChannel{}|{}".format(lag_id, interface)
+        member_entry = {"NULL": "NULL"}
+        self.config_db.create_entry("PORTCHANNEL_MEMBER", member, member_entry)
+
+    def remove_port_channel_member(self, lag_id, interface):
+        member = "PortChannel{}|{}".format(lag_id, interface)
+        self.config_db.delete_entry("PORTCHANNEL_MEMBER", member)
diff --git a/tests/dvslib/dvs_vlan.py b/tests/dvslib/dvs_vlan.py
@@ -0,0 +1,76 @@
+from dvs_database import DVSDatabase
+
+class DVSVlan(object):
+    def __init__(self, adb, cdb, sdb, cntrdb, appdb):
+        self.asic_db = adb
+        self.config_db = cdb
+        self.state_db = sdb
+        self.counters_db = cntrdb
+        self.app_db = appdb
+
+    def create_vlan(self, vlanID):
+        vlanName = "Vlan{}".format(vlanID)
+        vlan_entry = {"vlanid": vlanID}
+        self.config_db.create_entry("VLAN", vlanName, vlan_entry)
+
+    def remove_vlan(self, vlanID):
+        vlanName = "Vlan{}".format(vlanID)
+        self.config_db.delete_entry("VLAN", vlanName)
+
+    def create_vlan_member(self, vlan, interface, tagging_mode="untagged"):
+        member = "Vlan{}|{}".format(vlan, interface)
+        if tagging_mode:
+            member_entry = {"tagging_mode": tagging_mode}
+        else:
+            member_entry = {"no_tag_mode": ""}
+
+        self.config_db.create_entry("VLAN_MEMBER", member, member_entry)
+
+    def remove_vlan_member(self, vlan, interface):
+        member = "Vlan{}|{}".format(vlan, interface)
+        self.config_db.delete_entry("VLAN_MEMBER", member)
+
+    def check_app_db_vlan_fields(self, fvs, admin_status="up", mtu="9100"):
+        assert fvs.get("admin_status") == admin_status
+        assert fvs.get("mtu") == mtu
+
+    def check_app_db_vlan_member_fields(self, fvs, tagging_mode="untagged"):
+        assert fvs.get("tagging_mode") == tagging_mode
+
+    def check_state_db_vlan_fields(self, fvs, state="ok"):
+        assert fvs.get("state") == state
+
+    def check_state_db_vlan_member_fields(self, fvs, state="ok"):
+        assert fvs.get("state") == state
+
+    def verify_vlan(self, vlan_oid, vlan_id):
+        vlan = self.asic_db.wait_for_entry("ASIC_STATE:SAI_OBJECT_TYPE_VLAN", vlan_oid)
+        assert vlan.get("SAI_VLAN_ATTR_VLAN_ID") == vlan_id
+
+    def get_and_verify_vlan_ids(self,
+                                expected_num,
+                                polling_config=DVSDatabase.DEFAULT_POLLING_CONFIG):
+        vlan_entries = self.asic_db.wait_for_n_keys("ASIC_STATE:SAI_OBJECT_TYPE_VLAN",
+                                                    expected_num + 1,
+                                                    polling_config)
+        return [v for v in vlan_entries if v != self.asic_db.default_vlan_id]
+
+    def verify_vlan_member(self, vlan_oid, iface, tagging_mode="SAI_VLAN_TAGGING_MODE_UNTAGGED"):
+        member_ids = self.asic_db.wait_for_n_keys("ASIC_STATE:SAI_OBJECT_TYPE_VLAN_MEMBER", 1)
+        member = self.asic_db.wait_for_entry("ASIC_STATE:SAI_OBJECT_TYPE_VLAN_MEMBER", member_ids[0])
+        assert member == {"SAI_VLAN_MEMBER_ATTR_VLAN_TAGGING_MODE": tagging_mode,
+                          "SAI_VLAN_MEMBER_ATTR_VLAN_ID": vlan_oid,
+                          "SAI_VLAN_MEMBER_ATTR_BRIDGE_PORT_ID": self.get_bridge_port_id(iface)}
+
+    def get_and_verify_vlan_member_ids(self, expected_num):
+        return self.asic_db.wait_for_n_keys("ASIC_STATE:SAI_OBJECT_TYPE_VLAN_MEMBER", expected_num)
+
+    def get_bridge_port_id(self, expected_iface):
+        bridge_port_id = self.asic_db.wait_for_n_keys("ASIC_STATE:SAI_OBJECT_TYPE_BRIDGE_PORT", 1)[0]
+        bridge_port = self.asic_db.wait_for_entry("ASIC_STATE:SAI_OBJECT_TYPE_BRIDGE_PORT", bridge_port_id)
+        #TBD: port_to_id_map may NOT be updated one in case port is deleted and re-created.
+        #     Hence the map needs refreshed. Need to think trough and decide when and where
+        #     to do it.
+        assert self.asic_db.port_to_id_map[bridge_port["SAI_BRIDGE_PORT_ATTR_PORT_ID"]] == expected_iface
+        return bridge_port_id
+
diff --git a/tests/test_port_dpb_system.py b/tests/test_port_dpb_system.py
@@ -11,6 +11,7 @@
 
 @pytest.mark.usefixtures('dpb_setup_fixture')
 @pytest.mark.usefixtures('dvs_acl_manager')
+@pytest.mark.usefixtures('dvs_vlan_manager')
 class TestPortDPBSystem(object):
 
     def verify_only_ports_exist(self, dvs, port_names):
@@ -167,6 +168,49 @@ def test_port_breakout_one(self, dvs):
         self.verify_only_ports_exist(dvs, ["Ethernet0"])
         print "**** 2x25G(2)+1x50G(2) --> 1x100G passed ****"
 
+    def test_port_breakout_with_vlan(self, dvs):
+        dvs.setup_db()
+        portName = "Ethernet0"
+        vlanID = "100"
+        breakoutMode1 = "1x100G[40G]"
+        breakoutMode2 = "4x25G[10G]"
+        breakoutOption = "-f" #Force breakout by deleting dependencies
+
+        # Create VLAN
+        self.dvs_vlan.create_vlan(vlanID)
+
+        # Verify VLAN is created
+        self.dvs_vlan.get_and_verify_vlan_ids(1)
+
+        # Add port to VLAN
+        self.dvs_vlan.create_vlan_member(vlanID, portName)
+
+        # Verify VLAN member is created
+        self.dvs_vlan.get_and_verify_vlan_member_ids(1)
+
+        # Breakout port from 1x100G[40G] --> 4x25G[10G]
+        dvs.verify_port_breakout_mode("Ethernet0", breakoutMode1)
+        dvs.change_port_breakout_mode("Ethernet0", breakoutMode2, breakoutOption)
+
+        # Verify DPB is successful
+        dvs.verify_port_breakout_mode("Ethernet0", breakoutMode2)
+
+        # Verify port is removed from VLAN
+        self.dvs_vlan.get_and_verify_vlan_member_ids(0)
+
+        # Delete VLAN
+        self.dvs_vlan.remove_vlan(vlanID)
+
+        # Verify VLAN is deleted
+        self.dvs_vlan.get_and_verify_vlan_ids(0)
+
+        # Breakout port from 4x25G[10G] --> 1x100G[40G]
+        dvs.change_port_breakout_mode("Ethernet0", breakoutMode1)
+
+        # Verify DPB is successful
+        dvs.verify_port_breakout_mode("Ethernet0", breakoutMode1)
+
+
     @pytest.mark.skip()
     def test_port_breakout_with_acl(self, dvs):
         dvs.setup_db()