ykxmz/_tests/test_admin_live.py

# -*- coding: utf-8 -*-
r'''在**真实中转**上实测【二】管理页的两个功能。

== 为什么要单独做这个测试 ==

单元测试是直接调 Relay 对象的 Python 方法，而用户是在浏览器上点。
中间隔着 HTTP、HTML、JS 三层。之前就踩过「23 项测试全绿但按钮全是死的」：
admin.py 里一个 \n 写错（Python 三引号字符串把 \n 变成真换行），
拼进 JS 就是字符串里的裸换行 -> 整段 <script> 不执行 -> 所有按钮变幽灵按键。

所以这个测试走**真实 HTTP**：
  一、页面能打开（带 X-Admin 头绕开登录页）
  二、页面上真的有 allowNew 复选框和改码按钮
  三、内嵌 JS 能通过 node --check（防"按钮全是死的"）
  四、关掉 allowNew -> 新设备注册被拒 -> 打开 -> 能注册
  五、改某台设备的配对码 -> 真的变了

== 鉴权说明（踩过的坑）==

管理页是**账号+密码登录 + Cookie**，但脚本通道是 `X-Admin` 头
（钥匙或密码都行，见 relay.py 的 _admin_gate）。
一开始我只带 `?key=` 拿到的是登录页，还以为部署的是旧版本。

测完把 allowNew 恢复成原值。
'''
from __future__ import annotations

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

sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "_build"))
from devssh import run as ssh_run                          # noqa: E402

BASE = "https://relay.xxmxxs.top"
ADMIN_PATH = "/zhongzhuanzhan"
UA = {"User-Agent": "LanRemote/1.8.2"}

passed = 0
failed = 0
KEY = ""


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


def http(method, path, body=None, timeout=25, admin=False):
    h = dict(UA)
    if admin and KEY:
        h["X-Admin"] = KEY
    data = None
    if body is not None:
        data = body if isinstance(body, bytes) else json.dumps(body).encode()
        h.setdefault("Content-Type", "application/json")
    req = urllib.request.Request(BASE + path, data=data, headers=h,
                                 method=method)
    try:
        with urllib.request.urlopen(req, timeout=timeout) as resp:
            return resp.status, resp.read(), dict(resp.headers)
    except urllib.error.HTTPError as exc:
        return exc.code, exc.read(), dict(exc.headers)
    except Exception as exc:                              # noqa: BLE001
        return 0, str(exc).encode(), {}


def jbody(raw, default=None):
    try:
        return json.loads(raw.decode("utf-8", "replace"))
    except Exception:                                     # noqa: BLE001
        return default


