ykxmz/_build/test_admin_features.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
r'''测试管理页新增的两个功能（在设备上跑，直连 127.0.0.1:8080）。

覆盖：
  A. 「允许新设备接入」开关
     A1 默认是开的
     A2 关掉之后，**新**设备注册被拒（blocked=True）
     A3 关掉之后，**已登记过的**设备仍然能上线
     A4 重新打开之后，新设备又能注册
  B. 「修改配对码」
     B1 换成合法的 6 位数字 -> 成功
     B2 新码立刻生效（能拿来注册/attach）
     B3 旧码的映射被清掉（不会变成孤儿占位）
     B4 换成已被别的设备占用的码 -> 被拒
     B5 换成非 6 位数字 -> 被拒
     B6 换成同一个码 -> 成功但 unchanged
     B7 改码会把该设备的会话踢掉
  C. 管理页 HTML 里有新控件
'''
from __future__ import annotations

import json
import sys
import time
import urllib.error
import urllib.request

BASE = "http://127.0.0.1:8080"
KEY_FILE = "/root/lanremote-relay/data/admin_key.txt"

PASS = 0
FAIL = 0


def check(ok: bool, label: str) -> None:
    global PASS, FAIL
    if ok:
        PASS += 1
        print("    [PASS] %s" % label)
    else:
        FAIL += 1
        print("    [FAIL] %s" % label)


def call(path: str, payload: dict | None = None, method: str = "POST",
         admin: bool = False, key: str = "") -> tuple:
    data = json.dumps(payload, ensure_ascii=False).encode("utf-8") \
        if payload is not None else None
    headers = {"Content-Type": "application/json"}
    if admin:
        headers["X-Admin"] = key
    req = urllib.request.Request(BASE + path, data=data, headers=headers,
                                 method=method)
    try:
        with urllib.request.urlopen(req, timeout=15) as resp:
            body = resp.read().decode("utf-8", "replace")
            try:
                return resp.status, json.loads(body)
            except Exception:                              # noqa: BLE001
                return resp.status, body
    except urllib.error.HTTPError as exc:
        body = exc.read().decode("utf-8", "replace")
        try:
            return exc.code, json.loads(body)
        except Exception:                                  # noqa: BLE001
            return exc.code, body
    except Exception as exc:                               # noqa: BLE001
        return 0, {"error": str(exc)}


