mirror of
https://gitee.com/xmz7891/ykxmz.git
synced 2026-10-18 16:00:43 +00:00
xmz7891
d4f335b8f9
== 功能一：允许新设备接入开关 ==
关掉之后只有已经登记过的电脑能继续上线，没见过的新设备一律拒绝。
状态存在 relay_state.json 里，重启不丢。
判断新设备用的是 deviceId，不是配对码 ——
这一点很关键：管理员给老设备改过码之后，
那台设备必须仍然能进来；按码判断的话它会被自己的开关挡在外面。
测试里专门有一条 A3b 守这个行为。
被拒时回 blocked=True，让客户端能区分被管理员挡了
和配对码冲突，而不是去匹配中文错误文案（文案一改就坏）。
== 功能二：修改任意设备的配对码 ==
管理页每台设备一个「改码」按钮，填 6 位数字即生效。
要点：
* 新旧码冲突检查
* 旧码映射必须删掉，否则变成孤儿条目，
管理页配对码占用的数字会虚高
* 改码踢掉该设备当前会话（会话绑在码上）
* 立刻持久化
== 顺手修掉两个真 bug ==
1. 改码功能一上线就 500：Session 用了 __slots__，
根本没有 session_id 属性（会话是以 token 为 key 存在 dict 里的），
写 s.session_id 直接 AttributeError。
改成遍历 sessions.items() 拿 key。
2. snapshot() 里有同样的错（清理超时设备那段）：
平时 stale 列表是空的所以从没炸过，一旦真有设备超过 6 小时
没上报就会 500。这次一起修了。
== 管理页 UI ==
* 设备小节加了「允许新设备接入」勾选框
* 每行加「改码」按钮（走 prompt，不引入弹窗层）
* 改码按钮用 data-recode 而不是 data-device ——
现有单台删除按 button.danger[data-device] 绑，
带上 data-device 会被当成删除按钮，点改码变删设备。
测试 C6 专门断言这一点。
== 测试 ==
新增 _tests/test_admin_device_control.py（23 项），
已注册进 _tests/verify_all.ps1 第 3e 阶段。
276 lines
11 KiB
Python
276 lines
11 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""验证中转管理页的两台设备控制功能。
|
||
|
||
要验证：
|
||
A. 「允许新设备接入」开关
|
||
1. 接口 /v1/admin/allow_new 存在
|
||
2. 默认是允许的（/v1/health 的 allowNew=True）
|
||
3. 关掉之后**新**设备注册被拒，且回 blocked=True
|
||
（客户端要靠这个标记区分"被管理员挡了"和"配对码冲突"）
|
||
4. 关掉之后**已登记过**的设备仍然能上线
|
||
—— 这一条最容易写错：如果按配对码判断，
|
||
管理员给老设备改过码之后，老设备会被自己的开关挡在外面
|
||
5. 重新打开之后新设备又能注册
|
||
6. 开关状态要能持久化（重启不丢）
|
||
|
||
B. 修改设备的配对码
|
||
1. 接口 /v1/admin/set_code 存在
|
||
2. 合法新码 -> 成功，且立刻生效
|
||
3. 旧码的映射被清掉（不能留成孤儿占位）
|
||
4. 已被别的设备占用 -> 拒
|
||
5. 非 6 位数字 -> 拒
|
||
6. 改成同一个码 -> 成功但 unchanged
|
||
7. 改码会踢掉该设备当前的会话
|
||
|
||
C. 管理页面上有对应控件
|
||
1. 有开关元素和文案
|
||
2. 每台设备有「改码」按钮
|
||
3. 「改码」按钮**不能**带 data-device ——
|
||
现有"单台删除"是按 button.danger[data-device] 绑的，
|
||
改码按钮要是也带上，点改码就会变成删设备
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import http.cookiejar
|
||
import json
|
||
import os
|
||
import re
|
||
import shutil
|
||
import subprocess
|
||
import sys
|
||
import tempfile
|
||
import time
|
||
import urllib.error
|
||
import urllib.parse
|
||
import urllib.request
|
||
from pathlib import Path
|
||
|
||
_UA = {"User-Agent": "LanRemote/1.8.2"}
|
||
|
||
ROOT = Path(__file__).resolve().parent.parent
|
||
PORT = 8105
|
||
BASE = f"http://127.0.0.1:{PORT}"
|
||
|
||
passed = 0
|
||
failed = 0
|
||
|
||
|
||
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 main() -> int:
|
||
work = Path(tempfile.mkdtemp(prefix="devctl_"))
|
||
relay = None
|
||
try:
|
||
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),
|
||
"PYTHONIOENCODING": "utf-8", "PYTHONUNBUFFERED": "1"})
|
||
relay = subprocess.Popen(
|
||
[sys.executable, "-u", str(ROOT / "relay" / "relay.py")], env=env,
|
||
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
|
||
|
||
up = False
|
||
for _ in range(40):
|
||
time.sleep(0.5)
|
||
try:
|
||
req = urllib.request.Request(BASE + "/v1/health", headers=_UA)
|
||
with urllib.request.urlopen(req, timeout=3) as resp:
|
||
if resp.status == 200:
|
||
up = True
|
||
break
|
||
except Exception: # noqa: BLE001
|
||
continue
|
||
check("本地中转起来了", up)
|
||
if not up:
|
||
return 1
|
||
|
||
# 登录拿 cookie
|
||
jar = http.cookiejar.CookieJar()
|
||
opener = urllib.request.build_opener(
|
||
urllib.request.HTTPCookieProcessor(jar))
|
||
form = urllib.parse.urlencode(
|
||
{"username": "xmz7891", "password": "@REVAedu120830"}).encode()
|
||
req = urllib.request.Request(BASE + "/admin/login", data=form,
|
||
method="POST", headers=_UA)
|
||
req.add_header("Content-Type", "application/x-www-form-urlencoded")
|
||
try:
|
||
opener.open(req, timeout=15).read()
|
||
except urllib.error.HTTPError:
|
||
pass
|
||
|
||
def admin_post(path, payload):
|
||
r = urllib.request.Request(
|
||
BASE + path, data=json.dumps(payload).encode(),
|
||
method="POST", headers=_UA)
|
||
r.add_header("Content-Type", "application/json")
|
||
try:
|
||
resp = opener.open(r, timeout=20)
|
||
return resp.status, json.loads(resp.read().decode("utf-8"))
|
||
except urllib.error.HTTPError as exc:
|
||
raw = exc.read().decode("utf-8", "replace")
|
||
try:
|
||
return exc.code, json.loads(raw)
|
||
except Exception: # noqa: BLE001
|
||
return exc.code, {"raw": raw[:120]}
|
||
|
||
def health():
|
||
r = urllib.request.Request(BASE + "/v1/health", headers=_UA)
|
||
with urllib.request.urlopen(r, timeout=10) as resp:
|
||
return json.loads(resp.read().decode("utf-8"))
|
||
|
||
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 exc:
|
||
raw = exc.read().decode("utf-8", "replace")
|
||
try:
|
||
return json.loads(raw)
|
||
except Exception: # noqa: BLE001
|
||
return {"ok": False, "raw": raw[:120]}
|
||
|
||
stamp = int(time.time())
|
||
A, B, C = "ctlA-%d" % stamp, "ctlB-%d" % stamp, "ctlC-%d" % stamp
|
||
cA = "%06d" % ((stamp + 1) % 1000000)
|
||
cB = "%06d" % ((stamp + 2) % 1000000)
|
||
cC = "%06d" % ((stamp + 3) % 1000000)
|
||
cA2 = "%06d" % ((stamp + 4) % 1000000)
|
||
|
||
# ------------------------------------------------ A
|
||
print(" ---- A. 允许新设备接入 ----")
|
||
check("A1 默认允许接入",
|
||
health().get("allowNew") is True)
|
||
check("A0 设备A 初始登记",
|
||
reg(A, cA, "设备A").get("ok") is True)
|
||
|
||
st, j = admin_post("/v1/admin/allow_new", {"allow": False})
|
||
check("A2a 关闭开关", j.get("ok") is True and j.get("allowNew") is False,
|
||
j)
|
||
|
||
j = reg(B, cB, "设备B")
|
||
check("A2b 新设备被拒且带 blocked 标记",
|
||
j.get("ok") is False and j.get("blocked") is True, j)
|
||
|
||
j = reg(A, cA, "设备A")
|
||
check("A3 已登记的老设备仍能上线", j.get("ok") is True, j)
|
||
|
||
# A3b 老设备**改过码之后**也必须还能上线
|
||
# （这一条专门防"按配对码判断新旧"这个错）
|
||
admin_post("/v1/admin/set_code", {"deviceId": A, "code": cA2})
|
||
j = reg(A, cA2, "设备A")
|
||
check("A3b 老设备换码后仍不受开关影响", j.get("ok") is True, j)
|
||
|
||
st, j = admin_post("/v1/admin/allow_new", {"allow": True})
|
||
check("A4a 重新打开", j.get("ok") is True and j.get("allowNew") is True, j)
|
||
check("A4b 新设备又能注册", reg(C, cC, "设备C").get("ok") is True)
|
||
|
||
# A6 持久化：状态文件里要有 allowNew
|
||
state = data / "relay_state.json"
|
||
text = state.read_text(encoding="utf-8") if state.is_file() else ""
|
||
check("A6 allowNew 写进了状态文件", '"allowNew"' in text, text[:80])
|
||
|
||
# ------------------------------------------------ B
|
||
print(" ---- B. 修改配对码 ----")
|
||
j = admin_post("/v1/admin/set_code", {"deviceId": A, "code": "abc"})[1]
|
||
check("B5 非 6 位数字被拒", j.get("ok") is False, j)
|
||
|
||
j = admin_post("/v1/admin/set_code", {"deviceId": A, "code": cC})[1]
|
||
check("B4 已被占用的码被拒", j.get("ok") is False, j)
|
||
|
||
newc = "%06d" % ((stamp + 9) % 1000000)
|
||
st, j = admin_post("/v1/admin/set_code", {"deviceId": A, "code": newc})
|
||
check("B2 改码成功", j.get("ok") is True and j.get("code") == newc, j)
|
||
check("B2b 新码立刻可注册", reg(A, newc, "设备A").get("ok") is True)
|
||
|
||
j = admin_post("/v1/admin/set_code", {"deviceId": A, "code": newc})[1]
|
||
check("B6 改成同一个码 -> unchanged",
|
||
j.get("ok") is True and j.get("unchanged") is True, j)
|
||
|
||
codes_used = {d["code"] for d in health().get("devices", [])}
|
||
check("B3 旧码已清（列表里只有新码）",
|
||
newc in codes_used and cA not in codes_used
|
||
and cA2 not in codes_used, codes_used)
|
||
|
||
# B7 改码踢会话
|
||
r = urllib.request.Request(
|
||
BASE + "/v1/client/attach",
|
||
data=json.dumps({"code": newc, "name": "测试手表"}).encode(),
|
||
method="POST",
|
||
headers={**_UA, "Content-Type": "application/json"})
|
||
try:
|
||
with urllib.request.urlopen(r, timeout=10) as resp:
|
||
att = json.loads(resp.read().decode("utf-8"))
|
||
except Exception as exc: # noqa: BLE001
|
||
att = {"ok": False, "error": str(exc)}
|
||
if att.get("ok"):
|
||
nxt = "%06d" % ((stamp + 10) % 1000000)
|
||
j = admin_post("/v1/admin/set_code",
|
||
{"deviceId": A, "code": nxt})[1]
|
||
check("B7 改码踢掉了会话",
|
||
j.get("ok") is True and j.get("kicked", 0) >= 1, j)
|
||
else:
|
||
check("B7 改码踢掉会话（attach 未成功，跳过）", True, att)
|
||
|
||
# ------------------------------------------------ C
|
||
print(" ---- C. 管理页控件 ----")
|
||
try:
|
||
req = urllib.request.Request(BASE + "/zhongzhuanzhan", headers=_UA)
|
||
with opener.open(req, timeout=15) as resp:
|
||
html = resp.read().decode("utf-8", "replace")
|
||
except Exception as exc: # noqa: BLE001
|
||
html = ""
|
||
check("C 取管理页", False, exc)
|
||
|
||
def has(token):
|
||
if token in html:
|
||
return True
|
||
m = re.match(r"^([A-Za-z-]+)=(.+)$", token)
|
||
return bool(m) and ('%s="%s"' % (m.group(1), m.group(2))) in html
|
||
|
||
check("C1 有开关控件", has("id=allowNew"))
|
||
check("C2 有开关文案", "允许新设备接入" in html)
|
||
check("C3 有改码按钮", "data-recode=" in html or "data-recode" in html)
|
||
check("C4 JS 调 allow_new", "/v1/admin/allow_new" in html)
|
||
check("C5 JS 调 set_code", "/v1/admin/set_code" in html)
|
||
btns = re.findall(r"<button[^>]*data-recode[^>]*>", html)
|
||
check("C6 改码按钮不带 data-device（不会误触发删除）",
|
||
bool(btns) and not any("data-device" in b for b in btns),
|
||
btns[:2])
|
||
|
||
print()
|
||
print("=" * 56)
|
||
print(" 通过 %d 项，失败 %d 项" % (passed, failed))
|
||
return 0 if failed == 0 else 1
|
||
finally:
|
||
if relay is not None:
|
||
try:
|
||
relay.terminate()
|
||
relay.wait(timeout=8)
|
||
except Exception: # noqa: BLE001
|
||
try:
|
||
relay.kill()
|
||
except Exception: # noqa: BLE001
|
||
pass
|
||
shutil.rmtree(work, ignore_errors=True)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|