mirror of
https://github.com/taosdata/TDengine
synced 2026-05-24 10:09:01 +00:00
118 lines
No EOL
3.9 KiB
Python
118 lines
No EOL
3.9 KiB
Python
import time
|
|
import math
|
|
from new_test_framework.utils import tdLog, tdSql, tdStream, StreamTableType, StreamTable, StreamItem
|
|
from datetime import datetime
|
|
|
|
class TestStreamDevBasic:
|
|
|
|
def setup_class(cls):
|
|
tdLog.debug(f"start to execute {__file__}")
|
|
|
|
def test_stream_dev_basic(self):
|
|
"""Stream basic development testing 11
|
|
|
|
Verification testing during the development process.
|
|
|
|
Catalog:
|
|
- Streams:Others
|
|
|
|
Since: v3.3.3.7
|
|
|
|
Labels: common,ci
|
|
|
|
Jira: None
|
|
|
|
History:
|
|
- 2025-6-16 xsRen Created
|
|
|
|
"""
|
|
tdStream.createSnode()
|
|
self.basic1()
|
|
self.basic2()
|
|
|
|
@staticmethod
|
|
def custom_cint_generator(row):
|
|
return str(row * 10) # 每行的 cint 为 row * 10
|
|
|
|
def basic1(self):
|
|
tdLog.info(f"basic test 1")
|
|
|
|
tdStream.dropAllStreamsAndDbs()
|
|
|
|
tdStream.init_database("test")
|
|
tdStream.init_database("test1")
|
|
|
|
st1 = StreamTable("test", "st", StreamTableType.TYPE_SUP_TABLE)
|
|
st1.createTable()
|
|
st1.appendSubTables(200, 240)
|
|
st1.append_data(0, 40)
|
|
st1.append_subtable_data("st_1", 40, 50)
|
|
st1.update_data(3, 4)
|
|
st1.update_subtable_data("st_1", 6, 7)
|
|
st1.delete_data(7, 9)
|
|
st1.delete_subtable_data("st_2", 8, 10)
|
|
|
|
|
|
ntb = StreamTable("test", "ntb", StreamTableType.TYPE_NORMAL_TABLE)
|
|
ntb.createTable()
|
|
ntb.append_data(0, 40)
|
|
ntb.update_data(3, 4)
|
|
ntb.delete_data(7, 9)
|
|
|
|
ntb2 = StreamTable("test1", "ntb2", StreamTableType.TYPE_NORMAL_TABLE)
|
|
ntb2.set_columns("ts timestamp, c0 int, c1 int, c2 int")
|
|
ntb2.register_column_generator("c0", self.custom_cint_generator)
|
|
ntb2.createTable()
|
|
ntb2.append_data(0, 10)
|
|
ntb2.append_data(20, 30)
|
|
ntb2.update_data(3, 6)
|
|
|
|
def checkBaic2Results(self):
|
|
tdSql.query("select * from test.st7;", queryTimes=1)
|
|
|
|
ts = tdSql.getColData(0)
|
|
avg_cint = tdSql.getColData(1)
|
|
count_cint = tdSql.getColData(2)
|
|
|
|
for i in range(0, 39):
|
|
sql = f"select '{ts[i]}', avg(cint), count(cint) from test.st where cts <= '{ts[i]}'"
|
|
tdSql.query(sql, queryTimes=1)
|
|
|
|
expected_ts = datetime.strptime(tdSql.getData(0, 0), "%Y-%m-%d %H:%M:%S")
|
|
expected_avg_cint = tdSql.getData(0, 1)
|
|
expected_count_cint = tdSql.getData(0, 2)
|
|
|
|
assert ts[i] == expected_ts, f"Row {i} ts mismatch: expected {expected_ts}, got {ts[i]}"
|
|
assert math.isclose(avg_cint[i], expected_avg_cint, rel_tol=1e-9), f"Row {i} avg_cint mismatch: expected {expected_avg_cint}, got {avg_cint[i]}"
|
|
assert count_cint[i] == expected_count_cint, f"Row {i} count_cint mismatch: expected {expected_count_cint}, got {count_cint[i]}"
|
|
|
|
def basic2(self):
|
|
tdLog.info(f"basic test 1")
|
|
|
|
tdStream.dropAllStreamsAndDbs()
|
|
|
|
tdStream.init_database("test")
|
|
|
|
trigger = StreamTable("test", "trigger", StreamTableType.TYPE_SUP_TABLE)
|
|
trigger.createTable(10)
|
|
trigger.append_data(0, 40)
|
|
trigger.update_data(3, 4)
|
|
|
|
st1 = StreamTable("test", "st", StreamTableType.TYPE_SUP_TABLE)
|
|
st1.createTable()
|
|
st1.appendSubTables(200, 240)
|
|
st1.append_data(0, 40)
|
|
|
|
sql = f"create stream s7 state_window (cint) from test.trigger options(fill_history_first(1)) into st7 as select _twstart, avg(cint), count(cint) from test.st where cts <= _twstart;"
|
|
|
|
stream1 = StreamItem(
|
|
id=0,
|
|
stream=sql,
|
|
res_query="select * from test.st7;",
|
|
check_func=self.checkBaic2Results,
|
|
)
|
|
stream1.createStream()
|
|
stream1.awaitRowStability(39)
|
|
stream1.checkResults()
|
|
|
|
tdLog.info("======over") |