Haocode/tests/test_wv2_guard.py

# -*- coding: utf-8 -*-
"""
WebView2 守卫单测 —— 锁死 T0 级事故不再复发

事故链（已发生）：
  每个 MainWindow()（含 offscreen 测试）→ get_environment()
  → 无条件 `taskkill /F /IM msedgewebview2.exe`
  → 把【正在运行的生产 app】的 WebView2 浏览器进程一并杀掉
  → 它的 controller 变 disposed（set_bounds 报 0x8007139F）
  → DOM 照渲染但视觉层永久空白（“选中会话不渲染核心内容”）

覆盖：
  T1  无头环境（QT_QPA_PLATFORM=offscreen）→ 不启用 WebView2
  T2  HAOCODE_FORCE_QTWEBENGINE=1 → 不启用
  T3  桌面平台（QT_QPA_PLATFORM=windows）→ 允许
  T4  单实例锁：持有者独占；【另一个进程】拿不到（跨进程互斥，是真守卫）
  T5  offscreen 下 get_environment() 直接返回 None（根本不碰共享 profile）
  T6  同一进程重复 acquire 幂等返回 True（不会把自己锁死）

运行: PYTHONIOENCODING=utf-8 python tests/test_wv2_guard.py
"""
import os
import sys
import tempfile
import subprocess

sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

# 铁律：测试用独立锁文件，不得与正在运行的 app 争 data/app_instance.lock
# （否则断言会依赖“app 是否在跑”）
_LOCK_TMP = os.path.join(tempfile.gettempdir(), "haocode_wv2guard_%d.lock" % os.getpid())
os.environ["HAOCODE_INSTANCE_LOCK_FILE"] = _LOCK_TMP
if os.path.exists(_LOCK_TMP):
    try:
        os.remove(_LOCK_TMP)
    except Exception:
        pass

RESULTS = []


def check(name, cond, extra=""):
    RESULTS.append((name, bool(cond)))
    print(f"{'PASS' if cond else 'FAIL'}  {name}  {extra if not cond else ''}", flush=True)


import core.webview2 as W  # noqa: E402

_ORIG = dict(os.environ)

try:
    # ---------------- T1/T2/T3: 环境守卫 ----------------
    os.environ["QT_QPA_PLATFORM"] = "offscreen"
    os.environ.pop("HAOCODE_FORCE_QTWEBENGINE", None)
    check("T1.1 offscreen → 不允许 WebView2", W._wv2_allowed_here() is False)

    os.environ["QT_QPA_PLATFORM"] = "windows"
    check("T1.2 windows → 允许 WebView2", W._wv2_allowed_here() is True)

    for p in ("minimal", "minimalegl", "vnc", "wayland", "embedded", "offscreen"):
        os.environ["QT_QPA_PLATFORM"] = p
        if W._wv2_allowed_here() is not False:
            check(f"T1.3 平台 {p} 应被拒绝", False, p)
            break
    else:
        check("T1.3 非 windows 平台全部拒绝", True)

    os.environ["QT_QPA_PLATFORM"] = "windows"
    os.environ["HAOCODE_FORCE_QTWEBENGINE"] = "1"
    check("T2.1 强制回退开关生效", W._wv2_allowed_here() is False)
    os.environ.pop("HAOCODE_FORCE_QTWEBENGINE", None)

    # ---------------- T4: 跨进程单实例互斥（核心） ----------------
    os.environ["QT_QPA_PLATFORM"] = "windows"
    W._INSTANCE_LOCK["fh"] = None          # 从头开始，避免受本进程历史影响
    first = W.acquire_instance_lock()
    check("T4.1 首个 acquires 成功", first is True, str(first))
    check("T4.2 同进程重复 acquire 幂等为 True",
          W.acquire_instance_lock() is True)

    code = ("import sys; sys.path.insert(0, r'%s');"
            "import core.webview2 as W; print(W.acquire_instance_lock())"
            % os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
    r = subprocess.run([sys.executable, "-c", code], capture_output=True,
                       text=True, encoding="utf-8", errors="replace", timeout=60)
    got = (r.stdout or "").strip().splitlines()
    got = got[-1] if got else ""
    check("T4.3 另一进程拿不到锁（互斥生效）", got == "False", f"stdout={got!r} err={(r.stderr or '')[:120]}")

    # ---------------- T5: offscreen 下 get_environment 直接 None ----------------
    os.environ["QT_QPA_PLATFORM"] = "offscreen"
    W._env = None
    env = W.get_environment(None)
    check("T5.1 offscreen 下 get_environment → None（不碰共享 profile）", env is None)
    check("T5.2 被守卫拦下后 _env 仍为空", W._env is None)

    # ---------------- T6: 已有实例在跑时 → 不启用 WebView2 ----------------
    # 此时本进程仍持有锁；用子进程模拟“后来的实例”
    code2 = ("import sys; sys.path.insert(0, r'%s');"
             "import core.webview2 as W; print(W.get_environment(None) is None)"
             % os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
    r2 = subprocess.run([sys.executable, "-c", code2], capture_output=True,
                        text=True, encoding="utf-8", errors="replace",
                        env={**os.environ, "QT_QPA_PLATFORM": "windows"},
                        timeout=60)
    out2 = (r2.stdout or "").strip().splitlines()
    out2 = out2[-1] if out2 else ""
    check("T6.1 已有实例持锁 → 后来的实例拿到 None（回落 QtWebEngine，不 taskkill）",
          out2 == "True", f"stdout={out2!r} err={(r2.stderr or '')[:160]}")

finally:
    try:
        fh = W._INSTANCE_LOCK.get("fh")
        if fh is not None:
            fh.close()
    except Exception:
        pass
    try:
        if os.path.exists(_LOCK_TMP):
            os.remove(_LOCK_TMP)
    except Exception:
        pass
    os.environ.clear()
    os.environ.update(_ORIG)

failed = [n for n, ok in RESULTS if not ok]
print(f"\n===== {len(RESULTS) - len(failed)}/{len(RESULTS)} PASS =====", flush=True)
print("ALL PASS" if not failed else f"FAILED: {failed}", flush=True)
sys.exit(0 if not failed else 1)