[vstest]Migrating vs tests from using click commands to direct DB access

tests/conftest.py

@@ -1157,6 +1157,44 @@ def getCrmCounterValue(self, key, counter):
             if k[0] == counter:
                 return int(k[1])
 
+    def port_field_set(self, port, field, value):
+        cdb = swsscommon.DBConnector(4, self.redis_sock, 0)
+        tbl = swsscommon.Table(cdb, "PORT")
+        fvs = swsscommon.FieldValuePairs([(field, value)])
+        tbl.set(port, fvs)
+        time.sleep(1)
+
+    def port_admin_set(self, port, status):
+        self.port_field_set(port, "admin_status", status)
+
+    def interface_ip_add(self, port, ip_address):
+        cdb = swsscommon.DBConnector(4, self.redis_sock, 0)
+        tbl = swsscommon.Table(cdb, "INTERFACE")
+        fvs = swsscommon.FieldValuePairs([("NULL", "NULL")])
+        tbl.set(port, fvs)
+        tbl.set(port + "|" + ip_address, fvs)
+        time.sleep(1)
+
+    def crm_poll_set(self, value):
+        cdb = swsscommon.DBConnector(4, self.redis_sock, 0)
+        tbl = swsscommon.Table(cdb, "CRM")
+        fvs = swsscommon.FieldValuePairs([("polling_interval", value)])
+        tbl.set("Config", fvs)
+        time.sleep(1)
+
+    def clear_fdb(self):
+        adb = swsscommon.DBConnector(0, self.redis_sock, 0)
+        opdata = ["ALL", "ALL"]
+        msg = json.dumps(opdata,separators=(',',':'))
+        adb.publish('FLUSHFDBREQUEST', msg)
+
+    def warm_restart_swss(self, enable):
+        db = swsscommon.DBConnector(6, self.redis_sock, 0)
+
+        tbl = swsscommon.Table(db, "WARM_RESTART_ENABLE_TABLE")
+        fvs = swsscommon.FieldValuePairs([("enable",enable)])
+        tbl.set("swss", fvs)
+
     # deps: acl, crm, fdb
     def setReadOnlyAttr(self, obj, attr, val):
         db = swsscommon.DBConnector(swsscommon.ASIC_DB, self.redis_sock, 0)

tests/p4rt/test_l3.py

@@ -1328,7 +1328,7 @@ def test_PruneNextHopOnWarmBoot(self, dvs, testlog):
         util.set_interface_status(dvs, if_name)
 
         # Execute the warm reboot.
-        dvs.runcmd("config warm_restart enable swss")
+        dvs.warm_restart_swss("true")
         dvs.stop_swss()
         dvs.start_swss()
 

tests/p4rt/util.py

@@ -84,8 +84,8 @@ def get_port_oid_by_name(dvs, port_name):
   return port_oid
 
 def initialize_interface(dvs, port_name, ip):
-  dvs.runcmd("config interface startup {}".format(port_name))
-  dvs.runcmd("config interface ip add {} {}".format(port_name, ip))
+  dvs.port_admin_set(port_name, "up")
+  dvs.interface_ip_add(port_name, ip)
 
 def set_interface_status(dvs, if_name, status = "down", server = 0):
   dvs.servers[0].runcmd("ip link set {} dev {}".format(status, if_name)) == 0

tests/test_acl.py

@@ -553,11 +553,11 @@ def test_AclRuleRedirect(self, dvs, dvs_acl, l3_acl_table, setup_teardown_neighb
 class TestAclCrmUtilization:
     @pytest.fixture(scope="class", autouse=True)
     def configure_crm_polling_interval_for_test(self, dvs):
-        dvs.runcmd("crm config polling interval 1")
+        dvs.crm_poll_set("1")
 
         yield
 
-        dvs.runcmd("crm config polling interval 300")
+        dvs.crm_poll_set("300")
 
     def test_ValidateAclTableBindingCrmUtilization(self, dvs, dvs_acl):
         counter_db = dvs.get_counters_db()

tests/test_acl_portchannel.py

@@ -129,7 +129,7 @@ def check_asic_table_absent(self, dvs):
     # Second create ACL table
     def test_PortChannelAfterAcl(self, dvs):
         self.setup_db(dvs)
-        dvs.runcmd("crm config polling interval 1")
+        dvs.crm_poll_set("1")
         time.sleep(2)
 
         used_counter = dvs.getCrmCounterValue('ACL_STATS:INGRESS:LAG', 'crm_stats_acl_group_used')
@@ -162,7 +162,7 @@ def test_PortChannelAfterAcl(self, dvs):
             new_new_used_counter = 0
         assert new_used_counter - new_new_used_counter == 1
         # slow down crm polling
-        dvs.runcmd("crm config polling interval 10000")
+        dvs.crm_poll_set("10000")
 
     # Frist create ACL table
     # Second create port channel