You need to sign in or sign up before continuing.
提交 8a9a2759 编写于 作者: haoranc's avatar haoranc

test:modify testcase of stmt

上级 d6e1d1bc
...@@ -82,7 +82,7 @@ class TDTestCase: ...@@ -82,7 +82,7 @@ class TDTestCase:
return con return con
def test_stmt_set_tbname_tag(self,conn): def test_stmt_set_tbname_tag(self,conn):
dbname = "stmt_set_tbname_tag" dbname = "stmt_tag"
try: try:
conn.execute("drop database if exists %s" % dbname) conn.execute("drop database if exists %s" % dbname)
...@@ -196,31 +196,31 @@ class TDTestCase: ...@@ -196,31 +196,31 @@ class TDTestCase:
assert rows3[0][0] == 12, 'fourth case is failed' assert rows3[0][0] == 12, 'fourth case is failed'
assert rows3[1][0] == 12, 'fourth case is failed' assert rows3[1][0] == 12, 'fourth case is failed'
# #query: conversion Functions #query: conversion Functions
# querystmt4=conn.statement("select cast( ? as bigint) from log ") querystmt4=conn.statement("select cast( ? as bigint) from log ")
# queryparam4=new_bind_params(1) queryparam4=new_bind_params(1)
# print(type(queryparam4)) print(type(queryparam4))
# queryparam4[0].binary('1232a') queryparam4[0].binary('1232a')
# querystmt4.bind_param(queryparam4) querystmt4.bind_param(queryparam4)
# querystmt4.execute() querystmt4.execute()
# result4=querystmt4.use_result() result4=querystmt4.use_result()
# rows4=result4.fetch_all() rows4=result4.fetch_all()
# print("5",rows4) print("5",rows4)
# assert rows4[0][0] == 1232 assert rows4[0][0] == 1232
# assert rows4[1][0] == 1232 assert rows4[1][0] == 1232
# querystmt4=conn.statement("select cast( ? as binary(10)) from log ") querystmt4=conn.statement("select cast( ? as binary(10)) from log ")
# queryparam4=new_bind_params(1) queryparam4=new_bind_params(1)
# print(type(queryparam4)) print(type(queryparam4))
# queryparam4[0].int(123) queryparam4[0].int(123)
# querystmt4.bind_param(queryparam4) querystmt4.bind_param(queryparam4)
# querystmt4.execute() querystmt4.execute()
# result4=querystmt4.use_result() result4=querystmt4.use_result()
# rows4=result4.fetch_all() rows4=result4.fetch_all()
# print("6",rows4) print("6",rows4)
# assert rows4[0][0] == '123' assert rows4[0][0] == '123'
# assert rows4[1][0] == '123' assert rows4[1][0] == '123'
# #query: datatime Functions # #query: datatime Functions
......
...@@ -84,21 +84,21 @@ class TDTestCase: ...@@ -84,21 +84,21 @@ class TDTestCase:
def test_stmt_insert_multi(self,conn): def test_stmt_insert_multi(self,conn):
# type: (TaosConnection) -> None # type: (TaosConnection) -> None
dbname = "pytest_taos_stmt_multi" dbname = "db_stmt"
try: try:
conn.execute("drop database if exists %s" % dbname) conn.execute("drop database if exists %s" % dbname)
conn.execute("create database if not exists %s" % dbname) conn.execute("create database if not exists %s" % dbname)
conn.select_db(dbname) conn.select_db(dbname)
conn.execute( conn.execute(
"create table if not exists log(ts timestamp, bo bool, nil tinyint, ti tinyint, si smallint, ii int,\ "create table if not exists stb1(ts timestamp, bo bool, nil tinyint, ti tinyint, si smallint, ii int,\
bi bigint, tu tinyint unsigned, su smallint unsigned, iu int unsigned, bu bigint unsigned, \ bi bigint, tu tinyint unsigned, su smallint unsigned, iu int unsigned, bu bigint unsigned, \
ff float, dd double, bb binary(100), nn nchar(100), tt timestamp)", ff float, dd double, bb binary(100), nn nchar(100), tt timestamp)",
) )
# conn.load_table_info("log") # conn.load_table_info("log")
start = datetime.now() start = datetime.now()
stmt = conn.statement("insert into log values(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)") stmt = conn.statement("insert into stb1 values(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)")
params = new_multi_binds(16) params = new_multi_binds(16)
params[0].timestamp((1626861392589, 1626861392590, 1626861392591)) params[0].timestamp((1626861392589, 1626861392590, 1626861392591))
...@@ -125,7 +125,7 @@ class TDTestCase: ...@@ -125,7 +125,7 @@ class TDTestCase:
assert stmt.affected_rows == 3 assert stmt.affected_rows == 3
#query 1 #query 1
querystmt=conn.statement("select ?,bu from log") querystmt=conn.statement("select ?,bu from stb1")
queryparam=new_bind_params(1) queryparam=new_bind_params(1)
print(type(queryparam)) print(type(queryparam))
queryparam[0].binary("ts") queryparam[0].binary("ts")
...@@ -135,7 +135,7 @@ class TDTestCase: ...@@ -135,7 +135,7 @@ class TDTestCase:
# rows=result.fetch_all() # rows=result.fetch_all()
# print( querystmt.use_result()) # print( querystmt.use_result())
# result = conn.query("select * from log") # result = conn.query("select * from stb1")
rows=result.fetch_all() rows=result.fetch_all()
# rows=result.fetch_all() # rows=result.fetch_all()
print(rows) print(rows)
...@@ -144,7 +144,7 @@ class TDTestCase: ...@@ -144,7 +144,7 @@ class TDTestCase:
assert rows[2][1] == None assert rows[2][1] == None
#query 2 #query 2
querystmt1=conn.statement("select * from log where bu < ?") querystmt1=conn.statement("select * from stb1 where bu < ?")
queryparam1=new_bind_params(1) queryparam1=new_bind_params(1)
print(type(queryparam1)) print(type(queryparam1))
queryparam1[0].int(4) queryparam1[0].int(4)
......
Markdown is supported
0% .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑!
想要评论请 注册