Updating pytest for sflow (#1095)

Author: Sudharsan D.G

tests/test_sflow.py

@@ -3,48 +3,107 @@
 import time
 import os
 
+
 class TestSflow(object):
+    speed_rate_table = {
+        "400000":"40000",
+        "100000":"10000",
+        "50000":"5000",
+        "40000":"4000",
+        "25000":"2500",
+        "10000":"1000",
+        "1000":"100"
+    }
     def setup_sflow(self, dvs):
-        self.pdb = swsscommon.DBConnector(0, dvs.redis_sock, 0)
         self.adb = swsscommon.DBConnector(1, dvs.redis_sock, 0)
-        ptbl = swsscommon.ProducerStateTable(self.pdb, "SFLOW_TABLE")
+        self.cdb = swsscommon.DBConnector(4, dvs.redis_sock, 0)
+        ctbl = swsscommon.Table(self.cdb, "SFLOW")
+
         fvs = swsscommon.FieldValuePairs([("admin_state", "up")])
-        ptbl.set("global", fvs)
+        ctbl.set("global", fvs)
 
         time.sleep(1)
 
-    def test_SflowDisble(self, dvs, testlog):
+    def test_defaultGlobal(self, dvs, testlog):
         self.setup_sflow(dvs)
-        ptbl = swsscommon.ProducerStateTable(self.pdb, "SFLOW_SESSION_TABLE")
-        gtbl = swsscommon.ProducerStateTable(self.pdb, "SFLOW_TABLE")
+        atbl = swsscommon.Table(self.adb, "ASIC_STATE:SAI_OBJECT_TYPE_PORT")
+        (status, fvs) = atbl.get(dvs.asicdb.portnamemap["Ethernet0"])
+
+        assert status == True
+
+        sample_session = ""
+        speed = ""
+        for fv in fvs:
+            if fv[0] == "SAI_PORT_ATTR_INGRESS_SAMPLEPACKET_ENABLE":
+                sample_session = fv[1]
+            elif fv[0] == "SAI_PORT_ATTR_SPEED":
+                speed = fv[1]
+
+        assert sample_session != ""
+        assert speed != ""
+
+        rate = ""
+
+        if speed in self.speed_rate_table:
+            rate = self.speed_rate_table[speed]
+
+        assert rate != ""
+
+        atbl = swsscommon.Table(self.adb, "ASIC_STATE:SAI_OBJECT_TYPE_SAMPLEPACKET")
+        (status, fvs) = atbl.get(sample_session)
+
+        assert status == True
+
+        for fv in fvs:
+            if fv[0] == "SAI_SAMPLEPACKET_ATTR_SAMPLE_RATE":
+                assert fv[1] == rate
+
+        ctbl = swsscommon.Table(self.cdb, "SFLOW")
         fvs = swsscommon.FieldValuePairs([("admin_state", "down")])
-        gtbl.set("global", fvs)
+        ctbl.set("global", fvs)
 
         time.sleep(1)
-        fvs = swsscommon.FieldValuePairs([("admin_state", "up"),("sample_rate","1000")])
-        ptbl.set("Ethernet0", fvs)
 
-        time.sleep(1)
+        atbl = swsscommon.Table(self.adb, "ASIC_STATE:SAI_OBJECT_TYPE_PORT")
+        (status, fvs) = atbl.get(dvs.asicdb.portnamemap["Ethernet0"])
 
+        assert status == True
+
+        sample_session = ""
+        speed = ""
+        for fv in fvs:
+            if fv[0] == "SAI_PORT_ATTR_INGRESS_SAMPLEPACKET_ENABLE":
+                sample_session = fv[1]
+
+        assert sample_session == "oid:0x0"
+
+    def test_globalAll(self, dvs, testlog):
+        self.setup_sflow(dvs)
+
+        ctbl = swsscommon.Table(self.cdb, "SFLOW_SESSION")
+        fvs = swsscommon.FieldValuePairs([("admin_state", "down")])
+        ctbl.set("all", fvs)
+
+        time.sleep(1)
 
         atbl = swsscommon.Table(self.adb, "ASIC_STATE:SAI_OBJECT_TYPE_PORT")
         (status, fvs) = atbl.get(dvs.asicdb.portnamemap["Ethernet0"])
 
         assert status == True
 
         sample_session = ""
+        speed = ""
         for fv in fvs:
             if fv[0] == "SAI_PORT_ATTR_INGRESS_SAMPLEPACKET_ENABLE":
                 sample_session = fv[1]
 
-        assert sample_session == ""
+        assert sample_session == "oid:0x0"
 
         fvs = swsscommon.FieldValuePairs([("admin_state", "up")])
-        gtbl.set("global", fvs)
+        ctbl.set("all", fvs)
 
         time.sleep(1)
 
-        atbl = swsscommon.Table(self.adb, "ASIC_STATE:SAI_OBJECT_TYPE_PORT")
         (status, fvs) = atbl.get(dvs.asicdb.portnamemap["Ethernet0"])
 
         assert status == True
@@ -54,7 +113,45 @@ def test_SflowDisble(self, dvs, testlog):
             if fv[0] == "SAI_PORT_ATTR_INGRESS_SAMPLEPACKET_ENABLE":
                 sample_session = fv[1]
 
+        assert sample_session != ""
         assert sample_session != "oid:0x0"
+
+        ctbl._del("all")
+
+        time.sleep(1)
+
+        (status, fvs) = atbl.get(dvs.asicdb.portnamemap["Ethernet0"])
+
+        assert status == True
+
+        sample_session = ""
+        for fv in fvs:
+            if fv[0] == "SAI_PORT_ATTR_INGRESS_SAMPLEPACKET_ENABLE":
+                sample_session = fv[1]
+
+        assert sample_session != ""
+        assert sample_session != "oid:0x0"
+
+
+    def test_InterfaceSet(self, dvs, testlog):
+        self.setup_sflow(dvs)
+        ctbl = swsscommon.Table(self.cdb, "SFLOW_SESSION")
+        gtbl = swsscommon.Table(self.cdb, "SFLOW")
+        fvs = swsscommon.FieldValuePairs([("admin_state", "up"),("sample_rate","1000")])
+        ctbl.set("Ethernet0", fvs)
+
+        time.sleep(1)
+
+        atbl = swsscommon.Table(self.adb, "ASIC_STATE:SAI_OBJECT_TYPE_PORT")
+        (status, fvs) = atbl.get(dvs.asicdb.portnamemap["Ethernet0"])
+
+        assert status == True
+
+        sample_session = ""
+        for fv in fvs:
+            if fv[0] == "SAI_PORT_ATTR_INGRESS_SAMPLEPACKET_ENABLE":
+                sample_session = fv[1]
+
         assert sample_session != ""
 
         atbl = swsscommon.Table(self.adb, "ASIC_STATE:SAI_OBJECT_TYPE_SAMPLEPACKET")
@@ -66,14 +163,25 @@ def test_SflowDisble(self, dvs, testlog):
             if fv[0] == "SAI_SAMPLEPACKET_ATTR_SAMPLE_RATE":
                 assert fv[1] == "1000"
 
-        ptbl._del("Ethernet0")
+        fvs = swsscommon.FieldValuePairs([("admin_state", "down")])
+        ctbl.set("all", fvs)
 
-    def test_InterfaceSet(self, dvs, testlog):
-        self.setup_sflow(dvs)
-        ptbl = swsscommon.ProducerStateTable(self.pdb, "SFLOW_SESSION_TABLE")
-        gtbl = swsscommon.ProducerStateTable(self.pdb, "SFLOW_TABLE")
-        fvs = swsscommon.FieldValuePairs([("admin_state", "up"),("sample_rate","1000")])
-        ptbl.set("Ethernet0", fvs)
+        time.sleep(1)
+
+        atbl = swsscommon.Table(self.adb, "ASIC_STATE:SAI_OBJECT_TYPE_PORT")
+        (status, fvs) = atbl.get(dvs.asicdb.portnamemap["Ethernet0"])
+
+        assert status == True
+
+        sample_session = ""
+        for fv in fvs:
+            if fv[0] == "SAI_PORT_ATTR_INGRESS_SAMPLEPACKET_ENABLE":
+                sample_session = fv[1]
+        assert sample_session != ""
+        assert sample_session != "oid:0x0"
+
+        fvs = swsscommon.FieldValuePairs([("admin_state", "down")])
+        gtbl.set("global", fvs)
 
         time.sleep(1)
 
@@ -87,7 +195,40 @@ def test_InterfaceSet(self, dvs, testlog):
             if fv[0] == "SAI_PORT_ATTR_INGRESS_SAMPLEPACKET_ENABLE":
                 sample_session = fv[1]
 
+        assert sample_session == "oid:0x0"
+        ctbl._del("all")
+        ctbl._del("Ethernet0")
+
+    def test_defaultRate(self, dvs, testlog):
+        self.setup_sflow(dvs)
+        ctbl = swsscommon.Table(self.cdb, "SFLOW_SESSION")
+        fvs = swsscommon.FieldValuePairs([("admin_state", "up")])
+        ctbl.set("Ethernet4", fvs)
+
+        time.sleep(1)
+
+        atbl = swsscommon.Table(self.adb, "ASIC_STATE:SAI_OBJECT_TYPE_PORT")
+        (status, fvs) = atbl.get(dvs.asicdb.portnamemap["Ethernet4"])
+
+        assert status == True
+
+        sample_session = ""
+        speed = ""
+        for fv in fvs:
+            if fv[0] == "SAI_PORT_ATTR_INGRESS_SAMPLEPACKET_ENABLE":
+                sample_session = fv[1]
+            elif fv[0] == "SAI_PORT_ATTR_SPEED":
+                speed = fv[1]
+
         assert sample_session != ""
+        assert speed != ""
+
+        rate = ""
+
+        if speed in self.speed_rate_table:
+            rate = self.speed_rate_table[speed]
+
+        assert rate != ""
 
         atbl = swsscommon.Table(self.adb, "ASIC_STATE:SAI_OBJECT_TYPE_SAMPLEPACKET")
         (status, fvs) = atbl.get(sample_session)
@@ -96,19 +237,19 @@ def test_InterfaceSet(self, dvs, testlog):
 
         for fv in fvs:
             if fv[0] == "SAI_SAMPLEPACKET_ATTR_SAMPLE_RATE":
-                assert fv[1] == "1000"
+                assert fv[1] == rate
 
-        ptbl._del("Ethernet0")
+        ctbl._del("Ethernet4")
 
     def test_ConfigDel(self, dvs, testlog):
         self.setup_sflow(dvs)
-        ptbl = swsscommon.ProducerStateTable(self.pdb, "SFLOW_SESSION_TABLE")
+        ctbl = swsscommon.Table(self.cdb, "SFLOW_SESSION_TABLE")
         fvs = swsscommon.FieldValuePairs([("admin_state", "up"),("sample_rate","1000")])
-        ptbl.set("Ethernet0", fvs)
+        ctbl.set("Ethernet0", fvs)
 
         time.sleep(1)
 
-        ptbl._del("Ethernet0")
+        ctbl._del("Ethernet0")
 
         time.sleep(1)
 
@@ -127,4 +268,36 @@ def test_ConfigDel(self, dvs, testlog):
                 speed = fv[1]
 
         assert speed != ""
-        assert sample_session == "oid:0x0"
+        assert sample_session != ""
+        assert sample_session != "oid:0x0"
+
+        rate = ""
+
+        if speed in self.speed_rate_table:
+            rate = self.speed_rate_table[speed]
+
+        assert rate != ""
+
+        atbl = swsscommon.Table(self.adb, "ASIC_STATE:SAI_OBJECT_TYPE_SAMPLEPACKET")
+        (status, fvs) = atbl.get(sample_session)
+
+        assert status == True
+
+        rf = False
+        for fv in fvs:
+            if fv[0] == "SAI_SAMPLEPACKET_ATTR_SAMPLE_RATE":
+                assert fv[1] == rate
+                rf = True
+
+        assert rf == True
+
+    def test_Teardown(self, dvs, testlog):
+        self.setup_sflow(dvs)
+        ctbl = swsscommon.Table(self.cdb, "SFLOW")
+        ctbl._del("global")
+
+        time.sleep(1)
+
+
+        atbl = swsscommon.Table(self.adb, "ASIC_STATE:SAI_OBJECT_TYPE_SAMPLEPACKET")
+        assert len(atbl.getKeys()) == 0