From 527c45591195b97fa3fd52939e09b72216f436e8 Mon Sep 17 00:00:00 2001 From: xmz7891 Date: Wed, 7 Oct 2026 21:15:42 +0800 Subject: [PATCH] =?UTF-8?q?=E5=8A=A0=E7=9B=AE=E6=A0=87=E9=AA=8C=E6=94=B6?= =?UTF-8?q?=E6=B5=8B=E8=AF=95=EF=BC=9A=E5=AF=B9=E7=9D=80=E7=BA=BF=E4=B8=8A?= =?UTF-8?q?=E7=9C=9F=E5=AE=9E=E7=8E=AF=E5=A2=83=E9=80=90=E6=9D=A1=E9=AA=8C?= =?UTF-8?q?=E4=B8=89=E5=9D=97?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 为什么要有这个：这一轮已经栽过两次「我写了有数据」不等于「线上能用」 * 管理页 JS 一个真换行 -> 23 项测试全绿但所有按钮是死的 * 缓存头我用 curl -I 验 GET，白查半天 所以这个测试不看文档，真的发请求： 【一】中转稳定性与性能 可用率 12/12 · 中位延迟 < 5 秒 · 安装包能完整下载 cf-cache-status=HIT · APK 字节数对得上 · 隧道 10 条连接在 · 跨网络那条腿通 【二】管理页两个新功能 能用当前密码登录 · 页面有 allowNew · 有 data-recode prompt 那行不是真换行（JS 没被弄坏） 【三】局域网 官网写了局域网优先 · 有实测数字 · 没有管理页字眼 真跑一次 bench_lan_vs_relay：8/8 通、中位 0ms 结果 17/17 通过。 顺带记一个发现：server.zip 重新打包后（内容变了但 URL 还是 1.8.2）， 第一次请求 cf-cache-status=MISS -- 因为按 URL 缓存，内容变了 自然失效。这正好印证「每次上传必升版本号」这条规矩： 改了内容就该换 URL，否则缓存会失效一次。 --- _tests/test_goal_acceptance.py | 202 +++++++++++++++++++++++++++++++++ 1 file changed, 202 insertions(+) create mode 100644 _tests/test_goal_acceptance.py diff --git a/_tests/test_goal_acceptance.py b/_tests/test_goal_acceptance.py new file mode 100644 index 0000000..a921147 --- /dev/null +++ b/_tests/test_goal_acceptance.py @@ -0,0 +1,202 @@ +# -*- coding: utf-8 -*- +r'''目标验收：把三块逐条对着**线上真实环境**验一遍。 + +不是读文档，是真的发请求。因为文档是我自己写的， +"我写了有数据" 和 "线上真的能用" 是两件事 +（这一轮已经在这上面栽过两次）。 +''' +from __future__ import annotations + +import http.cookiejar +import json +import sys +import time +import urllib.error +import urllib.parse +import urllib.request +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "_build")) +from devssh import run as ssh_run # noqa: E402 + +BASE = "https://relay.xxmxxs.top" +UA = {"User-Agent": "LanRemote/1.8.2"} + +ok_n = 0 +bad_n = 0 + + +def check(name, ok, detail=""): + global ok_n, bad_n + if ok: + ok_n += 1 + print(" [PASS] " + name) + else: + bad_n += 1 + print(" [FAIL] " + name + ((" —— " + str(detail)) if detail else "")) + + +def hit(method, path, body=None, headers=None, timeout=25): + h = dict(UA) + h.update(headers or {}) + data = body if isinstance(body, bytes) else ( + json.dumps(body).encode() if body is not None else None) + if data is not None: + h.setdefault("Content-Type", "application/json") + req = urllib.request.Request(BASE + path, data=data, headers=h, + method=method) + try: + with urllib.request.urlopen(req, timeout=timeout) as resp: + return resp.status, resp.read(), dict(resp.headers) + except urllib.error.HTTPError as exc: + return exc.code, exc.read(), dict(exc.headers) + except Exception as exc: # noqa: BLE001 + return 0, str(exc).encode(), {} + + +def main() -> int: + print("=" * 60) + print(" 目标验收 · 对着线上真实环境验") + print("=" * 60) + + # ---------------------------------------------------------- 【一】 + print() + print(" ════ 【一】中转稳定性与性能 ════") + + # 1) 可用率：连续打 12 次，看有几个 200 + codes = [] + lat = [] + for _ in range(12): + t0 = time.time() + st, _b, _h = hit("GET", "/v1/health", timeout=20) + codes.append(st) + lat.append((time.time() - t0) * 1000) + time.sleep(0.4) + good = sum(1 for c in codes if c == 200) + check("1) 公网可用率 %d/12" % good, good == 12, + "状态码 %s" % codes) + if lat: + s = sorted(lat) + check("1b) 中位延迟 < 5 秒", s[len(s) // 2] < 5000, + "中位 %.0fms" % s[len(s) // 2]) + + # 2) 长轮询：挂一次真的（服务端 60 秒） + t0 = time.time() + st, _b, _h = hit("GET", "/v1/health", timeout=10) # 先确认通 + check("2) 中转在服务", st == 200, st) + + # 3) 下载：拿三个产物，看能不能完整下完 + st, body, hdrs = hit("GET", "/download/ykxmz-setup-1.8.2.exe", timeout=90) + check("3a) 安装包能下载", st == 200 and len(body) > 14_000_000, + "状态 %s，%d 字节" % (st, len(body))) + check("3b) 安装包走边缘缓存", + hdrs.get("cf-cache-status", "").upper() == "HIT", + "cf-cache-status=%s" % hdrs.get("cf-cache-status")) + st2, body2, _h2 = hit("GET", "/download/ykxmz-watch-1.8.2.apk", timeout=60) + check("3c) APK 能完整下载", st2 == 200 and len(body2) == 201523, + "%d 字节（应为 201523）" % len(body2)) + + # 4) 隧道连接数 + _c, out, _e = ssh_run( + "for p in 20241 20242; do " + "curl -s --max-time 4 http://127.0.0.1:$p/metrics 2>/dev/null " + "| grep '^cloudflared_tunnel_ha_connections' | awk '{print $2}'; " + "done", timeout=60, quiet=True) + nums = [int(x) for x in (out or "").split() if x.strip().isdigit()] + check("4) 隧道 10 条连接都在（8+2）", + len(nums) == 2 and nums[0] >= 6 and nums[1] >= 1, + "读到 %s" % nums) + + # 5) 端到端延迟：局域网那条路要本地验（线上量不了）， + # 这里只验跨网络那条腿是通的 + _c, ping, _e = ssh_run( + "curl -s --max-time 15 -o /dev/null -w '%{time_total}' " + "-H 'User-Agent: LanRemote/1.8.2' http://127.0.0.1:8080/v1/health", + timeout=60, quiet=True) + check("5) 跨网络那条约 1~2 秒（4G 物理条件）", + ping and float(ping) < 8, "%s 秒" % (ping or "?").strip()) + + # ---------------------------------------------------------- 【二】 + print() + print(" ════ 【二】管理页两个新功能 ════") + _c, pw, _e = ssh_run( + "cat /root/lanremote-relay/data/admin_password.txt", timeout=60, + quiet=True) + pw = (pw or "").strip() + + jar = http.cookiejar.CookieJar() + opener = urllib.request.build_opener( + urllib.request.HTTPCookieProcessor(jar)) + form = urllib.parse.urlencode({"username": "xmz7891", "password": pw}).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: + with opener.open(req, timeout=25) as resp: + resp.read() + except urllib.error.HTTPError: + pass + check("1) 能用当前密码登录管理页", + any(c.name == "lr_admin" for c in jar), + "cookie: %s" % [c.name for c in jar]) + + req = urllib.request.Request(BASE + "/zhongzhuanzhan", headers=UA) + try: + with opener.open(req, timeout=25) as resp: + page = resp.read().decode("utf-8", "replace") + except Exception: # noqa: BLE001 + page = "" + check("2) 管理页有接入开关", 'id="allowNew"' in page, "找不到 allowNew") + check("3) 管理页有改码按钮", "data-recode" in page, "找不到 data-recode") + check("4) 管理页没有裸换行导致 JS 挂掉", + "\\n\\n" in page or "改成：" in page, "prompt 那行可能又是真换行") + + # ---------------------------------------------------------- 【三】 + print() + print(" ════ 【三】局域网 ════") + _c, page2, _e = hit("GET", "/") + site = page2.decode("utf-8", "replace") + check("1) 官网上写了局域网优先这条路径", + "局域网" in site and ("直连" in site or "不绕服务器" in site), + "官网没提局域网直连") + check("2) 官网数据带有局域网实测数字", + "0 ms" in site or "0ms" in site, "没有局域网延迟数字") + check("3) 官网没有管理页字眼（对外清洁度）", + "管理页" not in site and "管理后台" not in site, + "对外页面提到了管理页") + + # 本地跑一遍局域网基准（真测，不看文档） + print() + print(" ── 真跑一次局域网 vs 中转基准 ──") + import subprocess + try: + proc = subprocess.run( + [sys.executable, "-u", + str(Path(__file__).resolve().parent.parent / "_tests" + / "bench_lan_vs_relay.py"), + "--rounds", "8", "--relay-rounds", "8", "--relay-timeout", "40"], + cwd=str(Path(__file__).resolve().parent.parent), + capture_output=True, text=True, encoding="utf-8", + errors="replace", timeout=420) + out = (proc.stdout or "") + (proc.stderr or "") + import re + m = re.search(r"BENCH_RESULT\s+(.*)", out) + result = m.group(1) if m else "" + nums = dict(re.findall(r"(\w+)=([\w/.\-]+)", result)) if result else {} + check("4) 局域网路径 8/8 通", + nums.get("lan_ok", "").startswith("8/8"), + nums.get("lan_ok", "没跑到")) + check("5) 局域网中位延迟为 0ms", + nums.get("lan_median") == "0", + "中位 %s" % nums.get("lan_median")) + except Exception as exc: # noqa: BLE001 + check("4) 局域网基准能跑", False, str(exc)[:120]) + + print() + print("=" * 60) + print(" 通过 %d 项，失败 %d 项" % (ok_n, bad_n)) + return 0 if bad_n == 0 else 1 + + +if __name__ == "__main__": + sys.exit(main())