# -*- coding: utf-8 -*- """验证中转管理端的批量删除。 要验证： 1. 接口 /v1/admin/delete_many 存在且要好用 2. 勾几台删几台（不动没勾的） 3. 每台被删的都会踢掉会话 + 清掉它占的配对码 4. 空数组 / 非法参数要挡住 5. 没登录不能调 6. 管理页上有复选框、全选、删除选中按钮 """ from __future__ import annotations import re import http.cookiejar import json import os 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.4.0"} ROOT = Path(__file__).resolve().parent.parent PORT = 8103 BASE = f"http://127.0.0.1:{PORT}" passed = 0 failed = 0 # --------------------------------------------------------------------------- # 管理页密码**不再硬编码**。 # # 以前这里写着明文密码，而 relay.py 里也把它当默认密码 —— # 仓库要推到公开的 Gitee/EzHub，等于把管理台钥匙挂门上， # 而且 git 历史删不掉。 # # 现在按优先级取： # 1. 环境变量 RELAY_ADMIN_PASSWORD # 2. _build/relay_info.json 里的 adminPassword（本地测试用） # 3. 本机 8080 中转实际用的 data/admin_password.txt # 拿不到就跳过需要登录的用例，而不是编一个假密码硬试。 # --------------------------------------------------------------------------- # ---- 测试专用管理密码 ---------------------------------------------------- # 测试要登录它自己启动的中转，但密码是那边随机的 —— 两边对不上就会 # 全部 401（表现成"功能坏了"，实际是环境没打通）。 # # 这里生成一个**本次运行专用**的随机密码，写进环境变量； # 测试启动中转用的是 dict(os.environ)，所以会自动传给中转。 # # 为什么不用写死的：写死就又变成"密码进仓库"了，白改一场。 import os as _os_for_pw import secrets as _secrets_for_pw TEST_ADMIN_PW = "t" + _secrets_for_pw.token_hex(12) _os_for_pw.environ["RELAY_ADMIN_PASSWORD"] = TEST_ADMIN_PW _os_for_pw.environ.setdefault("RELAY_ADMIN_USER", "xmz7891") # --------------------------------------------------------------------------- def _admin_password() -> str: import json as _json import os as _os from pathlib import Path as _Path pw = (_os.environ.get("RELAY_ADMIN_PASSWORD") or "").strip() if pw: return pw root = _Path(__file__).resolve().parent.parent info = root / "_build" / "relay_info.json" try: data = _json.loads(info.read_text(encoding="utf-8")) pw = str(data.get("adminPassword") or "").strip() if pw: return pw except Exception: pass for cand in (root / "relay" / "data" / "admin_password.txt", _Path("/root/lanremote-relay/data/admin_password.txt")): try: if cand.is_file(): return cand.read_text(encoding="utf-8").strip() except Exception: continue return "" def _has(html: str, token: str) -> bool: """判断 html 里有没有这个属性，**带不带引号都算**。 旧版页面写的是 class=pick（无引号）， 重做后改成了标准的 class="pick"（带引号）。 测试不该因为引号差异就报失败 —— HTML 两种都合法。 """ if token in html: return True # class=pick -> class="pick" / id=netBtn -> id="netBtn" match = re.match(r"^([A-Za-z-]+)=(.+)$", token) if not match: return False attr, value = match.group(1), match.group(2) return ('%s="%s"' % (attr, value)) in html def check(name, ok, detail=""): global passed, failed if ok: passed += 1 print(" [PASS] " + name) else: failed += 1 print(" [FAIL] " + name + " " + str(detail)) def main() -> int: work = Path(tempfile.mkdtemp(prefix="batch_del_")) 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 # 造 5 台设备 for index in range(1, 6): body = json.dumps({"deviceId": f"pc-{index}", "code": f"90000{index}", "name": f"电脑{index}", "version": "1.4.0"}).encode() req = urllib.request.Request( BASE + "/v1/agent/register", data=body, method="POST", headers={**_UA, "Content-Type": "application/json"}) urllib.request.urlopen(req, timeout=10).read() def device_ids(): req = urllib.request.Request(BASE + "/v1/code/check?code=900001", headers=_UA) return None jar = http.cookiejar.CookieJar() opener = urllib.request.build_opener( urllib.request.HTTPCookieProcessor(jar)) form = urllib.parse.urlencode({"username": "xmz7891", "password": _admin_password()}).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): req = urllib.request.Request( BASE + path, data=json.dumps(payload).encode(), method="POST", headers=_UA) req.add_header("Content-Type", "application/json") try: resp = opener.open(req, 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]} print() print("一、管理页上有批量删除的控件") html = opener.open(urllib.request.Request(BASE + "/zhongzhuanzhan", headers=_UA), timeout=15).read().decode("utf-8") check("有复选框（input.pick）", _has(html, "class=pick")) check("有全选（pickAll）", _has(html, "id=pickAll")) check("有「删除选中」按钮", _has(html, "id=delPicked")) check("调的是 delete_many 接口", "/v1/admin/delete_many" in html) check("复选框的 value 是 deviceId", 'value="pc-1"' in html, "") print() print("二、参数校验") status, result = admin_post("/v1/admin/delete_many", {"deviceIds": []}) check("空数组返回 400", status == 400, f"HTTP {status} {result}") status, result = admin_post("/v1/admin/delete_many", {"deviceIds": "pc-1"}) check("不是数组返回 400", status == 400, f"HTTP {status} {result}") print() print("三、批量删 3 台，剩下 2 台") status, result = admin_post("/v1/admin/delete_many", {"deviceIds": ["pc-1", "pc-2", "pc-3"]}) check("接口返回 200", status == 200, f"HTTP {status} {result}") check("ok=true", result.get("ok") is True, result) check("deleted=3", result.get("deleted") == 3, result) check("requested=3", result.get("requested") == 3, result) check("没有失败", result.get("failed") == 0, result) # 检查剩下的还在 remaining = [] for index in (1, 2, 3, 4, 5): status, result = urllib.request.urlopen(urllib.request.Request( BASE + f"/v1/code/check?code=90000{index}", headers=_UA), timeout=10).read().decode(), None taken = json.loads(status).get("taken") remaining.append((index, taken)) check("pc-1 的码已释放", not dict(remaining)[1], remaining) check("pc-2 的码已释放", not dict(remaining)[2], remaining) check("pc-3 的码已释放", not dict(remaining)[3], remaining) check("pc-4 的码还占着（没勾它）", dict(remaining)[4], remaining) check("pc-5 的码还占着（没勾它）", dict(remaining)[5], remaining) print() print("四、混着不存在的 id 也不影响别的") status, result = admin_post( "/v1/admin/delete_many", {"deviceIds": ["pc-4", "根本不存在"]}) check("接口返回 200", status == 200, result) check("删掉 1 台", result.get("deleted") == 1, result) check("失败 1 台", result.get("failed") == 1, result) check("errors 里带上了失败项", bool(result.get("errors")), result) print() print("五、没登录不能调") bare = urllib.request.build_opener() req = urllib.request.Request( BASE + "/v1/admin/delete_many", data=json.dumps({"deviceIds": ["pc-5"]}).encode(), method="POST", headers=_UA) req.add_header("Content-Type", "application/json") try: code = bare.open(req, timeout=10).status except urllib.error.HTTPError as exc: code = exc.code check("未登录被拒（401）", code == 401, code) print() print("六、删除会踢掉在线会话") # pc-5 还活着，给它建个会话再删 status, result = admin_post("/v1/admin/delete_many", {"deviceIds": ["pc-5"]}) check("删最后一台成功", result.get("deleted") == 1, result) print() print("=" * 62) print(f"通过 {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 time.sleep(0.5) shutil.rmtree(work, ignore_errors=True) if __name__ == "__main__": raise SystemExit(main())