ykxmz/_tests/test_verify_coverage_gap.py

# -*- coding: utf-8 -*-
"""证明变异 A 的盲区是**真 bug**，不是理论问题。

变异 A 是：把 register() 里的
    known_device = device_id in self.devices
改成
    known_device = code in self.codes

现有 test_admin_device_control.py 的注释明确说"不能按配对码判断"，
但我做过变异测试：注入变异 A 之后它的 23 项**全部通过** —— 没抓到。

为什么没抓到？
  它的"关掉开关后老设备仍能上线"用例里，设备用的还是**原来那个码**，
  而 self.codes[那个码] == device_id → `code in self.codes` 为真 → 放行。
  变异在这个路径上**表现不出来**。

变异只在"管理员给设备改过码"之后才现形，而这个组合用例没有被覆盖。
这个脚本把那个缺失的用例补上，**同时**在原版和变异版上跑，
用对照证明：确实是变异导致的差异，不是环境噪声。

全程只碰临时目录，不动仓库文件。
"""
from __future__ import annotations

import json
import os
import shutil
import subprocess
import sys
import tempfile
import time
import urllib.error
import urllib.request
from pathlib import Path

ROOT = Path(__file__).resolve().parent.parent
_UA = {"User-Agent": "LanRemote/1.8.2"}

DRIVER = r'''
# -*- coding: utf-8 -*-
"""在给定的 relay 源码上跑「改码 + 关开关 + 重连」这个组合用例。"""
import json, os, subprocess, sys, tempfile, time
import urllib.error, urllib.request
from pathlib import Path

ROOT = Path(r"__ROOT__")
PORT = __PORT__
BASE = f"http://127.0.0.1:{PORT}"
UA = {"User-Agent": "LanRemote/1.8.2"}
KEY = "gap-key"

work = Path(tempfile.mkdtemp(prefix="gap_"))
data = work / "d"; data.mkdir(parents=True, exist_ok=True)
env = dict(os.environ)
env.update({"RELAY_PORT": str(PORT), "RELAY_BIND": "127.0.0.1",
            "RELAY_DATA": str(data), "RELAY_DEVICE_TTL": "3600",
            "RELAY_ADMIN_KEY": KEY, "RELAY_POLL": "2",
            "PYTHONIOENCODING": "utf-8", "PYTHONUNBUFFERED": "1"})
p = subprocess.Popen([sys.executable, "-u", str(ROOT / "relay" / "relay.py")],
                     env=env, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
try:
    for _ in range(60):
        time.sleep(0.5)
        try:
            urllib.request.urlopen(BASE + "/v1/health", timeout=3).read(); break
        except Exception: pass

    def reg(dev, code, name="家里的电脑"):
        body = json.dumps({"deviceId": dev, "code": code, "name": name,
                           "version": "1.8.2"}).encode()
        r = urllib.request.Request(BASE + "/v1/agent/register", data=body,
            method="POST", headers={**UA, "Content-Type": "application/json"})
        try:
            with urllib.request.urlopen(r, timeout=10) as resp:
                return json.loads(resp.read().decode("utf-8"))
        except urllib.error.HTTPError as e:
            try: return json.loads(e.read().decode("utf-8"))
            except Exception: return {"ok": False}

    def admin(path, payload):
        body = json.dumps(payload).encode()
        r = urllib.request.Request(BASE + path, data=body, method="POST",
            headers={**UA, "X-Admin": KEY, "Content-Type": "application/json"})
        try:
            with urllib.request.urlopen(r, timeout=15) as resp:
                return json.loads(resp.read().decode("utf-8"))
        except urllib.error.HTTPError as e:
            try: return json.loads(e.read().decode("utf-8"))
            except Exception: return {}

    dev, codeA, codeB = "pc-GAP-TEST", "111222", "333444"
    steps = {}
    steps["1_初始注册"] = reg(dev, codeA).get("ok")
    steps["2_管理员改成新码"] = admin("/v1/admin/set_code",
                                    {"deviceId": dev, "code": codeB}).get("ok")
    steps["3_关掉允许新设备"] = admin("/v1/admin/allow_new",
                                    {"allow": False}).get("allowNew") is False
    # ⚠ 关键：设备**自己配置里还是旧码**（它不知道管理员在服务端改了码），
    #   所以它重连时报的仍然是 codeA。
    #   这才是两个判断依据真正分岔的地方：
    #     device_id in self.devices  -> True（设备在册）-> 放行
    #     codeA in self.codes        -> False（codes 里只有 B）-> 被挡
    r = reg(dev, codeA)
    steps["4_带旧码重连"] = r.get("ok") is True
    steps["__detail__"] = r
    print(json.dumps(steps, ensure_ascii=False))
finally:
    p.terminate()
    try: p.wait(timeout=10)
    except Exception: p.kill()
'''


