ykxmz/_tests/test_verify_mutation.py

# -*- coding: utf-8 -*-
"""变异测试：验证现有的 3e 阶段是真的能抓到 bug，还是永远通过。

做法（全程只碰临时目录，不动仓库里的任何文件）：
  1. 把 relay/ 整个复制到临时目录
  2. 在**副本**上注入一个变异（把实现改坏）
  3. 把现有测试 test_admin_device_control.py 复制一份，
     把它认的 ROOT 指向临时目录
  4. 跑它，看会不会 FAIL

如果注入变异后测试仍然通过，说明这个测试是假的（永远通过），
那它在 verify_all.ps1 里占的那一行就是虚假的安全感。

注入的变异都瞄准测试**自己声称**要守的规则：
  变异 A：known_device 按配对码判断（测试注释里明说这是错的）
  变异 B：改码后不删旧码（会留孤儿）
  变异 C：改码不踢会话
  变异 D：改码不校验新码是否被别人占用
"""
from __future__ import annotations

import re
import shutil
import subprocess
import sys
import tempfile
from pathlib import Path

ROOT = Path(__file__).resolve().parent.parent
passed = 0
failed = 0


def check(name: str, ok: bool, detail: str = "") -> bool:
    global passed, failed
    if ok:
        passed += 1
        print("  [PASS] " + name)
    else:
        failed += 1
        print("  [FAIL] " + name + (("  —— " + str(detail)) if detail else ""))
    return ok


MUTATIONS = {
    "A. known_device 改成按配对码判断":
        ('known_device = device_id in self.devices',
         'known_device = code in self.codes'),
    "B. 改码后不删旧码（留孤儿）":
        ("""            for code, owner_id in list(self.codes.items()):
                if owner_id == device_id:
                    self.codes.pop(code, None)
            self.codes[new_code] = device_id""",
         """            self.codes[new_code] = device_id"""),
    "C. 改码不踢会话":
        ("""            kicked = 0
            for sid in [t for t, s in self.sessions.items()
                        if s.device_id == device_id]:
                self.sessions.pop(sid, None)
                kicked += 1""",
         """            kicked = 0"""),
    "D. 改码不校验占用":
        ("""            owner = self.codes.get(new_code)
            if owner and owner != device_id:
                return {"ok": False,
                        "error": "这个配对码已经被别的设备占用了"}""",
         """            pass"""),
}


def run_test_with(relay_src: str, label: str) -> tuple:
    """把注入了变异的 relay.py 放到临时目录，跑现有测试。"""
    tmp = Path(tempfile.mkdtemp(prefix="mut_"))
    try:
        # 复制整个 relay 目录
        shutil.copytree(ROOT / "relay", tmp / "relay")
        (tmp / "relay" / "relay.py").write_text(relay_src, encoding="utf-8",
                                                newline="\n")
        # 复制测试并改 ROOT
        test_src = (ROOT / "_tests" /
                    "test_admin_device_control.py").read_text(encoding="utf-8")
        test_src = test_src.replace(
            'ROOT = Path(__file__).resolve().parent.parent',
            f'ROOT = Path(r"{tmp}")')
        # 端口错开，避免和别的东西撞
        test_src = test_src.replace("PORT = 8105", "PORT = 8146")
        new_test = tmp / "run_test.py"
        new_test.write_text(test_src, encoding="utf-8", newline="\n")

        proc = subprocess.run([sys.executable, "-u", str(new_test)],
                              capture_output=True, text=True, encoding="utf-8",
                              errors="replace", timeout=300)
        return proc.returncode, (proc.stdout or "") + (proc.stderr or "")
    finally:
        shutil.rmtree(tmp, ignore_errors=True)


def main() -> int:
    orig = (ROOT / "relay" / "relay.py").read_text(encoding="utf-8")

    print("═══ 0. 对照：未变异的代码，现有测试应该通过 ═══")
    rc, out = run_test_with(orig, "原版")
    tail = [l for l in out.splitlines() if "通过" in l or "失败" in l]
    check("原版代码上现有测试通过（rc=0）", rc == 0,
          f"rc={rc} {tail[-1] if tail else ''}")
    print(f"    {tail[-1] if tail else out[-200:]}")

    print()
    print("═══ 1. 逐个注入变异，看现有测试能不能抓到 ═══")
    for label, (old, new) in MUTATIONS.items():
        if old not in orig:
            check(f"变异点存在：{label}", False,
                  "源码里找不到锚点（实现可能已改）")
            continue
        # 变异 A 要小心：源码里可能有别的同名片段
        count = orig.count(old)
        mutated = orig.replace(old, new, 1)
        if mutated == orig:
            check(f"注入变异成功：{label}", False)
            continue
        print(f"    · {label}（锚点出现 {count} 次）")
        rc, out = run_test_with(mutated, label)
        caught = rc != 0
        check(f"现有测试能抓到「{label}」", caught,
              "注入变异后测试仍然通过 -> 这个测试是假的/有盲区")
        if not caught:
            print("      ⚠ 测试没抓到！输出末尾：")
            for line in out.strip().splitlines()[-6:]:
                print("        " + line)

    print()
    print("═══ 2. 现有测试的覆盖盲区：TTL 清理场景 ═══")
    src = (ROOT / "_tests" /
           "test_admin_device_control.py").read_text(encoding="utf-8")
    has_ttl = "DEVICE_TTL" in src or "last_seen" in src or "stale" in src
    check("现有测试覆盖了 DEVICE_TTL / 设备超时清理场景", has_ttl,
          "没有覆盖 —— 这正是我发现的 bug 所在")
    if not has_ttl:
        print("    ⚠ test_admin_device_control.py 里完全没有提到 "
              "DEVICE_TTL / last_seen / stale")
        print("      也就是说：它验证了「开关只挡新设备」，")
        print("      但没验证「什么算新设备」在被超时清理之后还成不成立。")

    print()
    print("=" * 62)
    print(f"  通过 {passed} 项，失败 {failed} 项")
    print("=" * 62)
    return 1 if failed else 0


if __name__ == "__main__":
    sys.exit(main())