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
61 lines
2.5 KiB
Python
61 lines
2.5 KiB
Python
# -*- 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"))
|