From af5bcb80bffddce7687a2b41efc220851ff9cb1c Mon Sep 17 00:00:00 2001 From: wangmm0220 Date: Mon, 24 Apr 2023 10:26:40 +0800 Subject: [PATCH] opti:test cases for tmq --- .../6-cluster/clusterCommonCreate.py | 40 ++++++++--------- tests/system-test/7-tmq/subscribeDb3.py | 45 ++++++++++--------- tests/system-test/7-tmq/tmqCommon.py | 22 +++++---- tests/system-test/7-tmq/tmqConsumerGroup.py | 2 +- tests/system-test/7-tmq/tmqDnodeRestart1.py | 4 +- 5 files changed, 56 insertions(+), 57 deletions(-) diff --git a/tests/system-test/6-cluster/clusterCommonCreate.py b/tests/system-test/6-cluster/clusterCommonCreate.py index 6e699e2396..a06c1233d8 100644 --- a/tests/system-test/6-cluster/clusterCommonCreate.py +++ b/tests/system-test/6-cluster/clusterCommonCreate.py @@ -94,26 +94,26 @@ class ClusterComCreate: tdLog.info(shellCmd) os.system(shellCmd) - def getStartConsumeNotifyFromTmqsim(self,cdbName='cdb'): - while 1: - tdSql.query("select * from %s.notifyinfo"%cdbName) - #tdLog.info("row: %d, %l64d, %l64d"%(tdSql.getData(0, 1),tdSql.getData(0, 2),tdSql.getData(0, 3)) - if (tdSql.getRows() == 1) and (tdSql.getData(0, 1) == 0): - break - else: - time.sleep(0.1) - return - - def getStartCommitNotifyFromTmqsim(self,cdbName='cdb'): - while 1: - tdSql.query("select * from %s.notifyinfo"%cdbName) - #tdLog.info("row: %d, %l64d, %l64d"%(tdSql.getData(0, 1),tdSql.getData(0, 2),tdSql.getData(0, 3)) - if tdSql.getRows() == 2 : - print(tdSql.getData(0, 1), tdSql.getData(1, 1)) - if tdSql.getData(1, 1) == 1: - break - time.sleep(0.1) - return + # def getStartConsumeNotifyFromTmqsim(self,cdbName='cdb'): + # while 1: + # tdSql.query("select * from %s.notifyinfo"%cdbName) + # #tdLog.info("row: %d, %l64d, %l64d"%(tdSql.getData(0, 1),tdSql.getData(0, 2),tdSql.getData(0, 3)) + # if (tdSql.getRows() == 1) and (tdSql.getData(0, 1) == 0): + # break + # else: + # time.sleep(0.1) + # return + # + # def getStartCommitNotifyFromTmqsim(self,cdbName='cdb'): + # while 1: + # tdSql.query("select * from %s.notifyinfo"%cdbName) + # #tdLog.info("row: %d, %l64d, %l64d"%(tdSql.getData(0, 1),tdSql.getData(0, 2),tdSql.getData(0, 3)) + # if tdSql.getRows() == 2 : + # print(tdSql.getData(0, 1), tdSql.getData(1, 1)) + # if tdSql.getData(1, 1) == 1: + # break + # time.sleep(0.1) + # return def create_database(self,tsql, dbName,dropFlag=1,vgroups=4,replica=1): if dropFlag == 1: diff --git a/tests/system-test/7-tmq/subscribeDb3.py b/tests/system-test/7-tmq/subscribeDb3.py index 1de9b62bcd..22004a95f1 100644 --- a/tests/system-test/7-tmq/subscribeDb3.py +++ b/tests/system-test/7-tmq/subscribeDb3.py @@ -10,6 +10,7 @@ from util.log import * from util.sql import * from util.cases import * from util.dnodes import * +from tmqCommon import * class TDTestCase: hostname = socket.gethostname() @@ -67,26 +68,26 @@ class TDTestCase: tdLog.info("consume info sql: %s"%sql) tdSql.query(sql) - def getStartConsumeNotifyFromTmqsim(self,cdbName='cdb'): - while 1: - tdSql.query("select * from %s.notifyinfo"%cdbName) - #tdLog.info("row: %d, %l64d, %l64d"%(tdSql.getData(0, 1),tdSql.getData(0, 2),tdSql.getData(0, 3)) - if (tdSql.getRows() == 1) and (tdSql.getData(0, 1) == 0): - break - else: - time.sleep(0.1) - return - - def getStartCommitNotifyFromTmqsim(self,cdbName='cdb'): - while 1: - tdSql.query("select * from %s.notifyinfo"%cdbName) - #tdLog.info("row: %d, %l64d, %l64d"%(tdSql.getData(0, 1),tdSql.getData(0, 2),tdSql.getData(0, 3)) - if tdSql.getRows() == 2 : - tdLog.info("row[0][1]: %d, row[1][1]: %d"%(tdSql.getData(0, 1), tdSql.getData(1, 1))) - if tdSql.getData(1, 1) == 1: - break - time.sleep(0.1) - return + # def getStartConsumeNotifyFromTmqsim(self,cdbName='cdb'): + # while 1: + # tdSql.query("select * from %s.notifyinfo"%cdbName) + # #tdLog.info("row: %d, %l64d, %l64d"%(tdSql.getData(0, 1),tdSql.getData(0, 2),tdSql.getData(0, 3)) + # if (tdSql.getRows() == 1) and (tdSql.getData(0, 1) == 0): + # break + # else: + # time.sleep(0.1) + # return + # + # def getStartCommitNotifyFromTmqsim(self,cdbName='cdb'): + # while 1: + # tdSql.query("select * from %s.notifyinfo"%cdbName) + # #tdLog.info("row: %d, %l64d, %l64d"%(tdSql.getData(0, 1),tdSql.getData(0, 2),tdSql.getData(0, 3)) + # if tdSql.getRows() == 2 : + # tdLog.info("row[0][1]: %d, row[1][1]: %d"%(tdSql.getData(0, 1), tdSql.getData(1, 1))) + # if tdSql.getData(1, 1) == 1: + # break + # time.sleep(0.1) + # return def selectConsumeResult(self,expectRows,cdbName='cdb'): resultList=[] @@ -233,7 +234,7 @@ class TDTestCase: self.startTmqSimProcess(buildPath,cfgPath,pollDelay,parameterDict["dbName"],showMsg, showRow) tdLog.info("wait the notify info of start consume") - self.getStartConsumeNotifyFromTmqsim() + tmqCom.getStartConsumeNotifyFromTmqsim() tdLog.info("pkill consume processor") if (platform.system().lower() == 'windows'): @@ -311,7 +312,7 @@ class TDTestCase: # time.sleep(6) tdLog.info("start to wait commit notify") - self.getStartCommitNotifyFromTmqsim() + tmqCom.getStartCommitNotifyFromTmqsim() tdLog.info("pkill consume processor") if (platform.system().lower() == 'windows'): diff --git a/tests/system-test/7-tmq/tmqCommon.py b/tests/system-test/7-tmq/tmqCommon.py index f63c70a4c6..44eef8bf24 100644 --- a/tests/system-test/7-tmq/tmqCommon.py +++ b/tests/system-test/7-tmq/tmqCommon.py @@ -145,31 +145,29 @@ class TMQCom: processID = subprocess.check_output(psCmd, shell=True).decode("utf-8") tdLog.debug("%s is stopped by kill -INT" % (processorName)) - def getStartConsumeNotifyFromTmqsim(self,cdbName='cdb',rows=1): + def getStartConsumeNotifyFromTmqsim(self,cdbName='cdb'): loopFlag = 1 while loopFlag: tdSql.query("select * from %s.notifyinfo"%cdbName) #tdLog.info("row: %d, %l64d, %l64d"%(tdSql.getData(0, 1),tdSql.getData(0, 2),tdSql.getData(0, 3)) actRows = tdSql.getRows() - if (actRows >= rows): - for i in range(actRows): - if tdSql.getData(i, 1) == 0: - loopFlag = 0 - break + for i in range(actRows): + if tdSql.getData(i, 1) == 0: + loopFlag = 0 + break time.sleep(0.02) return - def getStartCommitNotifyFromTmqsim(self,cdbName='cdb',rows=2): + def getStartCommitNotifyFromTmqsim(self,cdbName='cdb'): loopFlag = 1 while loopFlag: tdSql.query("select * from %s.notifyinfo"%cdbName) #tdLog.info("row: %d, %l64d, %l64d"%(tdSql.getData(0, 1),tdSql.getData(0, 2),tdSql.getData(0, 3)) actRows = tdSql.getRows() - if (actRows >= rows): - for i in range(actRows): - if tdSql.getData(i, 1) == 1: - loopFlag = 0 - break + for i in range(actRows): + if tdSql.getData(i, 1) == 1: + loopFlag = 0 + break time.sleep(0.02) return diff --git a/tests/system-test/7-tmq/tmqConsumerGroup.py b/tests/system-test/7-tmq/tmqConsumerGroup.py index f05f600f27..ae9671bcf4 100644 --- a/tests/system-test/7-tmq/tmqConsumerGroup.py +++ b/tests/system-test/7-tmq/tmqConsumerGroup.py @@ -100,7 +100,7 @@ class TDTestCase: tdLog.info("wait consumer commit notify") # tmqCom.getStartCommitNotifyFromTmqsim(rows=4) - tmqCom.getStartConsumeNotifyFromTmqsim(rows=2) + tmqCom.getStartConsumeNotifyFromTmqsim() tdLog.info("pkill one consume processor") tmqCom.stopTmqSimProcess('tmq_sim_new') diff --git a/tests/system-test/7-tmq/tmqDnodeRestart1.py b/tests/system-test/7-tmq/tmqDnodeRestart1.py index cff55a1239..2bde32800b 100644 --- a/tests/system-test/7-tmq/tmqDnodeRestart1.py +++ b/tests/system-test/7-tmq/tmqDnodeRestart1.py @@ -121,7 +121,7 @@ class TDTestCase: tmqCom.startTmqSimProcess(pollDelay=paraDict['pollDelay'],dbName=paraDict["dbName"],showMsg=paraDict['showMsg'], showRow=paraDict['showRow'],snapshot=paraDict['snapshot']) # time.sleep(3) - tmqCom.getStartCommitNotifyFromTmqsim('cdb',1) + tmqCom.getStartCommitNotifyFromTmqsim() tdLog.info("create some new child table and insert data for latest mode") paraDict["batchNum"] = 100 @@ -205,7 +205,7 @@ class TDTestCase: tdLog.info("start consume processor") tmqCom.startTmqSimProcess(pollDelay=paraDict['pollDelay'],dbName=paraDict["dbName"],showMsg=paraDict['showMsg'], showRow=paraDict['showRow'],snapshot=paraDict['snapshot']) - tmqCom.getStartCommitNotifyFromTmqsim('cdb',1) + tmqCom.getStartCommitNotifyFromTmqsim() tdLog.info("create some new child table and insert data for latest mode") paraDict["batchNum"] = 10