184 lines
7.7 KiB
Python
184 lines
7.7 KiB
Python
|
# -*- 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()
|