40 lines
1.4 KiB
Python
40 lines
1.4 KiB
Python
|
# -*- coding: utf-8 -*-
|
||
|
|
"""容灾演练套件引导：复用内核模块加载逻辑。
|
||
|
|
|
||
|
|
与 tests/perf/_bootstrap.py / tests/e2e/_bootstrap.py 等价：把 edge-gateway
|
||
|
|
加入 sys.path（裸导入），把 data-bus / rag-kb / llm-gateway 注册为下划线
|
||
|
|
包名（带连字符目录）。
|
||
|
|
"""
|
||
|
|
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")
|
||
|
|
|
||
|
|
_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:
|
||
|
|
init_file = os.path.join(dir_path, "__init__.py")
|
||
|
|
if not os.path.isfile(init_file) or 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)
|
||
|
|
|
||
|
|
|
||
|
|
_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"))
|