Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
taosdata
TDengine
提交
97a958fc
T
TDengine
项目概览
taosdata
/
TDengine
大约 1 年 前同步成功
通知
1184
Star
22015
Fork
4786
代码
文件
提交
分支
Tags
贡献者
分支图
Diff
Issue
1
列表
看板
标记
里程碑
合并请求
0
Wiki
0
Wiki
分析
仓库
DevOps
项目成员
Pages
T
TDengine
项目概览
项目概览
详情
发布
仓库
仓库
文件
提交
分支
标签
贡献者
分支图
比较
Issue
1
Issue
1
列表
看板
标记
里程碑
合并请求
0
合并请求
0
Pages
分析
分析
仓库分析
DevOps
Wiki
0
Wiki
成员
成员
收起侧边栏
关闭侧边栏
动态
分支图
创建新Issue
提交
Issue看板
体验新版 GitCode,发现更多精彩内容 >>
未验证
提交
97a958fc
编写于
7月 02, 2022
作者:
H
Hui Li
提交者:
GitHub
7月 02, 2022
浏览文件
操作
浏览文件
下载
差异文件
Merge pull request #14451 from taosdata/test3.0/lihui
test:add test case for tmq
上级
73b1101a
b9cc3e6e
变更
3
隐藏空白更改
内联
并排
Showing
3 changed file
with
108 addition
and
5 deletion
+108
-5
tests/pytest/util/common.py
tests/pytest/util/common.py
+6
-0
tests/system-test/7-tmq/stbFilter.py
tests/system-test/7-tmq/stbFilter.py
+3
-3
tests/system-test/7-tmq/tmqConsFromTsdb.py
tests/system-test/7-tmq/tmqConsFromTsdb.py
+99
-2
未找到文件。
tests/pytest/util/common.py
浏览文件 @
97a958fc
...
...
@@ -662,6 +662,12 @@ class TDCom:
return
res_list
else
:
tdLog
.
exit
(
f
"getOneRow out of range: row_index=
{
location
}
row_count=
{
self
.
query_row
}
"
)
def
killProcessor
(
self
,
processorName
):
if
(
platform
.
system
().
lower
()
==
'windows'
):
os
.
system
(
"TASKKILL /F /IM %s.exe"
%
processorName
)
else
:
os
.
system
(
'pkill %s'
%
processorName
)
def
is_json
(
msg
):
...
...
tests/system-test/7-tmq/stbFilter.py
浏览文件 @
97a958fc
...
...
@@ -122,9 +122,9 @@ class TDTestCase:
tdLog
.
info
(
"wait the consume result"
)
expectRows
=
1
resultList
=
tmqCom
.
selectConsumeResult
(
expectRows
)
#
if expectRowsList[2] != resultList[0]:
#
tdLog.info("expect consume rows: %d, act consume rows: %d"%(expectRowsList[2], resultList[0]))
#
tdLog.exit("2 tmq consume rows error!")
if
expectRowsList
[
2
]
!=
resultList
[
0
]:
tdLog
.
info
(
"expect consume rows: %d, act consume rows: %d"
%
(
expectRowsList
[
2
],
resultList
[
0
]))
tdLog
.
exit
(
"2 tmq consume rows error!"
)
time
.
sleep
(
10
)
for
i
in
range
(
len
(
topicNameList
)):
...
...
tests/system-test/7-tmq/tmqConsFromTsdb.py
浏览文件 @
97a958fc
...
...
@@ -16,7 +16,7 @@ sys.path.append("./7-tmq")
from
tmqCommon
import
*
class
TDTestCase
:
def
__int__
(
self
):
def
__in
i
t__
(
self
):
self
.
vgroups
=
1
self
.
ctbNum
=
10
self
.
rowsPerTbl
=
10000
...
...
@@ -216,7 +216,7 @@ class TDTestCase:
'rowsPerTbl'
:
10000
,
'batchNum'
:
10
,
'startTs'
:
1640966400000
,
# 2022-01-01 00:00:00.000
'pollDelay'
:
-
1
,
'pollDelay'
:
10
,
'showMsg'
:
1
,
'showRow'
:
1
,
'snapshot'
:
1
}
...
...
@@ -281,10 +281,107 @@ class TDTestCase:
tdLog
.
printNoPrefix
(
"======== test case 3 end ...... "
)
def
tmqCase4
(
self
):
tdLog
.
printNoPrefix
(
"======== test case 4: "
)
paraDict
=
{
'dbName'
:
'dbt'
,
'dropFlag'
:
1
,
'event'
:
''
,
'vgroups'
:
1
,
'stbName'
:
'stb'
,
'colPrefix'
:
'c'
,
'tagPrefix'
:
't'
,
'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'
:
1
,
'rowsPerTbl'
:
10000
,
'batchNum'
:
10
,
'startTs'
:
1640966400000
,
# 2022-01-01 00:00:00.000
'pollDelay'
:
10
,
'showMsg'
:
1
,
'showRow'
:
1
,
'snapshot'
:
1
}
paraDict
[
'vgroups'
]
=
self
.
vgroups
paraDict
[
'ctbNum'
]
=
self
.
ctbNum
paraDict
[
'rowsPerTbl'
]
=
self
.
rowsPerTbl
topicNameList
=
[
'topic1'
]
expectRowsList
=
[]
tmqCom
.
initConsumerTable
()
# tdCom.create_database(tdSql, paraDict["dbName"],paraDict["dropFlag"], vgroups=paraDict["vgroups"],replica=1)
# tdLog.info("create stb")
# tdCom.create_stable(tdSql, dbname=paraDict["dbName"],stbname=paraDict["stbName"], column_elm_list=paraDict['colSchema'], tag_elm_list=paraDict['tagSchema'])
# tdLog.info("create ctb")
# tdCom.create_ctable(tdSql, dbname=paraDict["dbName"],stbname=paraDict["stbName"],tag_elm_list=paraDict['tagSchema'],count=paraDict["ctbNum"], default_ctbname_prefix=paraDict['ctbPrefix'])
# tdLog.info("insert data")
# tmqCom.insert_data(tdSql,paraDict["dbName"],paraDict["ctbPrefix"],paraDict["ctbNum"],paraDict["rowsPerTbl"],paraDict["batchNum"],paraDict["startTs"])
# tdDnodes.stop(1)
# tdDnodes.start(1)
tdLog
.
info
(
"create topics from stb with filter"
)
queryString
=
"select * from %s.%s"
%
(
paraDict
[
'dbName'
],
paraDict
[
'stbName'
])
# sqlString = "create topic %s as stable %s" %(topicNameList[0], paraDict['stbName'])
sqlString
=
"create topic %s as %s"
%
(
topicNameList
[
0
],
queryString
)
tdLog
.
info
(
"create topic sql: %s"
%
sqlString
)
tdSql
.
execute
(
sqlString
)
tdSql
.
query
(
queryString
)
expectRowsList
.
append
(
tdSql
.
getRows
())
totalRowsInserted
=
expectRowsList
[
0
]
# init consume info, and start tmq_sim, then check consume result
tdLog
.
info
(
"insert consume info to consume processor"
)
consumerId
=
5
expectrowcnt
=
math
.
ceil
(
paraDict
[
"rowsPerTbl"
]
*
paraDict
[
"ctbNum"
])
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 commit notify"
)
tmqCom
.
getStartCommitNotifyFromTmqsim
()
tdLog
.
info
(
"pkill consume processor"
)
tdCom
.
killProcessor
(
"tmq_sim"
)
# time.sleep(10)
# reinit consume info, and start tmq_sim, then check consume result
tmqCom
.
initConsumerTable
()
consumerId
=
6
tmqCom
.
insertConsumerInfo
(
consumerId
,
expectrowcnt
,
topicList
,
keyList
,
ifcheckdata
,
ifManualCommit
)
tdLog
.
info
(
"start consume processor 1"
)
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
)
actConsumeTotalRows
=
resultList
[
0
]
if
not
(
actConsumeTotalRows
>
0
and
actConsumeTotalRows
<
totalRowsInserted
):
tdLog
.
info
(
"act consume rows: %d"
%
(
actConsumeTotalRows
))
tdLog
.
info
(
"and second consume rows should be between 0 and %d"
%
(
totalRowsInserted
))
tdLog
.
exit
(
"%d tmq consume rows error!"
%
consumerId
)
time
.
sleep
(
10
)
for
i
in
range
(
len
(
topicNameList
)):
tdSql
.
query
(
"drop topic %s"
%
topicNameList
[
i
])
tdLog
.
printNoPrefix
(
"======== test case 4 end ...... "
)
def
run
(
self
):
tdSql
.
prepare
()
self
.
tmqCase1
()
self
.
tmqCase2
()
self
.
tmqCase3
()
self
.
tmqCase4
()
def
stop
(
self
):
tdSql
.
close
()
...
...
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录