Merge pull request #2758 from taosdata/feature/subscribe-test

td-990: add python test cases for subscribe

src/client/src/tscSub.c

@@ -256,11 +256,12 @@ static int tscUpdateSubscription(STscObj* pObj, SSub* pSub) {
   }
   size_t numOfTables = taosArrayGetSize(tables);
 
+  SQueryInfo* pQueryInfo = tscGetQueryInfoDetail(pCmd, 0);
   SArray* progress = taosArrayInit(numOfTables, sizeof(SSubscriptionProgress));
   for( size_t i = 0; i < numOfTables; i++ ) {
     STidTags* tt = taosArrayGet( tables, i );
     SSubscriptionProgress p = { .uid = tt->uid };
-    p.key = tscGetSubscriptionProgress(pSub, tt->uid, INT64_MIN);
+    p.key = tscGetSubscriptionProgress(pSub, tt->uid, pQueryInfo->window.skey);
     taosArrayPush(progress, &p);
   }
   taosArraySort(progress, tscCompareSubscriptionProgress);

tests/pytest/subscribe/singlemeter.py

+###################################################################
+ #		   Copyright (c) 2020 by TAOS Technologies, Inc.
+ #				     All rights reserved.
+ #
+ #  This file is proprietary and confidential to TAOS Technologies.
+ #  No part of this file may be reproduced, stored, transmitted, 
+ #  disclosed or used in any form or by any means other than as 
+ #  expressly provided by the written permission from Jianhui Tao
+ #
+###################################################################
+
+# -*- coding: utf-8 -*-  
+
+import sys
+import taos
+import time
+from util.log import *
+from util.cases import *
+from util.sql import *
+from util.sub import *
+
+class TDTestCase:
+	def init(self, conn, logSql):
+		tdLog.debug("start to execute %s" % __file__)
+		tdSql.init(conn.cursor(), logSql)
+		self.conn = conn
+
+	def run(self):
+		sqlstr = "select * from t0"
+		topic = "test"
+		now = int(time.time() * 1000)
+		tdSql.prepare()
+
+		tdLog.info("create a table and insert 10 rows.")
+		tdSql.execute("create table t0(ts timestamp, a int, b int);")
+		for i in range(0, 10):
+			tdSql.execute("insert into t0 values (%d, %d, %d);" % (now + i, i, i))
+
+		tdLog.info("consumption 01.")
+		tdSub.init(self.conn.subscribe(True, topic, sqlstr, 0))
+		tdSub.consume()
+		tdSub.checkRows(10)
+
+		tdLog.info("consumption 02: no new rows inserted")
+		tdSub.consume()
+		tdSub.checkRows(0)
+
+		tdLog.info("consumption 03: after one new rows inserted")
+		tdSql.execute("insert into t0 values (%d, 10, 10);" % (now + 10))
+		tdSub.consume()
+		tdSub.checkRows(1)
+
+		tdLog.info("consumption 04: keep progress and continue previous subscription")
+		tdSub.close(True)
+		tdSub.init(self.conn.subscribe(False, topic, sqlstr, 0))
+		tdSub.consume()
+		tdSub.checkRows(0)
+
+		tdLog.info("consumption 05: remove progress and continue previous subscription")
+		tdSub.close(False)
+		tdSub.init(self.conn.subscribe(False, topic, sqlstr, 0))
+		tdSub.consume()
+		tdSub.checkRows(11)
+
+		tdLog.info("consumption 06: keep progress and restart the subscription")
+		tdSub.close(True)
+		tdSub.init(self.conn.subscribe(True, topic, sqlstr, 0))
+		tdSub.consume()
+		tdSub.checkRows(11)
+
+		tdSub.close(True)
+
+	def stop(self):
+		tdSub.close(False)
+		tdSql.close()
+		tdLog.success("%s successfully executed" % __file__)
+	
+tdCases.addWindows(__file__, TDTestCase())
+tdCases.addLinux(__file__, TDTestCase())

tests/pytest/subscribe/supertable.py

