mirror of
https://gitee.com/xmz7891/ykxmz.git
synced 2026-10-18 18:20:43 +00:00
xmz7891
527c455911
为什么要有这个：这一轮已经栽过两次「我写了有数据」不等于「线上能用」
* 管理页 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，否则缓存会失效一次。
203 lines
7.9 KiB
Python
203 lines
7.9 KiB
Python
# -*- 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())
|