feat(config): isolate config and database via HAOCODE_CONFIG_FILE override
core/config_paths.py resolves a single override so tests and multiple instances can run against isolated temp config/db without touching user data.
This commit is contained in:
@@ -0,0 +1,74 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""tests/_test_env.py —— 自动化测试的统一临时环境（P0-01 新增）
|
||||
|
||||
铁律（VERIFICATION.md §凭据与运行数据隔离）：
|
||||
1. 每个测试进程独立临时目录 + 最小临时配置；
|
||||
2. 在 import 任何可能间接加载 MainWindow 的模块之前设置 HAOCODE_CONFIG_FILE；
|
||||
3. 在 import MainWindow 之前把 core.db_manager._DEFAULT_DB 指向临时数据库；
|
||||
4. 写入、迁移、附件和截图产物只落到临时目录。
|
||||
|
||||
用法（放在测试文件顶部、任何 UI import 之前）：
|
||||
|
||||
from tests._test_env import isolate
|
||||
isolate("bashpanel", config={"providers": {}, "mode_switch": True})
|
||||
|
||||
说明：
|
||||
· 临时目录按 进程 pid 隔离，测试进程互不干扰；
|
||||
· 默认临时配置为最小可启动结构（空 providers + 测试 provider）；
|
||||
· 本模块不 import 任何 PyQt 模块，可在无 GUI 环境中安全调用。
|
||||
"""
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
import tempfile
|
||||
|
||||
_TESTS_DIR = os.path.dirname(os.path.abspath(__file__))
|
||||
_ROOT = os.path.dirname(_TESTS_DIR)
|
||||
if _ROOT not in sys.path:
|
||||
sys.path.insert(0, _ROOT)
|
||||
|
||||
|
||||
def default_config() -> dict:
|
||||
"""最小可启动临时配置（不含任何真实凭据）。"""
|
||||
return {
|
||||
"providers": {
|
||||
"testprov": {
|
||||
"api_key": "test-key-not-real",
|
||||
"base_url": "http://127.0.0.1:9/v1",
|
||||
"models": ["test-model"],
|
||||
"model_contexts": {"test-model": 100000},
|
||||
}
|
||||
},
|
||||
"default_provider": "testprov",
|
||||
"default_model": "test-model",
|
||||
}
|
||||
|
||||
|
||||
def isolate(tag: str = "t", config: dict | None = None) -> dict:
|
||||
"""创建临时配置 + 临时数据库并完成重定向。必须在 import MainWindow 之前调用。
|
||||
|
||||
返回 {"base": 临时目录, "config": 临时配置路径, "db": 临时数据库路径}。
|
||||
"""
|
||||
base = os.path.join(tempfile.gettempdir(),
|
||||
f"haocode_test_{tag}_{os.getpid()}")
|
||||
os.makedirs(base, exist_ok=True)
|
||||
|
||||
cfg_path = os.path.join(base, "config.json")
|
||||
with open(cfg_path, "w", encoding="utf-8") as f:
|
||||
json.dump(config if config is not None else default_config(),
|
||||
f, ensure_ascii=False, indent=2)
|
||||
|
||||
db_path = os.path.join(base, "chat_history.db")
|
||||
for p in (db_path, db_path + "-wal", db_path + "-shm"):
|
||||
if os.path.exists(p):
|
||||
os.remove(p)
|
||||
|
||||
os.environ["HAOCODE_CONFIG_FILE"] = cfg_path
|
||||
|
||||
# P1-03：QtWebEngine 独立 profile 重定向到临时目录（源码运行才在 data/webengine/）
|
||||
os.environ.setdefault("HAOCODE_WEBENGINE_PROFILE_DIR", os.path.join(base, "webengine_profile"))
|
||||
|
||||
import core.db_manager as _dbm
|
||||
_dbm._DEFAULT_DB = db_path
|
||||
|
||||
return {"base": base, "config": cfg_path, "db": db_path}
|
||||
@@ -0,0 +1,259 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""tests/test_config_isolation.py —— P0-01 配置路径与测试隔离回归
|
||||
|
||||
运行: python tests/test_config_isolation.py
|
||||
（仓库惯例：无 pytest 依赖，独立可跑；GUI 部分自动走 offscreen）
|
||||
|
||||
完成证据（REPAIR_BACKLOG.md P0-01）：
|
||||
1. 打开路径拦截器记录到的配置打开路径全部位于临时目录，真实配置路径从未被打开；
|
||||
该断言只用路径字符串比较，不读取、不散列真实配置文件；
|
||||
2. 临时配置读写用例通过，临时数据库之外没有数据库写入；
|
||||
3. 缺失配置与损坏配置各有一个回归用例：日志含明确警告、返回安全默认值、无异常；
|
||||
4. 静态扫描：源码中不存在绕过统一路径解析（core/config_paths）的运行时配置读取。
|
||||
"""
|
||||
import ast
|
||||
import contextlib
|
||||
import io
|
||||
import json
|
||||
import os
|
||||
import sqlite3
|
||||
import sys
|
||||
|
||||
_TESTS_DIR = os.path.dirname(os.path.abspath(__file__))
|
||||
_ROOT = os.path.dirname(_TESTS_DIR)
|
||||
sys.path.insert(0, _ROOT)
|
||||
|
||||
os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")
|
||||
os.environ.setdefault("HAOCODE_RENDER", "software")
|
||||
os.environ.setdefault("QTWEBENGINE_CHROMIUM_FLAGS", "--disable-gpu")
|
||||
os.environ.setdefault("PYTHONIOENCODING", "utf-8")
|
||||
|
||||
from tests._test_env import isolate, default_config # noqa: E402
|
||||
|
||||
_TMP = isolate("cfgiso") # 必须在 import MainWindow 之前
|
||||
_REAL_CONFIG = os.path.join(_ROOT, "data", "config.json") # 仅作路径字符串，永不打开
|
||||
|
||||
ok = True
|
||||
|
||||
|
||||
def check(name, cond, extra=""):
|
||||
global ok
|
||||
print((" PASS " if cond else " FAIL ") + name +
|
||||
("" if cond else f" {extra}"), flush=True)
|
||||
if not cond:
|
||||
ok = False
|
||||
|
||||
|
||||
# ======================================================================
|
||||
# 拦截器：记录 builtins.open 与 sqlite3.connect 触碰的路径（不读取真实配置内容）
|
||||
# ======================================================================
|
||||
class OpenInterceptor:
|
||||
def __init__(self):
|
||||
self.ops = [] # (path, mode)
|
||||
self._real_open = None
|
||||
|
||||
def __enter__(self):
|
||||
self._real_open = __builtins__["open"] if isinstance(
|
||||
__builtins__, dict) else __builtins__.open
|
||||
import builtins
|
||||
builtins.open = self._open
|
||||
return self
|
||||
|
||||
def _open(self, file, mode="r", *args, **kwargs):
|
||||
try:
|
||||
p = os.fspath(file)
|
||||
except TypeError:
|
||||
p = None
|
||||
if p is not None:
|
||||
self.ops.append((os.path.abspath(str(p)), str(mode)))
|
||||
return self._real_open(file, mode, *args, **kwargs)
|
||||
|
||||
def __exit__(self, *exc):
|
||||
import builtins
|
||||
builtins.open = self._real_open
|
||||
return False
|
||||
|
||||
# ---- 断言辅助 ----
|
||||
def config_ops(self):
|
||||
# 含原子写的临时文件 config.json.tmp
|
||||
return [(p, m) for p, m in self.ops
|
||||
if os.path.basename(p).startswith("config.json")]
|
||||
|
||||
def real_config_opened(self):
|
||||
return [op for op in self.ops
|
||||
if os.path.normcase(op[0]) == os.path.normcase(_REAL_CONFIG)]
|
||||
|
||||
|
||||
class SqliteInterceptor:
|
||||
def __init__(self):
|
||||
self.paths = []
|
||||
self._real_connect = None
|
||||
|
||||
def __enter__(self):
|
||||
self._real_connect = sqlite3.connect
|
||||
sqlite3.connect = self._connect
|
||||
return self
|
||||
|
||||
def _connect(self, database, *args, **kwargs):
|
||||
self.paths.append(str(database))
|
||||
return self._real_connect(database, *args, **kwargs)
|
||||
|
||||
def __exit__(self, *exc):
|
||||
sqlite3.connect = self._real_connect
|
||||
return False
|
||||
|
||||
|
||||
# ======================================================================
|
||||
# 1) 静态扫描：不得存在绕过统一路径解析的配置读取
|
||||
# （源码中不允许出现 "config.json" 字符串字面量，core/config_paths.py 除外；
|
||||
# 文档字符串与注释不计）
|
||||
# ======================================================================
|
||||
def _docstring_values(tree) -> set:
|
||||
vals = set()
|
||||
|
||||
def _first_expr_body(body):
|
||||
if body and isinstance(body[0], ast.Expr) and \
|
||||
isinstance(body[0].value, ast.Constant) and \
|
||||
isinstance(body[0].value.value, str):
|
||||
vals.add(body[0].value.value)
|
||||
|
||||
_first_expr_body(tree.body)
|
||||
for node in ast.walk(tree):
|
||||
if isinstance(node, (ast.ClassDef, ast.FunctionDef, ast.AsyncFunctionDef)):
|
||||
_first_expr_body(node.body)
|
||||
return vals
|
||||
|
||||
|
||||
def static_scan():
|
||||
allowed = os.path.normcase(os.path.join(_ROOT, "core", "config_paths.py"))
|
||||
bad = []
|
||||
scan_dirs = [os.path.join(_ROOT, d) for d in ("core", "ui", "tools")]
|
||||
scan_files = [os.path.join(_ROOT, "main.py")]
|
||||
targets = list(scan_files)
|
||||
for d in scan_dirs:
|
||||
for dirpath, _dirnames, filenames in os.walk(d):
|
||||
if "__pycache__" in dirpath:
|
||||
continue
|
||||
for fn in filenames:
|
||||
if fn.endswith(".py"):
|
||||
targets.append(os.path.join(dirpath, fn))
|
||||
for path in targets:
|
||||
with open(path, "r", encoding="utf-8") as f:
|
||||
src = f.read()
|
||||
tree = ast.parse(src, filename=path)
|
||||
docstrings = _docstring_values(tree)
|
||||
for node in ast.walk(tree):
|
||||
if isinstance(node, ast.Constant) and isinstance(node.value, str) \
|
||||
and "config.json" in node.value \
|
||||
and node.value not in docstrings:
|
||||
if os.path.normcase(path) != allowed:
|
||||
bad.append(f"{os.path.relpath(path, _ROOT)}:{node.lineno}")
|
||||
return bad
|
||||
|
||||
|
||||
bad_literal = static_scan()
|
||||
check("静态：源码无绕过统一解析的 config.json 字面量（仅 core/config_paths.py 允许）",
|
||||
not bad_literal, f"发现: {bad_literal}")
|
||||
|
||||
|
||||
# ======================================================================
|
||||
# 2) 缺失配置 / 损坏配置：可见警告 + 安全默认值 + 无异常
|
||||
# ======================================================================
|
||||
from core.config_paths import config_path, load_config # noqa: E402
|
||||
import core.llm_engine as le # noqa: E402
|
||||
|
||||
saved_env = os.environ.get("HAOCODE_CONFIG_FILE")
|
||||
|
||||
_missing = os.path.join(_TMP["base"], "no_such_config.json")
|
||||
os.environ["HAOCODE_CONFIG_FILE"] = _missing
|
||||
_buf = io.StringIO()
|
||||
with contextlib.redirect_stdout(_buf):
|
||||
_res = load_config()
|
||||
_log = _buf.getvalue()
|
||||
check("缺失配置：返回安全空 dict", _res == {})
|
||||
check("缺失配置：日志含明确警告", "缺失" in _log and "config" in _log, _log)
|
||||
|
||||
_corrupt = os.path.join(_TMP["base"], "corrupt.json")
|
||||
with open(_corrupt, "w", encoding="utf-8") as f:
|
||||
f.write("{ this is not valid json")
|
||||
os.environ["HAOCODE_CONFIG_FILE"] = _corrupt
|
||||
_buf = io.StringIO()
|
||||
with contextlib.redirect_stdout(_buf):
|
||||
_res = load_config()
|
||||
_log = _buf.getvalue()
|
||||
check("损坏配置：返回安全空 dict", _res == {})
|
||||
check("损坏配置：日志含明确警告", "解析失败" in _log or "读取/解析失败" in _log, _log)
|
||||
|
||||
_notdict = os.path.join(_TMP["base"], "notdict.json")
|
||||
with open(_notdict, "w", encoding="utf-8") as f:
|
||||
f.write("[1, 2, 3]")
|
||||
os.environ["HAOCODE_CONFIG_FILE"] = _notdict
|
||||
_buf = io.StringIO()
|
||||
with contextlib.redirect_stdout(_buf):
|
||||
_res = load_config()
|
||||
_log = _buf.getvalue()
|
||||
check("非对象配置：返回安全空 dict 且有警告", _res == {} and "不是 JSON 对象" in _log, _log)
|
||||
|
||||
os.environ["HAOCODE_CONFIG_FILE"] = saved_env # 恢复临时配置
|
||||
check("环境变量恢复：config_path 回到临时目录",
|
||||
os.path.normcase(config_path()) == os.path.normcase(_TMP["config"]))
|
||||
|
||||
|
||||
# ======================================================================
|
||||
# 3) 临时配置读取（llm_engine 统一入口）
|
||||
# ======================================================================
|
||||
_cfg_now = le._load_config()
|
||||
check("llm_engine._load_config 读取临时配置（含测试 provider）",
|
||||
"testprov" in _cfg_now.get("providers", {}))
|
||||
|
||||
|
||||
# ======================================================================
|
||||
# 4) MainWindow 全链路：构造主窗口期间，配置打开/数据库连接全部落在临时目录
|
||||
# ======================================================================
|
||||
from PyQt6.QtWidgets import QApplication # noqa: E402
|
||||
from ui.views.main_window import MainWindow # noqa: E402
|
||||
import ui.views.bash_panel as bp # noqa: E402
|
||||
|
||||
app = QApplication.instance() or QApplication(sys.argv)
|
||||
|
||||
oi = OpenInterceptor()
|
||||
si = SqliteInterceptor()
|
||||
with oi, si:
|
||||
win = MainWindow()
|
||||
# 宽度记录写回（config.json 写路径）
|
||||
bp.save_panel_width(340)
|
||||
win.close()
|
||||
win.deleteLater()
|
||||
app.processEvents()
|
||||
|
||||
check("MainWindow 构造期间真实配置从未被打开",
|
||||
not oi.real_config_opened(), f"打开记录: {oi.real_config_opened()}")
|
||||
_cfg_ops = oi.config_ops()
|
||||
check("MainWindow 链路至少发生一次配置读取（路径被验证而非空转）",
|
||||
len(_cfg_ops) >= 1, f"记录: {_cfg_ops}")
|
||||
_check_base = os.path.normcase(os.path.abspath(_TMP["base"]))
|
||||
_outside = [op for op in _cfg_ops
|
||||
if not os.path.normcase(op[0]).startswith(_check_base)]
|
||||
check("全部 config.json 打开路径均位于临时目录", not _outside, f"越界: {_outside}")
|
||||
_write_ops = [op for op in _cfg_ops if "w" in op[1]]
|
||||
check("配置写路径（bash_panel 宽度）落在临时目录",
|
||||
bool(_write_ops) and not [op for op in _write_ops
|
||||
if not os.path.normcase(op[0]).startswith(_check_base)],
|
||||
f"写记录: {_write_ops}")
|
||||
check("save_panel_width 写回可被读回", bp.load_panel_width() == 340)
|
||||
|
||||
_db_outside = [p for p in si.paths
|
||||
if not (os.path.normcase(os.path.abspath(p)).startswith(_check_base)
|
||||
or p in (":memory:", ""))]
|
||||
check("临时数据库之外没有数据库连接/写入", not _db_outside, f"越界: {_db_outside}")
|
||||
check("MainWindow.config_data 来自临时配置（providers 无真实凭据）",
|
||||
set(win.config_data.get("providers", {}).keys()) == {"testprov"})
|
||||
|
||||
|
||||
# ======================================================================
|
||||
# 汇总
|
||||
# 注：QtWebEngine 在 offscreen 下的 C++ 静态析构可能在正常 sys.exit 后段错误，
|
||||
# 测试进程用 os._exit 直接退出（退出码已在上方断言中确定），不改变测试结果。
|
||||
# ======================================================================
|
||||
print(f"\n===== test_config_isolation: {'ALL PASS' if ok else 'HAS FAILURES'} =====", flush=True)
|
||||
os._exit(0 if ok else 1)
|
||||
Reference in New Issue
Block a user