bot_dev1
33e64fe18b
新增 tests/e2e/ 端到端联调测试套件，覆盖四个 iAOP-Core 内核模块的全链路协作： 数据流（PRD 5.1→5.2）： - edge-gateway 只读采集（模拟驱动）→ spool 断点续传 → data-bus 批量写入 - 验证不丢不重（幂等去重）、样本字段完整、健康度满足 SLA 问答流（PRD 5.4）： - rag-kb 模板化知识库检索（命中片段+来源）→ llm-gateway 混合网关 - 验证敏感度路由、DLP 拦截、幻觉溯源校验、审计可追溯 跨链路：采集→落库→知识沉淀→安全问答业务闭环 共 14 个用例，全部基于可注入接口运行，零外部依赖（CI 可直接执行）。 运行：python -m unittest discover -s tests/e2e -v
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()
|