test: modify test case

tests/system-test/7-tmq/subscribeDb2.py

@@ -5,6 +5,7 @@ import time
 import socket
 import os
 import threading
+import math
 
 from util.log import *
 from util.sql import *
@@ -127,10 +128,14 @@ class TDTestCase:
         #tdLog.debug("doing insert data into stable:%s rows:%d ..."%(stbName, allRows))
         for i in range(ctbNum):
             sql += " %s_%d values "%(stbName,i)
+            batchRows = 0
             for j in range(rowsPerTbl):
                 sql += "(%d, %d, 'tmqrow_%d') "%(startTs + j, j, j)
-                if (j > 0) and ((j%batchNum == 0) or (j == rowsPerTbl - 1)):
+                batchRows += 1
+                # if (j > 0) and ((j%(batchNum-1) == 0) or (j == rowsPerTbl - 1)):
+                if (j > 0) and ((batchRows == batchNum) or (j == rowsPerTbl - 1)):
                     tsql.execute(sql)
+                    batchRows = 0
                     if j < rowsPerTbl - 1:
                         sql = "insert into %s_%d values " %(stbName,i)
                     else:
@@ -171,8 +176,8 @@ class TDTestCase:
                          'dbName':     'db8',    \
                          'vgroups':    4,        \
                          'stbName':    'stb',    \
-                         'ctbNum':     10,       \
-                         'rowsPerTbl': 10000,    \
+                         'ctbNum':     1,       \
+                         'rowsPerTbl': 1000,    \
                          'batchNum':   100,      \
                          'startTs':    1640966400000}  # 2022-01-01 00:00:00.000
         parameterDict['cfg'] = cfgPath
@@ -189,7 +194,7 @@ class TDTestCase:
         
         tdSql.execute("create topic %s as database %s" %(topicName1, parameterDict['dbName']))
         consumerId   = 0
-        expectrowcnt = parameterDict["rowsPerTbl"] * parameterDict["ctbNum"] / 2
+        expectrowcnt = math.ceil(parameterDict["rowsPerTbl"] * parameterDict["ctbNum"] / 2)
         topicList    = topicName1
         ifcheckdata  = 0
         ifManualCommit = 0
@@ -217,7 +222,7 @@ class TDTestCase:
         for i in range(expectRows):
             totalConsumeRows += resultList[i]
         
-        if totalConsumeRows != expectrowcnt:
+        if not (totalConsumeRows >= expectrowcnt):
             tdLog.info("act consume rows: %d, expect consume rows: %d"%(totalConsumeRows, expectrowcnt))
             tdLog.exit("tmq consume rows error!")
 
@@ -267,7 +272,7 @@ class TDTestCase:
         
         tdSql.execute("create topic %s as database %s" %(topicName1, parameterDict['dbName']))
         consumerId   = 0
-        expectrowcnt = parameterDict["rowsPerTbl"] * parameterDict["ctbNum"] / 2
+        expectrowcnt = math.ceil(parameterDict["rowsPerTbl"] * parameterDict["ctbNum"] / 2)
         topicList    = topicName1
         ifcheckdata  = 0
         ifManualCommit = 1

tests/system-test/7-tmq/tmqConsFromTsdb.py

@@ -5,6 +5,7 @@ import time
 import socket
 import os
 import threading
+import math
 
 from util.log import *
 from util.sql import *
@@ -15,6 +16,11 @@ sys.path.append("./7-tmq")
 from tmqCommon import *
 
 class TDTestCase:
+    def __int__(self):
+        self.vgroups    = 1
+        self.ctbNum     = 10
+        self.rowsPerTbl = 10000
+        
     def init(self, conn, logSql):
         tdLog.debug(f"start to excute {__file__}")
         tdSql.init(conn.cursor(), False)
@@ -40,6 +46,10 @@ class TDTestCase:
                     'showRow':    1,
                     'snapshot':   1}
 
+        paraDict['vgroups'] = self.vgroups
+        paraDict['ctbNum'] = self.ctbNum
+        paraDict['rowsPerTbl'] = self.rowsPerTbl
+        
         topicNameList = ['topic1']
         expectRowsList = []
         tmqCom.initConsumerTable()
@@ -113,6 +123,10 @@ class TDTestCase:
                     'showRow':    1,
                     'snapshot':   1}
 
+        paraDict['vgroups'] = self.vgroups
+        paraDict['ctbNum'] = self.ctbNum
+        paraDict['rowsPerTbl'] = self.rowsPerTbl
+        
         topicNameList = ['topic1']
         expectRowsList = []
         tmqCom.initConsumerTable()
@@ -140,7 +154,7 @@ class TDTestCase:
         # init consume info, and start tmq_sim, then check consume result
         tdLog.info("insert consume info to consume processor")
         consumerId   = 1
-        expectrowcnt = paraDict["rowsPerTbl"] * (paraDict["ctbNum"] - 7)
+        expectrowcnt = math.ceil(paraDict["rowsPerTbl"] * paraDict["ctbNum"] / 3)
         topicList    = topicNameList[0]
         ifcheckdata  = 1
         ifManualCommit = 1
@@ -163,7 +177,7 @@ class TDTestCase:
         # reinit consume info, and start tmq_sim, then check consume result
         tmqCom.initConsumerTable()
         consumerId   = 2
-        expectrowcnt = paraDict["rowsPerTbl"] * (paraDict["ctbNum"] - 3)
+        expectrowcnt = math.ceil(paraDict["rowsPerTbl"] * paraDict["ctbNum"] * 2/3)
         tmqCom.insertConsumerInfo(consumerId, expectrowcnt,topicList,keyList,ifcheckdata,ifManualCommit)
         
         tdLog.info("start consume processor 1")