+###################################################################
+ #		   Copyright (c) 2020 by TAOS Technologies, Inc.
+ #				     All rights reserved.
+ #
+ #  This file is proprietary and confidential to TAOS Technologies.
+ #  No part of this file may be reproduced, stored, transmitted, 
+ #  disclosed or used in any form or by any means other than as 
+ #  expressly provided by the written permission from Jianhui Tao
+ #
+###################################################################
+
+# -*- coding: utf-8 -*-  
+
+import sys
+import taos
+import time
+from util.log import *
+from util.cases import *
+from util.sql import *
+from util.sub import *
+
+class TDTestCase:
+	def init(self, conn, logSql):
+		tdLog.debug("start to execute %s" % __file__)
+		tdSql.init(conn.cursor(), logSql)
+		self.conn = conn
+
+	def run(self):
+		sqlstr = "select * from meters"
+		topic = "test"
+		now = int(time.time() * 1000)
+		tdSql.prepare()
+
+		tdLog.info("create a super table and 10 sub-tables, then insert 5 rows into each sub-table.")
+		tdSql.execute("create table meters(ts timestamp, a int, b int) tags(area int, loc binary(20));")
+		for i in range(0, 10):
+			for j in range(0, 5):
+				tdSql.execute("insert into t%d using meters tags(%d, 'area%d') values (%d, %d, %d);" % (i, i, i, now + j, j, j))
+
+		tdLog.info("consumption 01.")
+		tdSub.init(self.conn.subscribe(True, topic, sqlstr, 0))
+		tdSub.consume()
+		tdSub.checkRows(50)
+
+		tdLog.info("consumption 02: no new rows inserted")
+		tdSub.consume()
+		tdSub.checkRows(0)
+
+		tdLog.info("consumption 03: after one new rows inserted")
+		tdSql.execute("insert into t0 values (%d, 10, 10);" % (now + 10))
+		tdSub.consume()
+		tdSub.checkRows(1)
+
+		tdLog.info("consumption 04: keep progress and continue previous subscription")
+		tdSub.close(True)
+		tdSub.init(self.conn.subscribe(False, topic, sqlstr, 0))
+		tdSub.consume()
+		tdSub.checkRows(0)
+
+		tdLog.info("consumption 05: remove progress and continue previous subscription")
+		tdSub.close(False)
+		tdSub.init(self.conn.subscribe(False, topic, sqlstr, 0))
+		tdSub.consume()
+		tdSub.checkRows(51)
+
+		tdLog.info("consumption 06: keep progress and restart the subscription")
+		tdSub.close(True)
+		tdSub.init(self.conn.subscribe(True, topic, sqlstr, 0))
+		tdSub.consume()
+		tdSub.checkRows(51)
+
+		tdLog.info("consumption 07: insert one row to two table then remove one table")
+		tdSql.execute("insert into t0 values (%d, 11, 11);" % (now + 11))
+		tdSql.execute("insert into t1 values (%d, 11, 11);" % (now + 11))
+		tdSql.execute("drop table t0")
+		tdSub.consume()
+		tdSub.checkRows(1)
+
+		tdLog.info("consumption 08: check timestamp criteria")
+		tdSub.close(False)
+		tdSub.init(self.conn.subscribe(True, topic, sqlstr + " where ts > %d" % now, 0))
+		tdSub.consume()
+		tdSub.checkRows(37)
+
+		tdLog.info("consumption 09: insert large timestamp to t2 then insert smaller timestamp to t1")
+		tdSql.execute("insert into t2 values (%d, 100, 100);" % (now + 100))
+		tdSub.consume()
+		tdSub.checkRows(1)
+		tdSql.execute("insert into t1 values (%d, 12, 12);" % (now + 12))
+		tdSub.consume()
+		tdSub.checkRows(1)
+
+		tdLog.info("consumption 10: field criteria")
+		tdSub.close(True)
+		tdSub.init(self.conn.subscribe(False, topic, sqlstr + " where a > 100", 0))
+		tdSql.execute("insert into t2 values (%d, 101, 100);" % (now + 101))
+		tdSql.execute("insert into t2 values (%d, 100, 100);" % (now + 102))
+		tdSql.execute("insert into t2 values (%d, 102, 100);" % (now + 103))
+		tdSub.consume()
+		tdSub.checkRows(2)
+
+		tdLog.info("consumption 11: two vnodes")
+		tdSql.execute("insert into t2 values (%d, 102, 100);" % (now + 104))
+		tdSql.execute("insert into t9 values (%d, 102, 100);" % (now + 104))
+		tdSub.consume()
+		tdSub.checkRows(2)
+
+	def stop(self):
+		tdSub.close(False)
+		tdSql.close()
+		tdLog.success("%s successfully executed" % __file__)
+	
+tdCases.addWindows(__file__, TDTestCase())
+tdCases.addLinux(__file__, TDTestCase())

tests/pytest/util/sub.py

+###################################################################
+ #           Copyright (c) 2020 by TAOS Technologies, Inc.
+ #                     All rights reserved.
+ #
+ #  This file is proprietary and confidential to TAOS Technologies.
+ #  No part of this file may be reproduced, stored, transmitted, 
+ #  disclosed or used in any form or by any means other than as 
+ #  expressly provided by the written permission from Jianhui Tao
+ #
+###################################################################
+
+# -*- coding: utf-8 -*-  
+
+import sys
+import os
+import time
+import datetime
+from util.log import *
+
+class TDSub:
+	def __init__(self):
+		self.consumedRows = 0
+		self.consumedCols = 0
+
+	def init(self, sub):
+		self.sub = sub
+
+	def close(self, keepProgress):
+		self.sub.close(keepProgress)	
+
+	def consume(self):
+		self.data = self.sub.consume()
+		self.consumedRows = len(self.data)
+		self.consumedCols = len(self.sub.fields)
+		return self.consumedRows
+
+	def checkRows(self, expectRows):
+		if self.consumedRows != expectRows:
+			tdLog.exit("consumed rows:%d != expect:%d" % (self.consumedRows, expectRows))
+		tdLog.info("consumed rows:%d == expect:%d" % (self.consumedRows, expectRows))
+	
+
+tdSub = TDSub()