Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Updating pytest for sflow #1095

Merged
merged 1 commit into from
Oct 31, 2019
Merged

Updating pytest for sflow #1095

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
225 changes: 199 additions & 26 deletions tests/test_sflow.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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")
Expand All @@ -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)

Expand All @@ -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)
Expand All @@ -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")
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

schema for config_db is SFLOW_SESSION not 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)

Expand All @@ -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