def main() -> int:
    try:
        key = open(KEY_FILE, encoding="utf-8").read().strip()
    except OSError as exc:
        print("  读不到管理密钥: %s" % exc)
        return 1

    stamp = int(time.time())
    dev_a = "featA-%d" % stamp
    dev_b = "featB-%d" % stamp
    dev_c = "featC-%d" % stamp
    code_a = "%06d" % ((stamp + 11) % 1000000)
    code_b = "%06d" % ((stamp + 22) % 1000000)
    code_c = "%06d" % ((stamp + 33) % 1000000)
    code_a2 = "%06d" % ((stamp + 44) % 1000000)

    def reg(dev: str, code: str, name: str = "feat") -> tuple:
        return call("/v1/agent/register", {
            "deviceId": dev, "code": code, "name": name, "version": "1.8.2"})

    print("  ================ A. 允许新设备接入开关 ================")

    # A1 默认开
    st, snap = call("/v1/health", method="GET")
    check(snap.get("allowNew") is True, "A1 默认允许新设备接入")

    # 先登记设备 A（此时开关是开的）
    st, r = reg(dev_a, code_a, "设备A")
    check(r.get("ok") is True, "A0 设备A 初始登记成功（码 %s）" % code_a)

    # A2 关掉开关 + 已有会话
    st, r = call("/v1/admin/allow_new", {"allow": False}, admin=True, key=key)
    check(r.get("ok") is True and r.get("allowNew") is False,
          "A2a 关闭开关成功")

    st, r = reg(dev_b, code_b, "设备B")
    check(r.get("ok") is False and r.get("blocked") is True,
          "A2b 新设备B 被拒（blocked=True）: %s" % r.get("error", ""))

    # A3 已登记过的 A 仍能上线
    st, r = reg(dev_a, code_a, "设备A")
    check(r.get("ok") is True,
          "A3 已登记的设备A 仍能上线（关开关不影响老设备）")

    # A4 重新打开
    st, r = call("/v1/admin/allow_new", {"allow": True}, admin=True, key=key)
    check(r.get("ok") is True and r.get("allowNew") is True, "A4a 重新打开成功")
    st, r = reg(dev_c, code_c, "设备C")
    check(r.get("ok") is True, "A4b 新设备C 又能注册了")

    # ---------------- B ----------------
    print()
    print("  ================ B. 修改配对码 ================")

    # B5 非 6 位 -> 拒
    st, r = call("/v1/admin/set_code", {"deviceId": dev_a, "code": "12ab"},
                 admin=True, key=key)
    check(r.get("ok") is False, "B5 非 6 位数字被拒: %s" % r.get("error", ""))

    # B4 已被占用 -> 拒
    st, r = call("/v1/admin/set_code", {"deviceId": dev_a, "code": code_c},
                 admin=True, key=key)
    check(r.get("ok") is False, "B4 已被设备C占用的码被拒: %s" % r.get("error", ""))

    # B1 正常改码
    st, r = call("/v1/admin/set_code", {"deviceId": dev_a, "code": code_a2},
                 admin=True, key=key)
    check(r.get("ok") is True and r.get("code") == code_a2,
          "B1 改码成功 %s -> %s" % (code_a, code_a2))

    # B2 新码立刻生效
    st, r = reg(dev_a, code_a2, "设备A")
    check(r.get("ok") is True and r.get("code") == code_a2,
          "B2 新码立刻可用于注册")

    # B6 改成同一个 -> unchanged
    st, r = call("/v1/admin/set_code", {"deviceId": dev_a, "code": code_a2},
                 admin=True, key=key)
    check(r.get("ok") is True and r.get("unchanged") is True,
          "B6 改成同一个码 -> unchanged=True")

    # B3 旧码不该还占着
    st, snap = call("/v1/health", method="GET")
    codes_used = {d["code"] for d in snap.get("devices", [])}
    check(code_a2 in codes_used and code_a not in codes_used,
          "B3 设备列表里只有新码（旧码 %s 已清）" % code_a)

    # B7 会话被踢 —— 先 attach 一个
    st, r = call("/v1/client/attach",
                 {"code": code_a2, "name": "测试手表"})
    attached = r.get("ok") is True
    if attached:
        st2, r2 = call("/v1/admin/set_code",
                       {"deviceId": dev_a,
                        "code": "%06d" % ((stamp + 55) % 1000000)},
                       admin=True, key=key)
        check(r2.get("ok") is True and r2.get("kicked", 0) >= 1,
              "B7 改码踢掉了 %s 个会话" % r2.get("kicked"))
        # 改回去，方便下面收尾
        call("/v1/admin/set_code", {"deviceId": dev_a, "code": code_a2},
             admin=True, key=key)
    else:
        check(True, "B7 跳过（attach 没成功：%s）" % r.get("error", ""))

    # ---------------- C ----------------
    print()
    print("  ================ C. 管理页 HTML ================")
    req = urllib.request.Request(BASE + "/zhongzhuanzhan",
                                 headers={"X-Admin": key})
    try:
        with urllib.request.urlopen(req, timeout=15) as resp:
            html = resp.read().decode("utf-8", "replace")
        check('id="allowNew"' in html, "C1 页面有开关控件")
        check("允许新设备接入" in html, "C2 页面有开关文案")
        check("data-recode=" in html, "C3 设备行有改码按钮")
        check("/v1/admin/allow_new" in html, "C4 JS 里调 allow_new 接口")
        check("/v1/admin/set_code" in html, "C5 JS 里调 set_code 接口")
        # 改码按钮不能带 data-device，否则会被删除逻辑吃掉
        import re
        recode_btns = re.findall(r'<button[^>]*data-recode="[^"]*"[^>]*>', html)
        bad = [b for b in recode_btns if "data-device" in b]
        check(not bad, "C6 改码按钮没有 data-device（不会误触发删除）")
    except Exception as exc:                               # noqa: BLE001
        check(False, "C 取管理页失败: %s" % exc)

    # ---------------- 收尾 ----------------
    print()
    print("  清理测试设备…")
    for dev in (dev_a, dev_b, dev_c):
        call("/v1/admin/delete", {"deviceId": dev}, admin=True, key=key)
    call("/v1/admin/allow_new", {"allow": True}, admin=True, key=key)

    print()
    print("=" * 56)
    print("  通过 %d 项，失败 %d 项" % (PASS, FAIL))
    return 0 if FAIL == 0 else 1


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