xxq250
/
homework-jianmu
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

Test: Use parameters instead of numbers. for test_column_tag_boundary.py

This commit is contained in:
haoranchen 2024-09-03 09:20:37 +08:00 committed by GitHub
parent 5f477e0717
commit b8cca4b32c
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
1 changed files with 17 additions and 15 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

32
tests/army/insert/test_column_tag_boundary.py
View File

@ -22,6 +22,9 @@ class TDTestCase(TBase):
self.child_table_num = 1 self.child_table_num = 1
self.insert_round_num = 300 self.insert_round_num = 300
self.row_num_per_round = 15 self.row_num_per_round = 15
self.row_num_per_round_varbia_json = 8
self.rows_all1 = self.insert_round_num * self.row_num_per_round
self.rows_all2 = self.insert_round_num * self.row_num_per_round_varbia_json
self.start_ts = 1704082431000 self.start_ts = 1704082431000
def prepare_data(self): def prepare_data(self):
@ -47,8 +50,8 @@ class TDTestCase(TBase):
# check the data # check the data
for i in range(self.child_table_num): for i in range(self.child_table_num):
tdSql.query(f"select * from ct_binary{i+1};") tdSql.query(f"select * from ct_binary{i+1};")
tdSql.checkRows(4500) tdSql.checkRows(self.rows_all1)
row_num = random.randint(0, 3999) row_num = random.randint(0, self.rows_all1-1)
tdSql.checkData(row_num, 1, 'a' * self.max_column_length) tdSql.checkData(row_num, 1, 'a' * self.max_column_length)
tdSql.query(f"show tags from ct_binary{i+1};") tdSql.query(f"show tags from ct_binary{i+1};")
tdSql.checkData(0, 5, tag) tdSql.checkData(0, 5, tag)
@ -71,8 +74,8 @@ class TDTestCase(TBase):
# check the data # check the data
for i in range(self.child_table_num): for i in range(self.child_table_num):
tdSql.query(f"select * from ct_varchar{i+1};") tdSql.query(f"select * from ct_varchar{i+1};")
tdSql.checkRows(4500) tdSql.checkRows(self.rows_all1)
row_num = random.randint(0, 3999) row_num = random.randint(0, self.rows_all1-1)
tdSql.checkData(row_num, 1, 'b' * self.max_column_length) tdSql.checkData(row_num, 1, 'b' * self.max_column_length)
tdSql.query(f"show tags from ct_varchar{i+1};") tdSql.query(f"show tags from ct_varchar{i+1};")
tdSql.checkData(0, 5, tag) tdSql.checkData(0, 5, tag)
@ -106,8 +109,8 @@ class TDTestCase(TBase):
# check the data # check the data
for i in range(self.child_table_num): for i in range(self.child_table_num):
tdSql.query(f"select * from ct_nchar{i+1};") tdSql.query(f"select * from ct_nchar{i+1};")
tdSql.checkRows(4500) tdSql.checkRows(self.rows_all1)
row_num = random.randint(0, 3999) row_num = random.randint(0, self.rows_all1-1)
tdSql.checkData(row_num, 1, column) tdSql.checkData(row_num, 1, column)
tdSql.query(f"show tags from ct_nchar{i+1};") tdSql.query(f"show tags from ct_nchar{i+1};")
tdSql.checkData(0, 5, tag) tdSql.checkData(0, 5, tag)
@ -124,22 +127,21 @@ class TDTestCase(TBase):
# insert data # insert data
for j in range(self.insert_round_num): for j in range(self.insert_round_num):
sql = "insert into ct_varbinary%s values" % (i+1) sql = "insert into ct_varbinary%s values" % (i+1)
for k in range(row_num_per_round): for k in range(self.row_num_per_round_varbia_json):
sql += "(%s, '%s')," % (str(self.start_ts + (j * self.insert_round_num + k * self.row_num_per_round + 1)), '\\x' + column) sql += "(%s, '%s')," % (str(self.start_ts + (j * self.insert_round_num + k * self.row_num_per_round + 1)), '\\x' + column)
tdSql.execute(sql) tdSql.execute(sql)
tdLog.info(f"Insert {row_num_per_round} rows data into ct_varbinary{i+1} {j+1} times successfully") tdLog.info(f"Insert {self.row_num_per_round_varbia_json} rows data into ct_varbinary{i+1} {j+1} times successfully")
tdSql.execute("flush database db;") tdSql.execute("flush database db;")
# check the data # check the data
for i in range(self.child_table_num): for i in range(self.child_table_num):
tdSql.query(f"select * from ct_varbinary{i+1};") tdSql.query(f"select * from ct_varbinary{i+1};")
tdSql.checkRows(2400) tdSql.checkRows(self.rows_all2)
row_num = random.randint(0, 2399) row_num = random.randint(0, self.rows_all2-1)
tdSql.checkData(row_num, 1, bytes.fromhex(column)) tdSql.checkData(row_num, 1, bytes.fromhex(column))
tdSql.query(f"show tags from ct_varbinary{i+1};") tdSql.query(f"show tags from ct_varbinary{i+1};")
tdSql.checkData(0, 5, '\\x' + tag.upper()) tdSql.checkData(0, 5, '\\x' + tag.upper())
def test_json_tag_boundary(self): def test_json_tag_boundary(self):
row_num_per_round = 8
max_json_tag_length = 4095 max_json_tag_length = 4095
max_json_tag_key_length = 256 max_json_tag_key_length = 256
# create tables # create tables
@ -153,16 +155,16 @@ class TDTestCase(TBase):
# insert data # insert data
for j in range(self.insert_round_num): for j in range(self.insert_round_num):
sql = "insert into ct_json_tag%s values" % (i+1) sql = "insert into ct_json_tag%s values" % (i+1)
for k in range(row_num_per_round): for k in range(self.row_num_per_round_varbia_json):
sql += "(%s, '%s')," % (str(self.start_ts + (j * self.insert_round_num + k * self.row_num_per_round + 1)), '\\x' + column) sql += "(%s, '%s')," % (str(self.start_ts + (j * self.insert_round_num + k * self.row_num_per_round + 1)), '\\x' + column)
tdSql.execute(sql) tdSql.execute(sql)
tdLog.info(f"Insert {row_num_per_round} rows data into ct_json_tag{i+1} {j+1} times successfully") tdLog.info(f"Insert {self.row_num_per_round_varbia_json} rows data into ct_json_tag{i+1} {j+1} times successfully")
tdSql.execute("flush database db;") tdSql.execute("flush database db;")
# check the data # check the data
for i in range(self.child_table_num): for i in range(self.child_table_num):
tdSql.query(f"select * from ct_json_tag{i+1} where t1->'{tag_key}' = '{tag_value}';") tdSql.query(f"select * from ct_json_tag{i+1} where t1->'{tag_key}' = '{tag_value}';")
tdSql.checkRows(2400) tdSql.checkRows(self.rows_all2)
row_num = random.randint(0, 2399) row_num = random.randint(0, self.rows_all2-1)
tdSql.checkData(row_num, 1, bytes.fromhex(column)) tdSql.checkData(row_num, 1, bytes.fromhex(column))
def run(self): def run(self):