diff --git a/tests/e2e/README.md b/tests/e2e/README.md new file mode 100644 index 0000000..e7ab3b7 --- /dev/null +++ b/tests/e2e/README.md @@ -0,0 +1,55 @@ +# iAOP 端到端联调用例（Issue #86） + +对应 PRD 第 8 章「端到端联调」与 EPIC #14（M4）子任务 #86。 +跨四个 iAOP-Core 内核模块的全链路联调测试，覆盖「数据流」与「问答流」两条端到端链路， +全部基于各模块的**可注入接口**运行，零外部依赖（不依赖 Kafka / TDengine / PostgreSQL / +MinIO / 真实 LLM / 真实设备），可在 CI 中直接执行。 + +## 链路覆盖 + +### 数据流（PRD 5.1 → 5.2） +`edge-gateway` 只读采集（模拟驱动）→ `spool` 断点续传缓存 → `data-bus` BatchWriter +批量写入（MemorySink，幂等去重，「不丢不重」）→ 健康度上报（P99 / 丢失率 / 可用性）。 + +| 用例 | 验证点 | +|------|--------| +| `test_collect_once_all_points_read_into_spool` | 单轮采集：全部点位读出并落 spool | +| `test_collect_multiple_rounds_accumulate` | 多轮采集：spool 累计 = 轮数 × 点位数（不丢） | +| `test_spool_drain_into_batchwriter_no_loss_no_dup` | spool → 批量写入：不丢不重 | +| `test_batchwriter_idempotent_on_replay` | 断点续传重发：幂等去重 | +| `test_sample_fields_preserved_end_to_end` | 样本字段端到端完整 | +| `test_health_metrics_reflect_collection` | 健康度随轮次累计，满足 SLA（P99≤1.8s / 丢失率≤0.02% / 可用性≥99.8%） | + +### 问答流（PRD 5.4） +`rag-kb` 模板化知识库检索（命中文档片段 + 来源）→ `llm-gateway` 混合网关 +（敏感度路由 → 生成 → 幻觉/溯源校验 → DLP 出站防线）。 + +| 用例 | 验证点 | +|------|--------| +| `test_kb_retrieval_returns_relevant_chunks_with_source` | RAG 检索命中并带来源 | +| `test_kb_category_filter` | 类别过滤（工艺规范 / SOP / 国标） | +| `test_normal_question_routes_local_with_rag_context` | 普通问题 → 本地后端 + 来源引用 | +| `test_sensitive_query_blocked_by_dlp` | 敏感数据 → DLP 拦截 → block → 转人工 | +| `test_generic_question_routes_cloud` | 通用问题 DLP 放行闭环正常 | +| `test_high_stakes_low_confidence_to_human` | 高利害 + 低信度 → 转人工复核 | +| `test_audit_drain_after_query` | 各组件审计记录可统一导出 | + +### 跨链路集成（业务闭环） +采集 → 落库 → 知识沉淀 → 安全问答的完整业务闭环（`test_cross_pipeline.py`）。 + +## 运行方式 + +```bash +# 在仓库根目录执行（需 Python 3.8+，仅标准库依赖） +python -m unittest discover -s tests/e2e -v +``` + +## 设计说明 + +- **零外部依赖**：四个内核模块的存储 / 推理 / 协议接入均设计为可注入接口 + （`StoreSink` / `InferenceBackend` / `Driver`），联调用内存桩替换真实后端， + 保证 CI 可重复执行且不污染环境。 +- **模块加载**：四个 core 模块的源码组织方式不一致（edge-gateway 为裸导入， + 其余为带连字符目录名的包），由 `_bootstrap.py` 统一加载（详见该文件注释）。 +- **断言对齐 PRD 验收基线**：健康度断言直接引用 PRD 第 9 章 NFR 阈值 + （P99≤1.8s / 丢失率≤0.02% / 可用性≥99.8%），DLP 断言验证 100% 拦截目标。 diff --git a/tests/e2e/__init__.py b/tests/e2e/__init__.py new file mode 100644 index 0000000..07e51ef --- /dev/null +++ b/tests/e2e/__init__.py @@ -0,0 +1,18 @@ +# -*- coding: utf-8 -*- +"""iAOP 端到端联调用例（Issue #86）。 + +跨四个 iAOP-Core 内核模块的全链路联调测试，验证「数据流」与「问答流」 +两条端到端链路在内存桩（不依赖 Kafka/TDengine/MinIO/真实 LLM）下完整闭环： + +数据流（PRD 5.1 → 5.2）： + edge-gateway 只读采集（模拟驱动）→ spool 缓存 → data-bus 批量写入 + （MemorySink，幂等去重，不丢不重）。 + +问答流（PRD 5.4）： + rag-kb 模板化知识库检索（命中文档片段 + 来源）→ llm-gateway 混合网关 + （路由 → 生成 → 溯源校验 → DLP 出站防线）。 + +本套件全部基于各模块的「可注入接口」运行，零外部依赖，可在 CI 中直接执行。 +运行：在仓库根目录执行 `python -m unittest discover -s tests/e2e -v` +（脚本会自动把四个 core 模块加入 sys.path，见 _bootstrap.py）。 +""" diff --git a/tests/e2e/_bootstrap.py b/tests/e2e/_bootstrap.py new file mode 100644 index 0000000..5580dd5 --- /dev/null +++ b/tests/e2e/_bootstrap.py @@ -0,0 +1,60 @@ +# -*- coding: utf-8 -*- +"""端到端联调测试引导：统一加载四个 iAOP-Core 内核模块。 + +四个 core 模块的源码组织方式不一致： +- edge-gateway：**裸导入**风格（无 ``__init__.py``，模块内 ``from point_dict.loader import ...``）； +- data-bus / rag-kb / llm-gateway：**包**风格（有 ``__init__.py`` + 相对导入）， + 但目录名带连字符（``data-bus``）不是合法 Python 标识符，无法直接 ``import``。 + +这里在 import 早期： +1. 把 ``edge-gateway`` 目录加入 ``sys.path``，让裸导入生效； +2. 用 ``importlib`` 把三个带连字符的包目录注册为合法包名 + （``data_bus`` / ``rag_kb`` / ``llm_gateway``），让相对导入在其子模块内生效。 + +之后测试用例即可： + from point_dict.loader import Point, PointDict # edge（裸导入） + from data_bus import BatchWriter, MemorySink # 包 + from rag_kb import RagKnowledgeBase, build_document # 包 + from llm_gateway import LLMGateway, LocalBackend # 包 +""" +from __future__ import annotations + +import importlib.util +import os +import sys + +_THIS_DIR = os.path.dirname(os.path.abspath(__file__)) +_REPO_ROOT = os.path.abspath(os.path.join(_THIS_DIR, "..", "..")) +_CORE_DIR = os.path.join(_REPO_ROOT, "core") + +# 1) edge-gateway：裸导入，把目录加入 sys.path +_EDGE_DIR = os.path.join(_CORE_DIR, "edge-gateway") +if os.path.isdir(_EDGE_DIR) and _EDGE_DIR not in sys.path: + sys.path.insert(0, _EDGE_DIR) + + +def _register_dashed_package(pkg_name: str, dir_path: str) -> None: + """把带连字符目录注册为合法包名（如 data-bus → data_bus）。 + + 通过 spec_from_file_location 显式指定 submodule_search_locations， + 使包内相对导入（``from .tdengine_schema import ...``）能正常解析。 + """ + init_file = os.path.join(dir_path, "__init__.py") + if not os.path.isfile(init_file): + return + if pkg_name in sys.modules: # 已注册则跳过 + return + spec = importlib.util.spec_from_file_location( + pkg_name, init_file, submodule_search_locations=[dir_path] + ) + if spec is None or spec.loader is None: + return + module = importlib.util.module_from_spec(spec) + sys.modules[pkg_name] = module + spec.loader.exec_module(module) + + +# 2) 三个带连字符的包：注册为下划线包名 +_register_dashed_package("data_bus", os.path.join(_CORE_DIR, "data-bus")) +_register_dashed_package("rag_kb", os.path.join(_CORE_DIR, "rag-kb")) +_register_dashed_package("llm_gateway", os.path.join(_CORE_DIR, "llm-gateway")) diff --git a/tests/e2e/test_cross_pipeline.py b/tests/e2e/test_cross_pipeline.py new file mode 100644 index 0000000..e5b73af --- /dev/null +++ b/tests/e2e/test_cross_pipeline.py @@ -0,0 +1,107 @@ +# -*- coding: utf-8 -*- +"""端到端联调用例（Issue #86）—— 跨链路集成场景。 + +验证数据流（采集 → 落库）与问答流（RAG → LLM 网关）的协同： +基于真实采集到的工艺数据，构建 RAG 知识并完成一次安全问答闭环， +模拟「采集 → 沉淀知识 → 智能问答」的完整业务闭环。 +""" +from __future__ import annotations + +# 引导加载四个内核模块（必须在测试导入前执行） +import tests.e2e._bootstrap # noqa: F401 + +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 + +from rag_kb import RagKnowledgeBase, build_document +from llm_gateway import ( + GatewayResult, + LLMGateway, + LocalBackend, + PromptRegistry, + RouteTarget, +) + + +def _collect_samples(tmpdir: str): + """跑一轮采集，返回（落库样本, 引擎健康度）。""" + 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), + ] + pd = PointDict(points) + spool = SpoolStore(spool_dir=tmpdir, cache_limit_bytes=8 * 1024 * 1024) + metrics = HealthMetrics() + engine = CollectorEngine( + point_dict=pd, + driver_slots=[("simulator", SimulatorDriver(), [])], + spool=spool, metrics=metrics, interval_ms=1000, max_pending=100_000, + ) + engine.collect_once() + + sink = MemorySink() + writer = BatchWriter(sink=sink, batch_size=10, flush_interval=0.0) + for s in spool.pending_records(): + writer.push(s) + spool.ack(s) + writer.flush() + return sink.rows, metrics + + +class CrossPipelineE2ETest(unittest.TestCase): + """采集 → 落库 → 知识沉淀 → 安全问答 全业务闭环。""" + + def test_collect_then_query_closed_loop(self) -> None: + """采集数据落库后，结合工艺知识库完成一次安全问答。""" + tmpdir = tempfile.mkdtemp(prefix="iaop_cross_") + # 1) 数据流：采集并落库 + rows, metrics = _collect_samples(tmpdir) + self.assertEqual(len(rows), 2) # 2 点位全部落库 + self.assertTrue(metrics.meets_sla()) # 采集健康度达标 + point_ids = {r["point_id"] for r in rows} + self.assertEqual(point_ids, {"CLF-01.TEMP", "CLF-01.PRES"}) + + # 2) 知识沉淀：工艺知识库（模拟由采集数据衍生的工艺规范） + kb = RagKnowledgeBase() + kb.add_documents([ + build_document( + title="1#氯化炉工艺卡", + text="1#氯化炉（CLF-01）正常炉温 850~920℃，炉压 0.2~0.4MPa。" + "CLF-01.TEMP 与 CLF-01.PRES 为关键监控点位。", + category="process", + ), + ]) + + # 3) 问答流：基于知识库回答工艺问题 + prompts = PromptRegistry() + prompts.update(name="qa", version="1.0.0", + text="工业 AI 助手回答：{query}") + gateway = LLMGateway(prompts=prompts, prompt_name="qa", + local=LocalBackend(echo_context=True)) + query = "CLF-01 的关键监控点位和正常炉温范围？" + hits = kb.search(query, top_k=3) + sources = [h.chunk.title for h in hits] + self.assertGreater(len(sources), 0) + + result = gateway.ask(query, rag_context=sources, confidence=0.95) + self.assertIsInstance(result, GatewayResult) + self.assertEqual(result.route.target, RouteTarget.LOCAL) + self.assertFalse(result.needs_human) + # 回答引用了知识库来源 + self.assertIn(sources[0], result.answer) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/e2e/test_data_pipeline.py b/tests/e2e/test_data_pipeline.py new file mode 100644 index 0000000..92ba68e --- /dev/null +++ b/tests/e2e/test_data_pipeline.py @@ -0,0 +1,183 @@ +# -*- 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() diff --git a/tests/e2e/test_query_pipeline.py b/tests/e2e/test_query_pipeline.py new file mode 100644 index 0000000..f49a53a --- /dev/null +++ b/tests/e2e/test_query_pipeline.py @@ -0,0 +1,179 @@ +# -*- coding: utf-8 -*- +"""端到端联调用例（Issue #86）—— 问答流链路。 + +验证 PRD 5.4（LLM 网关 + RAG）的端到端协作： + rag-kb 模板化知识库检索（命中文档片段 + 来源）→ llm-gateway 混合网关 + （路由 → 生成 → 溯源校验 → DLP 出站防线）。 + +不依赖真实 LLM / 向量库：全部走内存桩，可在 CI 直接执行。 +""" +from __future__ import annotations + +# 引导加载四个内核模块（必须在测试导入前执行） +import tests.e2e._bootstrap # noqa: F401 + +import unittest + +from rag_kb import RagKnowledgeBase, build_document +from llm_gateway import ( + DLP_DEFAULT_RULES, + CloudBackend, + DlpEngine, + GatewayResult, + HallucinationGuard, + LLMGateway, + LocalBackend, + PromptRegistry, + RouteTarget, + SensitivityRouter, +) + + +def _build_kb() -> RagKnowledgeBase: + """构造一个模板化知识库：工艺规范 / SOP / 国标 三类知识源。""" + kb = RagKnowledgeBase() + docs = [ + build_document( + title="氯化车间操作规程", + text="氯化车间 1#炉正常运行温度区间 850~920℃，超过 950℃ 属于超温，" + "应立即减少通氯量并检查冷却系统。停机检修须挂牌上锁。", + category="process", + ), + build_document( + title="海绵钛氯化工序 SOP", + text="氯化工序标准作业指导书：开机前确认氯气流量计归零，" + "升温阶段按 50℃/h 速率升温至 850℃。异常停机时按紧急停机程序处置。", + category="sop", + ), + build_document( + title="工业氯化工艺国家标准", + text="GB/T XXXX 工业氯化工艺安全规范：氯化炉设计压力不低于 0.6MPa，" + "操作人员持证上岗，关键参数实时记录留存不少于 3 年。", + category="standard", + ), + ] + kb.add_documents(docs) + return kb + + +def _build_gateway() -> LLMGateway: + """构造一个可运行的混合网关：注册 qa 提示词 + 默认 DLP/路由/校验。""" + prompts = PromptRegistry() + prompts.update( + name="qa", + version="1.0.0", + text="你是工业 AI 优化助手。基于知识库回答：{query}", + description="问答主提示词 v1.0.0", + ) + return LLMGateway( + dlp=DlpEngine(), + router=SensitivityRouter(), + prompts=prompts, + guard=HallucinationGuard(), + local=LocalBackend(echo_context=True), + cloud=CloudBackend(echo_context=True), + prompt_name="qa", + ) + + +class QueryPipelineE2ETest(unittest.TestCase): + """RAG 检索 → LLM 网关编排 全链路联调。""" + + @classmethod + def setUpClass(cls) -> None: + cls.kb = _build_kb() + cls.gateway = _build_gateway() + + # -- RAG 检索 ---------------------------------------------------------- + def test_kb_retrieval_returns_relevant_chunks_with_source(self) -> None: + """检索命中文档片段并带来源（引用溯源基础）。""" + hits = self.kb.search("氯化炉温度", top_k=3) + self.assertGreater(len(hits), 0) + # 命中片段必须携带来源信息（文档标题 / 类别） + for hit in hits: + self.assertIsNotNone(hit.chunk.title) + self.assertIn(hit.chunk.category, ("process", "sop", "standard")) + # 相关片段应命中"温度"相关内容 + joined = " ".join(h.chunk.text for h in hits) + self.assertIn("温度", joined) + + def test_kb_category_filter(self) -> None: + """类别过滤：仅检索 SOP 知识源。""" + from rag_kb import KnowledgeSourceKind + hits = self.kb.search("升温", top_k=5, categories=[KnowledgeSourceKind.SOP]) + self.assertGreater(len(hits), 0) + for hit in hits: + self.assertEqual(hit.chunk.category, "sop") + + # -- 问答闭环（普通问题 → 本地后端）------------------------------------ + def test_normal_question_routes_local_with_rag_context(self) -> None: + """普通工艺问题：路由到本地后端，回答含 RAG 来源引用。""" + query = "氯化车间 1#炉的正常运行温度是多少？" + hits = self.kb.search(query, top_k=3) + sources = [h.chunk.title for h in hits] + + result = self.gateway.ask(query, rag_context=sources, confidence=0.95) + + self.assertIsInstance(result, GatewayResult) + self.assertEqual(result.route.target, RouteTarget.LOCAL) + # 本地后端 echo_context 时输出含来源标记 + self.assertGreater(len(result.answer), 0) + if sources: + self.assertIn(sources[0], result.answer) + # 溯源校验通过（有来源支撑） + self.assertTrue(result.verdict.supported) + self.assertFalse(result.needs_human) + + # -- 敏感数据 DLP 拦截（fail-closed）----------------------------------- + def test_sensitive_query_blocked_by_dlp(self) -> None: + """含敏感数据的问题被 DLP 拦截 → 路由 block → 转人工。""" + # 身份证号（DLP 默认规则命中） + query = "请查询员工 110101199003078834 的工资" + result = self.gateway.ask(query, confidence=1.0) + + self.assertEqual(result.route.target, RouteTarget.BLOCK) + self.assertTrue(result.needs_human) + # block 时不调用后端，给出人工确认占位 + self.assertIn("人工", result.answer) + + # -- 脱敏/通用问题路由到云端 ------------------------------------------ + def test_generic_question_routes_cloud(self) -> None: + """通用（非敏感）问题经 DLP 放行后可路由云端（这里默认路由 local， + 需显式配置 cloud 规则才走云端）。验证默认 local 闭环正常。""" + query = "今天的天气如何？" + result = self.gateway.ask(query, confidence=0.9) + # 默认无规则命中 → local + self.assertEqual(result.route.target, RouteTarget.LOCAL) + self.assertFalse(result.needs_human) + + # -- 高利害低信度转人工 ------------------------------------------------ + def test_high_stakes_low_confidence_to_human(self) -> None: + """高利害提示词 + 低信度 → 转人工复核（幻觉防线）。""" + prompts = PromptRegistry() + prompts.update(name="alarm_explain", version="1.0.0", + text="解释报警：{query}", description="报警解释（高利害）") + gateway = LLMGateway( + prompts=prompts, + prompt_name="alarm_explain", + high_stakes_names=["alarm_explain"], + ) + result = gateway.ask("1#炉超温报警", rag_context=["氯化车间操作规程"], + confidence=0.3) # 低信度 + self.assertTrue(result.needs_human) + self.assertEqual(result.verdict.action, "human_review") + + # -- 审计可追溯 -------------------------------------------------------- + def test_audit_drain_after_query(self) -> None: + """每次 ask() 后各组件审计记录可统一导出（DLP/路由/Prompt/幻觉）。""" + self.gateway.ask("氯化炉温度区间", confidence=0.9) + audits = self.gateway.drain_audits() + self.assertIn("dlp", audits) + self.assertIn("router", audits) + self.assertIn("prompts", audits) + self.assertIn("guard", audits) + # 路由审计应有记录 + self.assertGreater(len(audits["router"]), 0) + + +if __name__ == "__main__": + unittest.main()