提交 f6c11395 编写于 作者: B Bomin Zhang

td-990: add python test cases for subscribe

上级 b1a77eda
......@@ -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);
......
###################################################################
# 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())
###################################################################
# 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())
###################################################################
# 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()
Markdown is supported
0% .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑!
想要评论请 注册