90 lines
3.1 KiB
Python
90 lines
3.1 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""模板化命名（templating）单元测试：Kafka / TDengine / PostgreSQL / MinIO。"""
|
|
import os
|
|
import sys
|
|
import unittest
|
|
|
|
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
|
import _bootstrap # noqa: F401
|
|
|
|
from data_bus.templating import TemplateNaming, sanitize, sanitize_sql
|
|
|
|
|
|
class SanitizeTest(unittest.TestCase):
|
|
def test_lower_and_replace_unsafe(self):
|
|
# 小写 + 空格/中文等非安全字符替换为 `_`，随后 strip 首尾 `._`
|
|
self.assertEqual(sanitize("Ti-Cl4 模板"), "ti-cl4")
|
|
self.assertEqual(sanitize("A.B-c_d"), "a.b-c_d")
|
|
|
|
def test_empty_fallback(self):
|
|
self.assertEqual(sanitize(""), "tpl")
|
|
self.assertEqual(sanitize("..."), "tpl")
|
|
|
|
def test_sql_extra_hyphen(self):
|
|
# SQL 标识符额外把 `-` 换成 `_`，避免不加引号时报错
|
|
self.assertEqual(sanitize_sql("ti-cl4"), "ti_cl4")
|
|
self.assertEqual(sanitize_sql("A.B-c"), "a.b_c")
|
|
|
|
|
|
class KafkaNamingTest(unittest.TestCase):
|
|
def setUp(self):
|
|
self.naming = TemplateNaming(template="ti-cl4", topic_prefix="ti-cl4", num_partitions=12)
|
|
|
|
def test_topic_format(self):
|
|
self.assertEqual(self.naming.topic("CLF-01"), "ti-cl4.clf-01.points")
|
|
|
|
def test_partition_deterministic_and_in_range(self):
|
|
p1 = self.naming.partition("CLF-01")
|
|
p2 = self.naming.partition("CLF-01")
|
|
self.assertEqual(p1, p2)
|
|
self.assertGreaterEqual(p1, 0)
|
|
self.assertLess(p1, 12)
|
|
|
|
def test_partitions_mapping(self):
|
|
mapping = self.naming.partitions(["CLF-01", "CLF-02"])
|
|
self.assertEqual(set(mapping), {"CLF-01", "CLF-02"})
|
|
self.assertEqual(mapping["CLF-01"], self.naming.partition("CLF-01"))
|
|
|
|
|
|
class TdengineNamingTest(unittest.TestCase):
|
|
def setUp(self):
|
|
self.naming = TemplateNaming(template="ti-cl4")
|
|
|
|
def test_stable_and_subtable(self):
|
|
self.assertEqual(self.naming.stable(), "ti_cl4_points")
|
|
# SQL 标识符保留 `.`（点位 ID 常用 `设备.测点` 形态）
|
|
self.assertEqual(self.naming.subtable("CLF-01.TEMP"), "ti_cl4_pt_clf_01.temp")
|
|
|
|
|
|
class PostgresNamingTest(unittest.TestCase):
|
|
def setUp(self):
|
|
self.naming = TemplateNaming(template="ti-cl4")
|
|
|
|
def test_schema_and_table(self):
|
|
self.assertEqual(self.naming.pg_schema(), "tpl_ti_cl4")
|
|
self.assertEqual(self.naming.pg_table("models"), "tpl_ti_cl4.models")
|
|
|
|
|
|
class MinioNamingTest(unittest.TestCase):
|
|
def setUp(self):
|
|
self.naming = TemplateNaming(template="ti-cl4")
|
|
|
|
def test_bucket(self):
|
|
self.assertEqual(self.naming.bucket(), "ti-cl4-artifacts")
|
|
|
|
def test_snapshot_key(self):
|
|
self.assertEqual(
|
|
self.naming.snapshot_key("quality-forecast", "2026-08-04", 3),
|
|
"features/quality_forecast/2026-08-04/000003.jsonl",
|
|
)
|
|
|
|
def test_model_artifact_key(self):
|
|
self.assertEqual(
|
|
self.naming.model_artifact_key("quality-forecast", "v1.2"),
|
|
"models/quality_forecast/v1.2/model.bin",
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|