260 lines
10 KiB
Python
260 lines
10 KiB
Python
# -*- 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)
|