# -*- 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 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": "@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): 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())