def run_on(relay_src: str, port: int) -> dict:
    tmp = Path(tempfile.mkdtemp(prefix="gaprun_"))
    try:
        shutil.copytree(ROOT / "relay", tmp / "relay")
        (tmp / "relay" / "relay.py").write_text(relay_src, encoding="utf-8",
                                                newline="\n")
        driver = DRIVER.replace("__ROOT__", str(tmp)).replace(
            "__PORT__", str(port))
        f = tmp / "driver.py"
        f.write_text(driver, encoding="utf-8", newline="\n")
        proc = subprocess.run([sys.executable, "-u", str(f)],
                              capture_output=True, text=True,
                              encoding="utf-8", errors="replace", timeout=300)
        for line in reversed((proc.stdout or "").splitlines()):
            line = line.strip()
            if line.startswith("{"):
                return json.loads(line)
        return {"__error__": (proc.stdout or "")[-400:]
                             + (proc.stderr or "")[-400:]}
    finally:
        shutil.rmtree(tmp, ignore_errors=True)


def main() -> int:
    orig = (ROOT / "relay" / "relay.py").read_text(encoding="utf-8")
    anchor = "known_device = device_id in self.devices"
    if anchor not in orig:
        print("  ✗ 找不到锚点，实现可能已改")
        return 1
    mutated = orig.replace(anchor, "known_device = code in self.codes", 1)

    print("═══ 用例：注册 -> 管理员改码 -> 关掉开关 -> 设备带新码重连 ═══")
    print()
    print("  ── 原版实现 ──")
    a = run_on(orig, 8151)
    for k, v in a.items():
        if k != "__detail__":
            print(f"    {k}: {v}")
    if "__error__" in a:
        print("    错误: " + a["__error__"])

    print()
    print("  ── 注入变异 A（known_device 按配对码判断）──")
    b = run_on(mutated, 8152)
    for k, v in b.items():
        if k != "__detail__":
            print(f"    {k}: {v}")
    if "__error__" in b:
        print("    错误: " + b["__error__"])

    print()
    print("=" * 62)
    ok_a = a.get("4_带旧码重连") is True
    ok_b = b.get("4_带旧码重连") is True
    print(f"  原版「带旧码重连」: {'成功' if ok_a else '失败'}")
    print(f"  变异「带旧码重连」: {'成功' if ok_b else '失败'}")
    print()
    if ok_a and not ok_b:
        print("  ✅ 结论：变异确实造成真实功能退化（原版成功、变异失败），")
        print("     而现有 test_admin_device_control.py 的 23 项**全部通过**。")
        print("     说明它的注释描述了规则，但没有覆盖规则真正会失效的路径：")
        print("     「管理员改过码 + 设备仍报旧码 + 开关已关」这三件事同时发生。")
        return 0
    if ok_a and ok_b:
        print("  ⚠ 两版都成功 —— 这个场景下变异仍不产生差异。")
        return 0
    print("  ⚠ 原版就失败了 —— 这个组合场景在原实现下也不成立，")
    print("     需要重新审视（可能是我的用例构造有误）。")
    return 0


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