39 lines
1.7 KiB
Python
39 lines
1.7 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""Sanity check: data_bus package imports and postgres DDL escaping fix."""
|
|
import importlib.machinery
|
|
import importlib.util
|
|
import os
|
|
import sys
|
|
|
|
d = os.getcwd()
|
|
|
|
# 以包身份加载 data-bus（执行 __init__.py，相对导入可用）
|
|
spec = importlib.util.spec_from_file_location(
|
|
"data_bus", os.path.join(d, "__init__.py"), submodule_search_locations=[d]
|
|
)
|
|
pkg = importlib.util.module_from_spec(spec)
|
|
sys.modules["data_bus"] = pkg
|
|
spec.loader.exec_module(pkg)
|
|
|
|
from data_bus import BatchWriter, MemorySink, TdengineSink, TemplateNaming, __version__ # noqa: E402
|
|
from data_bus.postgres_schema import generate_schema_ddl # noqa: E402
|
|
from data_bus.tdengine_schema import generate_schema, specs_from_point_dict_rows # noqa: E402
|
|
|
|
ddl = generate_schema_ddl(TemplateNaming(template="ti-cl4"))
|
|
assert "DEFAULT '{}'::jsonb" in ddl, "jsonb default escaped wrongly"
|
|
assert "REFERENCES tpl_ti_cl4.templates(template)" in ddl
|
|
print("version:", __version__)
|
|
print("ddl ok; lines:", len(ddl.splitlines()))
|
|
print("exports ok:", [c.__name__ for c in (BatchWriter, MemorySink, TdengineSink)])
|
|
|
|
# 子任务 #29：依据点位字典自动生成 TDengine 完整 schema（超级表 + 子表）
|
|
_td_rows = [
|
|
{"device_id": "CLF-01", "point_id": "CLF-01.TEMP", "unit": "℃", "dataType": "float"},
|
|
{"device_id": "CLF-02", "point_id": "CLF-02.RUN", "unit": "%", "dataType": "bool"},
|
|
]
|
|
_td_ddls = generate_schema(TemplateNaming(template="ti-cl4"), specs_from_point_dict_rows(_td_rows), retention_days=90)
|
|
assert _td_ddls[0].startswith("CREATE STABLE IF NOT EXISTS ti_cl4_points")
|
|
assert len(_td_ddls) == 3 # 1 超级表 + 2 子表
|
|
assert all("USING ti_cl4_points" in d for d in _td_ddls[1:])
|
|
print("tdengine schema ok; statements:", len(_td_ddls))
|