[vstest]: get subscribed objects from appdb and asicdb (#664)

Signed-off-by: Guohan Lu <gulv@microsoft.com>

tests/conftest.py

@@ -327,6 +327,12 @@ def add_log_marker(self):
         self.ctn.exec_run("logger {}".format(marker))
         return marker
 
+    def SubscribeAppDbObject(self, objpfx):
+        r = redis.Redis(unix_socket_path=self.redis_sock, db=swsscommon.APP_DB)
+        pubsub = r.pubsub()
+        pubsub.psubscribe("__keyspace@0__:%s*" % objpfx)
+        return pubsub
+
     def SubscribeAsicDbObject(self, objpfx):
         r = redis.Redis(unix_socket_path=self.redis_sock, db=swsscommon.ASIC_DB)
         pubsub = r.pubsub()
@@ -356,6 +362,64 @@ def CountSubscribedObjects(self, pubsub, ignore=None, timeout=10):
 
         return (nadd, ndel)
 
+    def GetSubscribedAppDbObjects(self, pubsub, ignore=None, timeout=10):
+        r = redis.Redis(unix_socket_path=self.redis_sock, db=swsscommon.APP_DB)
+
+        addobjs = []
+        delobjs = []
+        idle = 0
+
+        while True and idle < timeout:
+            message = pubsub.get_message()
+            if message:
+                print message
+                key = message['channel'].split(':', 1)[1]
+                if ignore:
+                    fds = message['channel'].split(':')
+                    if fds[2] in ignore:
+                        continue
+                if message['data'] == 'hset':
+                    value=r.hgetall(key)
+                    addobjs.append({'key':k, 'vals':value})
+                elif message['data'] == 'del':
+                    delobjs.append(key)
+                idle = 0
+            else:
+                time.sleep(1)
+                idle += 1
+
+        return (addobjs, delobjs)
+
+
+    def GetSubscribedAsicDbObjects(self, pubsub, ignore=None, timeout=10):
+        r = redis.Redis(unix_socket_path=self.redis_sock, db=swsscommon.ASIC_DB)
+
+        addobjs = []
+        delobjs = []
+        idle = 0
+
+        while True and idle < timeout:
+            message = pubsub.get_message()
+            if message:
+                print message
+                key = message['channel'].split(':', 1)[1]
+                if ignore:
+                    fds = message['channel'].split(':')
+                    if fds[2] in ignore:
+                        continue
+                if message['data'] == 'hset':
+                    value=r.hgetall(key)
+                    (_, t, k) = key.split(':', 2)
+                    addobjs.append({'type':t, 'key':k, 'vals':value})
+                elif message['data'] == 'del':
+                    delobjs.append(key)
+                idle = 0
+            else:
+                time.sleep(1)
+                idle += 1
+
+        return (addobjs, delobjs)
+
     def get_map_iface_bridge_port_id(self, asic_db):
         port_id_2_iface = self.asicdb.portoidmap
         tbl = swsscommon.Table(asic_db, "ASIC_STATE:SAI_OBJECT_TYPE_BRIDGE_PORT")

tests/test_route.py

@@ -22,23 +22,16 @@ def test_RouteAdd(dvs, testlog):
     ps = swsscommon.ProducerStateTable(db, "ROUTE_TABLE")
     fvs = swsscommon.FieldValuePairs([("nexthop","10.0.0.1"), ("ifname", "Ethernet0")])
 
-    ps.set("2.2.2.0/24", fvs)
+    pubsub = dvs.SubscribeAsicDbObject("SAI_OBJECT_TYPE_ROUTE_ENTRY")
 
-    time.sleep(1)
+    ps.set("2.2.2.0/24", fvs)
 
     # check if route was propagated to ASIC DB
 
-    db = swsscommon.DBConnector(1, dvs.redis_sock, 0)
-
-    tbl = swsscommon.Table(db, "ASIC_STATE:SAI_OBJECT_TYPE_ROUTE_ENTRY")
-
-    keys = tbl.getKeys()
+    (addobjs, delobjs) = dvs.GetSubscribedAsicDbObjects(pubsub)
 
-    found_route = False
-    for k in keys:
-        rt_key = json.loads(k)
+    assert len(addobjs) == 1
 
-        if rt_key['dest'] == "2.2.2.0/24":
-            found_route = True
+    rt_key = json.loads(addobjs[0]['key'])
 
-    assert found_route
+    assert rt_key['dest'] == "2.2.2.0/24"