ykxmz/_build/add_orphan_test.py

# -*- coding: utf-8 -*-
"""给 test_admin_delete.py 补一个「孤儿绑定」用例。

背景：
    实测发现 delete_all 删了 0 台设备，但 relay_state.json 里还剩 9 条
    配对码绑定 —— 那些绑定的设备早就下线了（devices 里没有），
    但码还占着。用户会遇到"我这个码怎么被占了"，而管理页上
    根本看不到是哪台机器占的。

    所以 delete_all（清场）必须连这些孤儿绑定一起清。
"""
from __future__ import annotations

import sys
from pathlib import Path

TARGET = Path(r"C:\Users\shu\Desktop\deepseekagent\LanRemote\_tests"
              r"\test_admin_delete.py")

OLD = '''        print()
        print("=" * 62)
        print(f"通过 {passed} 项，失败 {failed} 项")'''

NEW = '''        print()
        print("九、孤儿绑定也要能清掉")
        # 造一个"设备下线了但码还占着"的状态：
        # 登记一台设备，然后直接把它从 devices 里摘掉（模拟超时清理），
        # 但保留 codes 里的绑定。
        post("/v1/agent/register", {"deviceId": "pc-orphan",
                                    "code": "778899", "name": "孤儿",
                                    "version": "1.4.0"})
        state_file = relay_data / "relay_state.json"
        before = json.loads(state_file.read_text(encoding="utf-8"))
        codes_before = [i.get("code") for i in before.get("codes", [])]
        check("778899 已经有绑定了", "778899" in codes_before, codes_before)

        # 制造孤儿：把设备记录摘掉，只留绑定
        import subprocess as _sp
        _sp.run([sys.executable, "-c", f"""
import json
p = {str(state_file)!r}
d = json.load(open(p))
# 保留 778899，其余清掉，模拟"只剩绑定没有设备"
d["codes"] = [c for c in d.get("codes", []) if c.get("code") == "778899"]
json.dump(d, open(p, "w"))
"""], check=False)
        # 重启中转，让它从盘上加载这个孤儿状态
        relay.terminate()
        relay.wait(timeout=8)
        relay = subprocess.Popen(
            [sys.executable, "-u", str(ROOT / "relay" / "relay.py")], env=env,
            stdout=open(work / "relay2.log", "wb"),
            stderr=subprocess.STDOUT)
        for _ in range(30):
            time.sleep(0.5)
            try:
                if get("/v1/health")[0] == 200:
                    break
            except Exception:                             # noqa: BLE001
                continue

        status, body = get("/v1/health")
        devices = json.loads(body).get("devices", [])
        check("设备列表是空的（只剩孤儿绑定）", len(devices) == 0, len(devices))

        status, again = post("/v1/agent/register", {
            "deviceId": "pc-new", "code": "778899", "name": "新设备",
            "version": "1.4.0"})
        check("孤儿绑定确实挡着这个码（复现了用户的问题）",
              again.get("ok") is False, again)

        admin_key = (relay_data / "admin_key.txt").read_text().strip()
        status, result = post("/v1/admin/delete_all",
                              headers={"X-Admin": admin_key})
        check("清场成功", result.get("ok") is True, result)
        check("报告清掉了孤儿绑定",
              result.get("orphansCleared", 0) >= 1,
              result.get("orphansCleared"))

        status, body = get("/v1/health")
        state = json.loads(state_file.read_text(encoding="utf-8"))
        check("落盘里没有残留绑定了",
              len(state.get("codes", [])) == 0, state.get("codes"))

        status, again = post("/v1/agent/register", {
            "deviceId": "pc-final", "code": "778899", "name": "终于能用了",
            "version": "1.4.0"})
        check("清场之后这个码能用了", again.get("ok") is True, again)

        print()
        print("=" * 62)
        print(f"通过 {passed} 项，失败 {failed} 项")'''


def main() -> int:
    text = TARGET.read_text(encoding="utf-8")
    if "孤儿" in text:
        print("  已经加过孤儿用例")
        return 0
    if OLD not in text:
        print("  [失败] 锚点没找到")
        return 1
    text = text.replace(OLD, NEW, 1)
    TARGET.write_text(text, encoding="utf-8", newline="\n")

    import ast
    try:
        ast.parse(text)
        print("  语法 OK，已加孤儿绑定用例")
    except SyntaxError as exc:
        print(f"  语法错误: 行 {exc.lineno}: {exc.msg}")
        return 1
    return 0


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