# -*- coding: utf-8 -*- """端到端联调用例（Issue #86）—— 数据流链路。 验证 PRD 5.1（边缘采集网关）→ PRD 5.2（数据总线 + 时序库）的端到端协作： edge-gateway 只读采集（模拟驱动）→ spool 缓存 → BatchWriter 批量写入 （MemorySink 幂等去重，「不丢不重」）。 不依赖 Kafka / TDengine / 真实设备：全部走内存桩，可在 CI 直接执行。 """ from __future__ import annotations # 引导加载四个内核模块（必须在测试导入前执行） import tests.e2e._bootstrap # noqa: F401 import os import tempfile import unittest from collector.engine import CollectorEngine from collector.metrics import HealthMetrics from collector.spool import SpoolStore from drivers import SimulatorDriver from point_dict.loader import Point, PointDict from data_bus import BatchWriter, MemorySink def _make_point_dict() -> PointDict: """模拟氯化车间（Template-Ti）点位集：2 台设备 × 3 测点。""" points = [ Point(device_id="CLF-01", point_id="CLF-01.TEMP", name="1#炉温", unit="℃", data_type="float", sample_rate=1000, quality_code=True, row_number=2), Point(device_id="CLF-01", point_id="CLF-01.PRES", name="1#炉压", unit="kPa", data_type="float", sample_rate=1000, quality_code=True, row_number=3), Point(device_id="CLF-01", point_id="CLF-01.FLOW", name="1#氯气流量", unit="m³/h", data_type="float", sample_rate=1000, quality_code=True, row_number=4), Point(device_id="CLF-02", point_id="CLF-02.TEMP", name="2#炉温", unit="℃", data_type="float", sample_rate=1000, quality_code=True, row_number=5), Point(device_id="CLF-02", point_id="CLF-02.PRES", name="2#炉压", unit="kPa", data_type="float", sample_rate=1000, quality_code=True, row_number=6), Point(device_id="CLF-02", point_id="CLF-02.FLOW", name="2#氯气流量", unit="m³/h", data_type="float", sample_rate=1000, quality_code=True, row_number=7), ] return PointDict(points) def _make_engine(tmpdir: str, point_dict: PointDict): """组装一个最小可运行的采集引擎（模拟驱动，spool 落临时目录）。""" spool = SpoolStore(spool_dir=tmpdir, cache_limit_bytes=16 * 1024 * 1024) metrics = HealthMetrics() engine = CollectorEngine( point_dict=point_dict, driver_slots=[("simulator", SimulatorDriver(), [])], spool=spool, metrics=metrics, interval_ms=1000, max_pending=100_000, ) return engine, spool, metrics class DataPipelineE2ETest(unittest.TestCase): """采集 → spool → 批量写入 全链路联调。""" def setUp(self) -> None: self._tmp = tempfile.mkdtemp(prefix="iaop_e2e_") self.point_dict = _make_point_dict() self.engine, self.spool, self.metrics = _make_engine( self._tmp, self.point_dict ) # -- 采集 → spool ------------------------------------------------------ def test_collect_once_all_points_read_into_spool(self) -> None: """一轮采集：6 点位全部读出并落入 spool（expected == got）。""" got = self.engine.collect_once() self.assertEqual(got, len(self.point_dict)) # spool 待上行记录数应等于本轮样本数 self.assertEqual(self.spool.total_pending(), got) # 健康度：一轮全成功，失败计 0 self.assertEqual(self.metrics.total_rounds, 1) self.assertEqual(self.metrics.failed_rounds, 0) def test_collect_multiple_rounds_accumulate(self) -> None: """多轮采集：spool 累计样本数 = 轮数 × 点位数（未上行前不丢）。""" rounds = 5 total = 0 for _ in range(rounds): total += self.engine.collect_once() self.assertEqual(total, rounds * len(self.point_dict)) self.assertEqual(self.spool.total_pending(), total) self.assertEqual(self.metrics.total_rounds, rounds) # -- spool → BatchWriter（幂等去重，不丢不重）------------------------- def test_spool_drain_into_batchwriter_no_loss_no_dup(self) -> None: """采集 → spool → 批量写入：样本不丢、重复不重。""" rounds = 3 for _ in range(rounds): self.engine.collect_once() sink = MemorySink() writer = BatchWriter(sink=sink, batch_size=len(self.point_dict), flush_interval=0.0) accepted = 0 # 从 spool 取出待上行样本（pending_records 读取，ack 确认删除） for sample in self.spool.pending_records(): if writer.push(sample): accepted += 1 self.spool.ack(sample) # 上行确认 writer.flush() expected = rounds * len(self.point_dict) self.assertEqual(accepted, expected) # 全部接受（无重复） self.assertEqual(len(sink.rows), expected) # 全部落库（不丢） self.assertEqual(sink.write_count, expected) # spool 已全部确认，待上行归零 self.assertEqual(self.spool.total_pending(), 0) def test_batchwriter_idempotent_on_replay(self) -> None: """断点续传重发同一批样本：落库不重复（幂等去重）。""" sink = MemorySink() writer = BatchWriter(sink=sink, batch_size=10, flush_interval=0.0) self.engine.collect_once() samples = self.spool.pending_records() # 第一次写入 for s in samples: writer.push(s) writer.flush() first_count = len(sink.rows) # 模拟断点续传：重发同一批样本 for s in samples: writer.push(s) writer.flush() self.assertEqual(len(sink.rows), first_count) # 重发不产生重复 # BatchWriter 统计：duplicates 应等于重发条数 stats = writer.stats() self.assertEqual(stats["duplicates"], len(samples)) # -- 全链路数据完整性 -------------------------------------------------- def test_sample_fields_preserved_end_to_end(self) -> None: """端到端：样本字段（device/point/value/unit/ts）完整落库。""" self.engine.collect_once() sink = MemorySink() writer = BatchWriter(sink=sink, batch_size=10, flush_interval=0.0) for s in self.spool.pending_records(): writer.push(s) writer.flush() point_ids = {p.point_id for p in self.point_dict.points} written_ids = {r["point_id"] for r in sink.rows} self.assertEqual(written_ids, point_ids) # 所有点位都落库 for row in sink.rows: # 字段完整性 for key in ("device_id", "point_id", "value", "ts"): self.assertIn(key, row) # 设备与点位字典一致 src = next(p for p in self.point_dict.points if p.point_id == row["point_id"]) self.assertEqual(row["device_id"], src.device_id) # -- 健康度上报贯穿链路 ------------------------------------------------ def test_health_metrics_reflect_collection(self) -> None: """采集健康度（成功率/丢失率）随采集轮次正确累计。""" for _ in range(10): self.engine.collect_once() self.assertEqual(self.metrics.total_rounds, 10) self.assertEqual(self.metrics.failed_rounds, 0) # 可用性 = 1 - 失败轮次/总轮次，目标 ≥ 99.8%（此处 100%） availability = self.metrics.availability self.assertGreaterEqual(availability, 0.998) self.assertTrue(self.metrics.meets_sla()) if __name__ == "__main__": unittest.main()