Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
taosdata
TDengine
提交
c55d20b1
T
TDengine
项目概览
taosdata
/
TDengine
大约 1 年 前同步成功
通知
1181
Star
22014
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,发现更多精彩内容 >>
提交
c55d20b1
编写于
5月 12, 2022
作者:
P
plum-lihui
浏览文件
操作
浏览文件
下载
电子邮件补丁
差异文件
test: error script for debug
上级
d538f9b8
变更
1
隐藏空白更改
内联
并排
Showing
1 changed file
with
372 addition
and
0 deletion
+372
-0
tests/system-test/99-TDcase/TD-15517.py
tests/system-test/99-TDcase/TD-15517.py
+372
-0
未找到文件。
tests/system-test/99-TDcase/TD-15517.py
0 → 100644
浏览文件 @
c55d20b1
import
taos
import
sys
import
time
import
socket
import
os
import
threading
from
util.log
import
*
from
util.sql
import
*
from
util.cases
import
*
from
util.dnodes
import
*
class
TDTestCase
:
hostname
=
socket
.
gethostname
()
#rpcDebugFlagVal = '143'
#clientCfgDict = {'serverPort': '', 'firstEp': '', 'secondEp':'', 'rpcDebugFlag':'135', 'fqdn':''}
#clientCfgDict["rpcDebugFlag"] = rpcDebugFlagVal
#updatecfgDict = {'clientCfg': {}, 'serverPort': '', 'firstEp': '', 'secondEp':'', 'rpcDebugFlag':'135', 'fqdn':''}
#updatecfgDict["rpcDebugFlag"] = rpcDebugFlagVal
#print ("===================: ", updatecfgDict)
def
init
(
self
,
conn
,
logSql
):
tdLog
.
debug
(
f
"start to excute
{
__file__
}
"
)
#tdSql.init(conn.cursor())
tdSql
.
init
(
conn
.
cursor
(),
logSql
)
# output sql.txt file
def
getBuildPath
(
self
):
selfPath
=
os
.
path
.
dirname
(
os
.
path
.
realpath
(
__file__
))
if
(
"community"
in
selfPath
):
projPath
=
selfPath
[:
selfPath
.
find
(
"community"
)]
else
:
projPath
=
selfPath
[:
selfPath
.
find
(
"tests"
)]
for
root
,
dirs
,
files
in
os
.
walk
(
projPath
):
if
(
"taosd"
in
files
):
rootRealPath
=
os
.
path
.
dirname
(
os
.
path
.
realpath
(
root
))
if
(
"packaging"
not
in
rootRealPath
):
buildPath
=
root
[:
len
(
root
)
-
len
(
"/build/bin"
)]
break
return
buildPath
def
newcur
(
self
,
cfg
,
host
,
port
):
user
=
"root"
password
=
"taosdata"
con
=
taos
.
connect
(
host
=
host
,
user
=
user
,
password
=
password
,
config
=
cfg
,
port
=
port
)
cur
=
con
.
cursor
()
print
(
cur
)
return
cur
def
create_tables
(
self
,
tsql
,
dbName
,
vgroups
,
stbName
,
ctbNum
,
rowsPerTbl
):
tsql
.
execute
(
"create database if not exists %s vgroups %d"
%
(
dbName
,
vgroups
))
tsql
.
execute
(
"use %s"
%
dbName
)
tsql
.
execute
(
"create table %s (ts timestamp, c1 bigint, c2 binary(16)) tags(t1 int)"
%
stbName
)
pre_create
=
"create table"
sql
=
pre_create
#tdLog.debug("doing create one stable %s and %d child table in %s ..." %(stbname, count ,dbname))
for
i
in
range
(
ctbNum
):
sql
+=
" %s_%d using %s tags(%d)"
%
(
stbName
,
i
,
stbName
,
i
+
1
)
if
(
i
>
0
)
and
(
i
%
100
==
0
):
tsql
.
execute
(
sql
)
sql
=
pre_create
if
sql
!=
pre_create
:
tsql
.
execute
(
sql
)
tdLog
.
debug
(
"complete to create database[%s], stable[%s] and %d child tables"
%
(
dbName
,
stbName
,
ctbNum
))
return
def
insert_data
(
self
,
tsql
,
dbName
,
stbName
,
ctbNum
,
rowsPerTbl
,
batchNum
,
startTs
):
tdLog
.
debug
(
"start to insert data ............"
)
tsql
.
execute
(
"use %s"
%
dbName
)
pre_insert
=
"insert into "
sql
=
pre_insert
#tdLog.debug("doing insert data into stable:%s rows:%d ..."%(stbName, allRows))
for
i
in
range
(
ctbNum
):
sql
+=
" %s_%d values "
%
(
stbName
,
i
)
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
)):
tsql
.
execute
(
sql
)
if
j
<
rowsPerTbl
-
1
:
sql
=
"insert into %s_%d values "
%
(
stbName
,
i
)
else
:
sql
=
"insert into "
#end sql
if
sql
!=
pre_insert
:
#print("insert sql:%s"%sql)
tsql
.
execute
(
sql
)
tdLog
.
debug
(
"insert data ............ [OK]"
)
return
def
prepareEnv
(
self
,
**
parameterDict
):
print
(
"input parameters:"
)
print
(
parameterDict
)
# create new connector for my thread
tsql
=
self
.
newcur
(
parameterDict
[
'cfg'
],
'localhost'
,
6030
)
self
.
create_tables
(
tsql
,
\
parameterDict
[
"dbName"
],
\
parameterDict
[
"vgroups"
],
\
parameterDict
[
"stbName"
],
\
parameterDict
[
"ctbNum"
],
\
parameterDict
[
"rowsPerTbl"
])
self
.
insert_data
(
tsql
,
\
parameterDict
[
"dbName"
],
\
parameterDict
[
"stbName"
],
\
parameterDict
[
"ctbNum"
],
\
parameterDict
[
"rowsPerTbl"
],
\
parameterDict
[
"batchNum"
],
\
parameterDict
[
"startTs"
])
return
def
run
(
self
):
tdSql
.
prepare
()
buildPath
=
self
.
getBuildPath
()
if
(
buildPath
==
""
):
tdLog
.
exit
(
"taosd not found!"
)
else
:
tdLog
.
info
(
"taosd found in %s"
%
buildPath
)
cfgPath
=
buildPath
+
"/../sim/psim/cfg"
tdLog
.
info
(
"cfgPath: %s"
%
cfgPath
)
tdLog
.
printNoPrefix
(
"======== test scenario 1: "
)
tdLog
.
info
(
"step 1: create database, stb, ctb and insert data"
)
# create and start thread
parameterDict
=
{
'cfg'
:
''
,
\
'dbName'
:
'db'
,
\
'vgroups'
:
1
,
\
'stbName'
:
'stb'
,
\
'ctbNum'
:
10
,
\
'rowsPerTbl'
:
10000
,
\
'batchNum'
:
10
,
\
'startTs'
:
1640966400000
}
# 2022-01-01 00:00:00.000
parameterDict
[
'cfg'
]
=
cfgPath
prepareEnvThread
=
threading
.
Thread
(
target
=
self
.
prepareEnv
,
kwargs
=
parameterDict
)
prepareEnvThread
.
start
()
time
.
sleep
(
2
)
# wait stb ready
while
1
:
tdSql
.
query
(
"show %s.stables"
%
parameterDict
[
'dbName'
])
if
tdSql
.
getRows
()
==
1
:
break
else
:
time
.
sleep
(
1
)
tdLog
.
info
(
"create topics from super table"
)
topicFromStb
=
'topic_stb_column'
topicFromCtb
=
'topic_ctb_column'
tdSql
.
execute
(
"create topic %s as select ts, c1, c2 from %s.%s"
%
(
topicFromStb
,
parameterDict
[
'dbName'
],
parameterDict
[
'stbName'
]))
tdSql
.
execute
(
"create topic %s as select ts, c1, c2 from %s.%s_0"
%
(
topicFromCtb
,
parameterDict
[
'dbName'
],
parameterDict
[
'stbName'
]))
time
.
sleep
(
1
)
tdSql
.
query
(
"show topics"
)
#tdSql.checkRows(2)
topic1
=
tdSql
.
getData
(
0
,
0
)
topic2
=
tdSql
.
getData
(
1
,
0
)
print
(
topic1
)
print
(
topic2
)
print
(
topicFromStb
)
print
(
topicFromCtb
)
#tdLog.info("show topics: %s, %s"%topic1, topic2)
#if topic1 != topicFromStb or topic1 != topicFromCtb:
# tdLog.exit("topic error1")
#if topic2 != topicFromStb or topic2 != topicFromCtb:
# tdLog.exit("topic error2")
tdLog
.
info
(
"create consume info table and consume result table"
)
cdbName
=
parameterDict
[
"dbName"
]
tdSql
.
query
(
"create table consumeinfo (ts timestamp, consumerid int, topiclist binary(1024), keylist binary(1024), expectmsgcnt bigint, ifcheckdata int)"
)
tdSql
.
query
(
"create table consumeresult (ts timestamp, consumerid int, consummsgcnt bigint, consumrowcnt bigint, checkresult int)"
)
consumerId
=
0
expectmsgcnt
=
(
parameterDict
[
"rowsPerTbl"
]
/
parameterDict
[
"batchNum"
]
)
*
parameterDict
[
"ctbNum"
]
expectmsgcnt1
=
expectmsgcnt
+
parameterDict
[
"ctbNum"
]
topicList
=
topicFromStb
ifcheckdata
=
0
keyList
=
'group.id:cgrp1,\
enable.auto.commit:false,\
auto.commit.interval.ms:6000,\
auto.offset.reset:earliest'
sql
=
"insert into consumeinfo values "
sql
+=
"(now, %d, '%s', '%s', %d, %d)"
%
(
consumerId
,
topicList
,
keyList
,
expectmsgcnt1
,
ifcheckdata
)
tdSql
.
query
(
sql
)
tdLog
.
info
(
"check stb if there are data"
)
while
1
:
tdSql
.
query
(
"select count(*) from %s"
%
parameterDict
[
"stbName"
])
#tdLog.info("row: %d, %l64d, %l64d"%(tdSql.getData(0, 1),tdSql.getData(0, 2),tdSql.getData(0, 3))
countOfStb
=
tdSql
.
getData
(
0
,
0
)
if
countOfStb
!=
0
:
tdLog
.
info
(
"count from stb: %d"
%
countOfStb
)
break
else
:
time
.
sleep
(
1
)
tdLog
.
info
(
"start consume processor"
)
pollDelay
=
5
showMsg
=
1
showRow
=
1
shellCmd
=
'nohup '
+
buildPath
+
'/build/bin/tmq_sim -c '
+
cfgPath
shellCmd
+=
" -y %d -d %s -g %d -r %d -w %s "
%
(
pollDelay
,
parameterDict
[
"dbName"
],
showMsg
,
showRow
,
cdbName
)
shellCmd
+=
"> /dev/null 2>&1 &"
tdLog
.
info
(
shellCmd
)
os
.
system
(
shellCmd
)
# wait for data ready
prepareEnvThread
.
join
()
tdLog
.
info
(
"insert process end, and start to check consume result"
)
while
1
:
tdSql
.
query
(
"select * from consumeresult"
)
#tdLog.info("row: %d, %l64d, %l64d"%(tdSql.getData(0, 1),tdSql.getData(0, 2),tdSql.getData(0, 3))
if
tdSql
.
getRows
()
==
1
:
break
else
:
time
.
sleep
(
5
)
expectrowcnt
=
parameterDict
[
"rowsPerTbl"
]
*
parameterDict
[
"ctbNum"
]
tdSql
.
checkData
(
0
,
1
,
consumerId
)
tdSql
.
checkData
(
0
,
2
,
expectmsgcnt
)
tdSql
.
checkData
(
0
,
3
,
expectrowcnt
)
tdSql
.
query
(
"drop topic %s"
%
topicFromStb
)
tdSql
.
query
(
"drop topic %s"
%
topicFromCtb
)
# ==============================================================================
tdLog
.
printNoPrefix
(
"======== test scenario 2: add child table with consuming "
)
tdLog
.
info
(
" clean database"
)
# create and start thread
parameterDict
=
{
'cfg'
:
''
,
\
'dbName'
:
'db2'
,
\
'vgroups'
:
1
,
\
'stbName'
:
'stb'
,
\
'ctbNum'
:
10
,
\
'rowsPerTbl'
:
10000
,
\
'batchNum'
:
100
,
\
'startTs'
:
1640966400000
}
# 2022-01-01 00:00:00.000
parameterDict
[
'cfg'
]
=
cfgPath
prepareEnvThread
=
threading
.
Thread
(
target
=
self
.
prepareEnv
,
kwargs
=
parameterDict
)
prepareEnvThread
.
start
()
# wait db ready
while
1
:
tdSql
.
query
(
"show databases"
)
if
tdSql
.
getRows
()
==
3
:
print
(
tdSql
.
getData
(
0
,
0
),
tdSql
.
getData
(
1
,
0
),
tdSql
.
getData
(
2
,
0
),)
break
else
:
time
.
sleep
(
1
)
tdSql
.
query
(
"use %s"
%
parameterDict
[
'dbName'
])
# wait stb ready
while
1
:
tdSql
.
query
(
"show %s.stables"
%
parameterDict
[
'dbName'
])
if
tdSql
.
getRows
()
==
1
:
break
else
:
time
.
sleep
(
1
)
tdLog
.
info
(
"create topics from super table"
)
topicFromStb
=
'topic_stb_column2'
topicFromCtb
=
'topic_ctb_column2'
tdSql
.
execute
(
"create topic %s as select ts, c1, c2 from %s.%s"
%
(
topicFromStb
,
parameterDict
[
'dbName'
],
parameterDict
[
'stbName'
]))
tdSql
.
execute
(
"create topic %s as select ts, c1, c2 from %s.%s_0"
%
(
topicFromCtb
,
parameterDict
[
'dbName'
],
parameterDict
[
'stbName'
]))
time
.
sleep
(
1
)
tdSql
.
query
(
"show topics"
)
topic1
=
tdSql
.
getData
(
0
,
0
)
topic2
=
tdSql
.
getData
(
1
,
0
)
print
(
topic1
)
print
(
topic2
)
print
(
topicFromStb
)
print
(
topicFromCtb
)
#tdLog.info("show topics: %s, %s"%topic1, topic2)
#if topic1 != topicFromStb or topic1 != topicFromCtb:
# tdLog.exit("topic error1")
#if topic2 != topicFromStb or topic2 != topicFromCtb:
# tdLog.exit("topic error2")
tdLog
.
info
(
"create consume info table and consume result table"
)
cdbName
=
parameterDict
[
"dbName"
]
tdSql
.
query
(
"create table %s.consumeinfo (ts timestamp, consumerid int, topiclist binary(1024), keylist binary(1024), expectmsgcnt bigint, ifcheckdata int)"
%
cdbName
)
tdSql
.
query
(
"create table %s.consumeresult (ts timestamp, consumerid int, consummsgcnt bigint, consumrowcnt bigint, checkresult int)"
%
cdbName
)
consumerId
=
0
expectmsgcnt
=
(
parameterDict
[
"rowsPerTbl"
]
/
parameterDict
[
"batchNum"
]
)
*
parameterDict
[
"ctbNum"
]
expectmsgcnt1
=
expectmsgcnt
+
parameterDict
[
"ctbNum"
]
topicList
=
topicFromStb
ifcheckdata
=
0
keyList
=
'group.id:cgrp1,\
enable.auto.commit:false,\
auto.commit.interval.ms:6000,\
auto.offset.reset:earliest'
sql
=
"insert into consumeinfo values "
sql
+=
"(now, %d, '%s', '%s', %d, %d)"
%
(
consumerId
,
topicList
,
keyList
,
expectmsgcnt1
,
ifcheckdata
)
tdSql
.
query
(
sql
)
tdLog
.
info
(
"check stb if there are data"
)
while
1
:
tdSql
.
query
(
"select count(*) from %s"
%
parameterDict
[
"stbName"
])
#tdLog.info("row: %d, %l64d, %l64d"%(tdSql.getData(0, 1),tdSql.getData(0, 2),tdSql.getData(0, 3))
countOfStb
=
tdSql
.
getData
(
0
,
0
)
if
countOfStb
!=
0
:
tdLog
.
info
(
"count from stb: %d"
%
countOfStb
)
break
else
:
time
.
sleep
(
1
)
tdLog
.
info
(
"start consume processor"
)
pollDelay
=
5
showMsg
=
1
showRow
=
1
shellCmd
=
'nohup '
+
buildPath
+
'/build/bin/tmq_sim -c '
+
cfgPath
shellCmd
+=
" -y %d -d %s -g %d -r %d -w %s "
%
(
pollDelay
,
parameterDict
[
"dbName"
],
showMsg
,
showRow
,
cdbName
)
shellCmd
+=
"> /dev/null 2>&1 &"
tdLog
.
info
(
shellCmd
)
os
.
system
(
shellCmd
)
# create new child table and insert data
newCtbName
=
'newctb'
rowsOfNewCtb
=
1000
tdSql
.
query
(
"create table %s.%s using %s.%s tags(9999)"
%
(
parameterDict
[
"dbName"
],
newCtbName
,
parameterDict
[
"dbName"
],
parameterDict
[
"stbName"
]))
startTs
=
parameterDict
[
"startTs"
]
for
j
in
range
(
rowsOfNewCtb
):
sql
=
"insert into %s.%s values (%d, %d, 'tmqrow_%d') "
%
(
parameterDict
[
"dbName"
],
newCtbName
,
startTs
+
j
,
j
,
j
)
tdSql
.
execute
(
sql
)
tdLog
.
debug
(
"insert data into new child table ............ [OK]"
)
# wait for data ready
prepareEnvThread
.
join
()
tdLog
.
info
(
"insert process end, and start to check consume result"
)
while
1
:
tdSql
.
query
(
"select * from consumeresult"
)
#tdLog.info("row: %d, %l64d, %l64d"%(tdSql.getData(0, 1),tdSql.getData(0, 2),tdSql.getData(0, 3))
if
tdSql
.
getRows
()
==
1
:
break
else
:
time
.
sleep
(
5
)
expectmsgcnt
+=
rowsOfNewCtb
expectrowcnt
=
parameterDict
[
"rowsPerTbl"
]
*
parameterDict
[
"ctbNum"
]
+
rowsOfNewCtb
tdSql
.
checkData
(
0
,
1
,
consumerId
)
tdSql
.
checkData
(
0
,
2
,
expectmsgcnt
)
tdSql
.
checkData
(
0
,
3
,
expectrowcnt
)
# ==============================================================================
tdLog
.
printNoPrefix
(
"======== test scenario 3: "
)
#os.system('pkill tmq_sim')
def
stop
(
self
):
tdSql
.
close
()
tdLog
.
success
(
f
"
{
__file__
}
successfully executed"
)
tdCases
.
addLinux
(
__file__
,
TDTestCase
())
tdCases
.
addWindows
(
__file__
,
TDTestCase
())
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录