[ppi]: Improve code coverage.

Signed-off-by: Nazarii Hnydyn <[email protected]>

Author: nazariig

tests/conftest.py

@@ -1858,7 +1858,7 @@ def vct(request):
 @pytest.fixture
 def testlog(request, dvs):
     dvs.runcmd(f"logger -t pytest === start test {request.node.nodeid} ===")
-    yield testlog
+    yield
     dvs.runcmd(f"logger -t pytest === finish test {request.node.nodeid} ===")
 
 ################# DVSLIB module manager fixtures #############################
@@ -1903,6 +1903,7 @@ def dvs_vlan_manager(request, dvs):
 @pytest.fixture(scope="class")
 def dvs_port_manager(request, dvs):
     request.cls.dvs_port = dvs_port.DVSPort(dvs.get_asic_db(),
+                                            dvs.get_app_db(),
                                             dvs.get_config_db())
 
 

tests/dvslib/dvs_database.py

@@ -6,6 +6,7 @@
 """
 from typing import Dict, List
 from swsscommon import swsscommon
+from swsscommon.swsscommon import SonicDBConfig
 from dvslib.dvs_common import wait_for_result, PollingConfig
 
 
@@ -21,6 +22,12 @@ def __init__(self, db_id: int, connector: str):
                 redis (e.g. UNIX socket, TCP socket, etc.).
         """
         self.db_connection = swsscommon.DBConnector(db_id, connector, 0)
+        self._separator = SonicDBConfig.getSeparator(self.db_connection)
+
+    @property
+    def separator(self) -> str:
+        """Get DB separator."""
+        return self._separator
 
     def create_entry(self, table_name: str, key: str, entry: Dict[str, str]) -> None:
         """Add the mapping {`key` -> `entry`} to the specified table.

tests/dvslib/dvs_port.py

@@ -1,8 +1,44 @@
+"""Utilities for interacting with PORT objects when writing VS tests."""
+from typing import Dict, List
+from swsscommon import swsscommon
+
 
 class DVSPort(object):
-    def __init__(self, adb, cdb):
-        self.asic_db = adb
-        self.config_db = cdb
+    """Manage PORT objects on the virtual switch."""
+    ASIC_DB = swsscommon.ASIC_DB
+    APPL_DB = swsscommon.APPL_DB
+
+    CFGDB_PORT = "PORT"
+    APPDB_PORT = "PORT_TABLE"
+    ASICDB_PORT = "ASIC_STATE:SAI_OBJECT_TYPE_PORT"
+
+    def __init__(self, asicdb, appdb, cfgdb):
+        self.asic_db = asicdb
+        self.app_db = appdb
+        self.config_db = cfgdb
+
+    def create_port_generic(
+        self,
+        port_name: str,
+        lanes: str,
+        speed: str,
+        qualifiers: Dict[str, str] = {}
+    ) -> None:
+        """Create PORT in Config DB."""
+        attr_dict = {
+            "lanes": lanes,
+            "speed": speed,
+            **qualifiers
+        }
+
+        self.config_db.create_entry(self.CFGDB_PORT, port_name, attr_dict)
+
+    def remove_port_generic(
+        self,
+        port_name: str
+    )-> None:
+        """Remove PORT from Config DB."""
+        self.config_db.delete_entry(self.CFGDB_PORT, port_name)
 
     def remove_port(self, port_name):
         self.config_db.delete_field("CABLE_LENGTH", "AZURE", port_name)
@@ -18,3 +54,42 @@ def remove_port(self, port_name):
         self.config_db.delete_entry("BREAKOUT_CFG|%s" % port_name, "")
         self.config_db.delete_entry("INTERFACE|%s" % port_name, "")
         self.config_db.delete_entry("PORT", port_name)
+
+    def update_port(
+        self,
+        port_name: str,
+        attr_dict: Dict[str, str]
+    ) -> None:
+        """Update PORT in Config DB."""
+        self.config_db.update_entry(self.CFGDB_PORT, port_name, attr_dict)
+
+    def get_port_ids(
+        self,
+        expected: int = None,
+        dbid: int = swsscommon.ASIC_DB
+    ) -> List[str]:
+        """Get all of the PORT objects in ASIC/APP DB."""
+        conn = None
+        table = None
+
+        if dbid == swsscommon.ASIC_DB:
+            conn = self.asic_db
+            table = self.ASICDB_PORT
+        elif dbid == swsscommon.APPL_DB:
+            conn = self.app_db
+            table = self.APPDB_PORT
+        else:
+            raise RuntimeError("Interface not implemented")
+
+        if expected is None:
+            return conn.get_keys(table)
+
+        return conn.wait_for_n_keys(table, expected)
+
+    def verify_port_count(
+        self,
+        expected: int,
+        dbid: int = swsscommon.ASIC_DB
+    ) -> None:
+        """Verify that there are N PORT objects in ASIC/APP DB."""
+        self.get_port_ids(expected, dbid)