#!/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']*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())