@@ -176,8 +190,8 @@ class TDTestCase:
         actConsumeTotalRows = firstConsumeRows + resultList[0]
         
         if not (expectrowcnt >= resultList[0] and totalRowsInserted == actConsumeTotalRows):
-            tdLog.info("act consume rows: %d, expect consume rows <= %d "%(resultList[0], expectrowcnt))
-            tdLog.info("and sum of two consume rows: %d , total inserted rows: %d"%(actConsumeTotalRows, totalRowsInserted))
+            tdLog.info("act consume rows, first: %d, second: %d "%(firstConsumeRows, resultList[0]))            
+            tdLog.info("and sum of two consume rows: %d should be equal to total inserted rows: %d"%(actConsumeTotalRows, totalRowsInserted))
             tdLog.exit("%d tmq consume rows error!"%consumerId)
 
         time.sleep(10)        
@@ -198,15 +212,19 @@ class TDTestCase:
                     'colSchema':   [{'type': 'INT', 'count':1}, {'type': 'binary', 'len':20, 'count':1}],
                     'tagSchema':   [{'type': 'INT', 'count':1}, {'type': 'binary', 'len':20, 'count':1}],
                     'ctbPrefix':  'ctb',
-                    'ctbNum':     10,
+                    'ctbNum':     1,
                     'rowsPerTbl': 10000,
                     'batchNum':   10,
                     'startTs':    1640966400000,  # 2022-01-01 00:00:00.000
-                    'pollDelay':  3,
+                    'pollDelay':  -1,
                     'showMsg':    1,
                     'showRow':    1,
                     'snapshot':   1}
 
+        paraDict['vgroups'] = self.vgroups
+        paraDict['ctbNum'] = self.ctbNum
+        paraDict['rowsPerTbl'] = self.rowsPerTbl
+        
         topicNameList = ['topic1']
         expectRowsList = []
         tmqCom.initConsumerTable()
@@ -229,49 +247,33 @@ class TDTestCase:
         tdSql.execute(sqlString)
         tdSql.query(queryString)        
         expectRowsList.append(tdSql.getRows())
-        totalRowsInserted = expectrowcnt = paraDict["rowsPerTbl"] * paraDict["ctbNum"]
+        totalRowsInserted = expectRowsList[0]
 
         # init consume info, and start tmq_sim, then check consume result
         tdLog.info("insert consume info to consume processor")
         consumerId   = 3
-        expectrowcnt = paraDict["rowsPerTbl"] * (paraDict["ctbNum"] - 7)
+        expectrowcnt = math.ceil(paraDict["rowsPerTbl"] * paraDict["ctbNum"] / 3)
         topicList    = topicNameList[0]
         ifcheckdata  = 1
         ifManualCommit = 1
         keyList      = 'group.id:cgrp1, enable.auto.commit:true, auto.commit.interval.ms:1000, auto.offset.reset:earliest'
         tmqCom.insertConsumerInfo(consumerId, expectrowcnt,topicList,keyList,ifcheckdata,ifManualCommit)
 
-        tdLog.info("start consume processor 0")
-        tmqCom.startTmqSimProcess(pollDelay=paraDict['pollDelay'],dbName=paraDict["dbName"],showMsg=paraDict['showMsg'], showRow=paraDict['showRow'],snapshot=paraDict['snapshot'])
-        tdLog.info("wait the consume result") 
-        
-        expectRows = 1
-        resultList = tmqCom.selectConsumeResult(expectRows)
-        
-        if not (expectrowcnt <= resultList[0] and totalRowsInserted >= resultList[0]):
-            tdLog.info("act consume rows: %d, expect consume rows between %d and %d"%(resultList[0], expectrowcnt, totalRowsInserted))
-            tdLog.exit("0 tmq consume rows error!")
-            
-        firstConsumeRows = resultList[0]
-
-        # reinit consume info, and start tmq_sim, then check consume result
-        tmqCom.initConsumerTable()
         consumerId   = 4
-        expectrowcnt = paraDict["rowsPerTbl"] * (paraDict["ctbNum"] - 3)
+        expectrowcnt = math.ceil(paraDict["rowsPerTbl"] * paraDict["ctbNum"] * 2/3)
         tmqCom.insertConsumerInfo(consumerId, expectrowcnt,topicList,keyList,ifcheckdata,ifManualCommit)
-        
-        tdLog.info("start consume processor 1")
+
+        tdLog.info("start consume processor 0")
         tmqCom.startTmqSimProcess(pollDelay=paraDict['pollDelay'],dbName=paraDict["dbName"],showMsg=paraDict['showMsg'], showRow=paraDict['showRow'],snapshot=paraDict['snapshot'])
         tdLog.info("wait the consume result") 
         
-        expectRows = 1
+        expectRows = 2
         resultList = tmqCom.selectConsumeResult(expectRows)
+        actConsumeTotalRows = resultList[0] + resultList[1]
         
-        actConsumeTotalRows = firstConsumeRows + resultList[0]
-        
-        if not (expectrowcnt >= resultList[0] and totalRowsInserted == actConsumeTotalRows):
-            tdLog.info("act consume rows: %d, expect consume rows between %d and %d"%(resultList[0], expectrowcnt, totalRowsInserted))
-            tdLog.exit("0 tmq consume rows error!")
+        if not (totalRowsInserted == actConsumeTotalRows):
+            tdLog.info("sum of two consume rows: %d should be equal to total inserted rows: %d"%(actConsumeTotalRows, totalRowsInserted))
+            tdLog.exit("%d tmq consume rows error!"%consumerId)
 
         time.sleep(10)        
         for i in range(len(topicNameList)):