feat: 完成 issue #28 ② Kafka topic 命名/分区模板化（按模板+点位维度）

2026-08-05 00:35:19 +08:00
parent afee113901
commit 793dd0a3b8
5 changed files with 307 additions and 0 deletions
@@ -0,0 +1,137 @@
+# -*- coding: utf-8 -*-
+"""Kafka topic 命名/分区模板化测试（issue #28）。
+
+覆盖：
+1. 模板配置资产加载（config/kafka.template.yaml，含点位维度覆盖规则）；
+2. topic 推导：`{topic_prefix}.{device_id}.points`（与 edge-gateway 上行一致）；
+3. 分区：device_id 一致性哈希（确定性、单设备分区内有序）；
+4. 点位维度覆盖：LAB-* 前缀 → 质检独立 topic，其余走设备 topic；
+5. 生产路由：按 (topic, partition) 批量分组。
+"""
+import os
+import sys
+import unittest
+
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+import _bootstrap  # noqa: F401
+
+from data_bus.kafka_naming import (  # noqa: E402
+    DEFAULT_CONFIG_PATH,
+    KafkaTopicNaming,
+)
+
+CONFIG = os.path.join(
+    os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
+    "config", "kafka.template.yaml",
+)
+
+
+def sample(device_id="CLF-01", point_id="CLF-01.TEMP", value=850.5, ts=1.0):
+    return {"device_id": device_id, "point_id": point_id,
+            "value": value, "ts": ts}
+
+
+class TestConfigLoad(unittest.TestCase):
+    """模板配置资产加载与摘要。"""
+
+    def test_from_template_config(self):
+        naming = KafkaTopicNaming.from_template_config(CONFIG)
+        self.assertEqual(naming._tpl.template, "ti-cl4")
+        self.assertEqual(naming.num_partitions, 12)
+        self.assertEqual(naming.acks, "all")
+        self.assertEqual(naming.retention_hours, 168)
+        brief = naming.brief()
+        self.assertIn("LAB-", brief["per_point_rules"])
+        self.assertIn("ALM-", brief["per_point_rules"])
+
+    def test_default_config_path_exists(self):
+        self.assertTrue(os.path.isfile(DEFAULT_CONFIG_PATH))
+
+
+class TestTopicNaming(unittest.TestCase):
+    """topic 推导（与 edge-gateway 上行格式一致）。"""
+
+    def setUp(self):
+        self.naming = KafkaTopicNaming.from_template_config(CONFIG)
+
+    def test_device_topic_format(self):
+        # 与 upstream/kafka_sink.py 的 {prefix}.{device}.points 一致
+        self.assertEqual(self.naming.topic("CLF-01"),
+                         "iaop.ti-cl4.CLF-01.points")
+        self.assertEqual(self.naming.topic("S7-01"),
+                         "iaop.ti-cl4.S7-01.points")
+
+    def test_point_level_topic_override(self):
+        # LAB-* 前缀命中质量标签独立通道；普通点走设备 topic
+        self.assertEqual(
+            self.naming.topic("CLF-01", point_id="LAB-01.TI_PURITY"),
+            "iaop.ti-cl4.quality.points")
+        self.assertEqual(
+            self.naming.topic("CLF-01", point_id="ALM-01.TEMP_HI"),
+            "iaop.ti-cl4.alarm.points")
+        self.assertEqual(
+            self.naming.topic("CLF-01", point_id="CLF-01.TEMP"),
+            "iaop.ti-cl4.CLF-01.points")
+
+    def test_sanitize_device_id(self):
+        # device_id 原样保留（与 edge-gateway 上行一致，点位字典保证合法）
+        self.assertEqual(self.naming.topic("CLF-01"),
+                         "iaop.ti-cl4.CLF-01.points")
+
+
+class TestPartition(unittest.TestCase):
+    """分区一致性哈希：确定性 + 单设备有序。"""
+
+    def setUp(self):
+        self.naming = KafkaTopicNaming.from_template_config(CONFIG)
+
+    def test_deterministic(self):
+        self.assertEqual(self.naming.partition("CLF-01"),
+                         self.naming.partition("CLF-01"))
+        self.assertIn(self.naming.partition("CLF-01"), range(12))
+
+    def test_same_device_same_partition(self):
+        # 单设备所有点位落到同一分区 → 分区内有序
+        p1 = self.naming.partition("CLF-01")
+        for pid in ("CLF-01.TEMP", "CLF-01.PRES", "CLF-01.FEED"):
+            self.assertEqual(self.naming.partition("CLF-01"), p1)
+
+    def test_partitions_preview(self):
+        mapping = self.naming.partitions(["CLF-01", "S7-01", "E-01"])
+        self.assertEqual(set(mapping), {"CLF-01", "S7-01", "E-01"})
+        self.assertTrue(all(0 <= v < 12 for v in mapping.values()))
+
+
+class TestRouting(unittest.TestCase):
+    """按 (topic, partition) 批量分组（生产路由）。"""
+
+    def setUp(self):
+        self.naming = KafkaTopicNaming.from_template_config(CONFIG)
+
+    def test_group_by_topic_and_partition(self):
+        rows = [
+            sample("CLF-01", "CLF-01.TEMP"),
+            sample("CLF-01", "CLF-01.PRES"),
+            sample("CLF-01", "LAB-01.TI_PURITY"),   # 质量标签 → 独立 topic
+            sample("S7-01", "S7-01.PUMP_A"),
+        ]
+        groups = self.naming.routing(rows)
+        # 3 个不同 (topic, partition) 组：CLF 默认 / LAB 质量 / S7
+        self.assertEqual(len(groups), 3)
+        topics = {t for t, _ in groups}
+        self.assertIn("iaop.ti-cl4.quality.points", topics)
+        # CLF 默认 topic 内 2 条同一分区
+        clf_key = ("iaop.ti-cl4.CLF-01.points",
+                   self.naming.partition("CLF-01"))
+        self.assertEqual(len(groups[clf_key]), 2)
+
+    def test_rows_without_point_id(self):
+        rows = [{"device_id": "CLF-01", "value": 1.0, "ts": 2.0}]
+        groups = self.naming.routing(rows)
+        self.assertEqual(len(groups), 1)
+        key = next(iter(groups))
+        self.assertEqual(key[0], "iaop.ti-cl4.CLF-01.points")
+
+
+if __name__ == "__main__":
+    unittest.main()