def main() -> int:
    global KEY
    print("=" * 58)
    print("  管理页两个新功能 · 真实中转实测")
    print("=" * 58)

    _c, out, _e = ssh_run(
        "cat /root/lanremote-relay/data/admin_key.txt 2>/dev/null",
        timeout=60, quiet=True)
    KEY = (out or "").strip().splitlines()[0] if (out or "").strip() else ""
    print("  管理钥匙: %s" % (KEY[:6] + "…" if KEY else "（拿不到）"))
    if not KEY:
        print("  拿不到钥匙就无法测管理接口")
        return 1
    print()

    # ---------------------------------------------------------- 一
    print("  ---- 一、管理页能打开（带 X-Admin）----")
    st, body, hdrs = http("GET", ADMIN_PATH, admin=True)
    html = body.decode("utf-8", "replace")
    check("管理页返回 200", st == 200, "实际 %d" % st)
    check("返回的是管理页不是登录页",
          "登录" not in html[:600] or "allowNew" in html,
          "前 200 字含「登录」：%r" % html[:120])
    check("页面里有 allowNew 复选框", 'id="allowNew"' in html,
          "找不到 id=allowNew（只有 %d 字节）" % len(html))
    check("页面里有改码按钮", "data-recode" in html, "找不到 data-recode")

    # ---------------------------------------------------------- 二
    print("  ---- 二、内嵌 JS 能被解析（防按钮全是死的）----")
    scripts = re.findall(r"<script[^>]*>(.*?)</script>", html, re.S)
    inline = [s for s in scripts if s.strip()]
    check("页面上有内嵌脚本", bool(inline), "一个 <script> 都没有")
    if inline:
        ok_all = True
        err = ""
        for i, js in enumerate(inline):
            with tempfile.NamedTemporaryFile("w", suffix=".js", delete=False,
                                             encoding="utf-8") as fh:
                fh.write(js)
                path = fh.name
            try:
                proc = subprocess.run(["node", "--check", path],
                                      capture_output=True, text=True,
                                      encoding="utf-8", errors="replace",
                                      timeout=30)
                if proc.returncode != 0:
                    ok_all = False
                    err = "第 %d 段: %s" % (i + 1, (proc.stderr or "")[:200])
                    break
            except FileNotFoundError:
                err = "__skip__"
                break
            finally:
                Path(path).unlink(missing_ok=True)
        if err == "__skip__":
            print("  [SKIP] JS 语法检查（没装 node）")
        else:
            check("内嵌 JS 语法合法", ok_all, err)

    # ---------------------------------------------------------- 三
    print("  ---- 三、接入开关真的生效 ----")
    st, b, _h = http("GET", "/v1/health")
    snap = jbody(b, {})
    before = snap.get("allowNew")
    check("能从 /v1/health 读到 allowNew", before is not None, "字段不存在")

    fake = "livetest-%d" % int(time.time())
    reg = {"deviceId": fake, "code": "808080", "name": "实机测试机"}

    if before is not None:
        st, b, _h = http("POST", "/v1/admin/allow_new", {"allow": False},
                         admin=True)
        check("关掉接入开关返回 200", st == 200, "实际 %d %s" % (st, b[:120]))
        time.sleep(0.5)
        now = jbody(http("GET", "/v1/health")[1], {}).get("allowNew")
        check("关掉之后 allowNew 真的变 false", now is False, "实际 %r" % now)

        st, _b, _h = http("POST", "/v1/agent/register", reg)
        check("关掉时新设备注册被拒", st != 200, "实际 %d（本应被拒）" % st)

        st, _b, _h = http("POST", "/v1/admin/allow_new", {"allow": True},
                          admin=True)
        time.sleep(0.5)
        now2 = jbody(http("GET", "/v1/health")[1], {}).get("allowNew")
        check("打开之后 allowNew 真的变 true", now2 is True, "实际 %r" % now2)

        st, _b, _h = http("POST", "/v1/agent/register", reg)
        check("打开时新设备能注册", st == 200, "实际 %d" % st)

        if before is not None and now2 != before:
            http("POST", "/v1/admin/allow_new", {"allow": before}, admin=True)
            print("  （已把 allowNew 恢复成 %r）" % before)

    # ---------------------------------------------------------- 四
    print("  ---- 四、改配对码真的生效 ----")
    snap = jbody(http("GET", "/v1/health")[1], {})
    devices = snap.get("devices", [])
    mine = [d for d in devices if str(d.get("deviceId", "")).startswith("livetest-")]
    check("能看到刚注册的测试设备", bool(mine),
          "health 里共 %d 台设备，没有 livetest-" % len(devices))
    if mine:
        did = mine[0]["deviceId"]
        old = mine[0].get("code")
        new = "707070"
        st, b, _h = http("POST", "/v1/admin/set_code",
                         {"deviceId": did, "code": new}, admin=True)
        check("改码请求返回 200", st == 200, "实际 %d %s" % (st, b[:120]))
        time.sleep(0.5)
        devs2 = jbody(http("GET", "/v1/health")[1], {}).get("devices", [])
        after = [d for d in devs2 if d.get("deviceId") == did]
        check("改码后设备还在", bool(after), "设备从列表消失了")
        if after:
            check("配对码真的变成新的了", after[0].get("code") == new,
                  "旧 %r -> 实际 %r" % (old, after[0].get("code")))

    # ---------------------------------------------------------- 五
    print("  ---- 五、管理接口不登录要被挡 ----")
    st, b, _h = http("GET", "/v1/admin/devices")
    check("不带 X-Admin 访问管理接口返回 401", st == 401, "实际 %d" % st)

    print()
    print("=" * 58)
    print("  通过 %d 项，失败 %d 项" % (passed, failed))
    return 0 if failed == 0 else 1


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