164 lines
6.9 KiB
Python
164 lines
6.9 KiB
Python
"""全量路由回归测试：启动后遍历所有 API 路由发请求，抓 500/异常。
|
||
|
||
用法（项目根目录）：
|
||
.venv/bin/python scripts/regression_test.py
|
||
|
||
设计：
|
||
- 用 Flask test_client（不占用 18050，不影响运行中的服务）
|
||
- GET 路由全量请求（路径参数自动替换）；流式/二进制接口跳过
|
||
- 写操作（POST/PUT/DELETE）发空 body 探测：期望 400/403/404（参数校验），
|
||
返回 500 = 代码缺陷（import 遗漏/引用错误），立即报出
|
||
- 关键业务 POST 用真实参数验证（亮屏/剪贴板/测试步骤/设备池增删）
|
||
- 退出码：有失败返回 1（可接入检查流程）
|
||
|
||
背景：web/ 蓝图拆分后多次出现 import 遗漏导致接口 500（ADB_PATH/_log/常量等），
|
||
每次都是线上报错才暴露。此脚本让这类问题在改动后 30 秒内现形。
|
||
"""
|
||
import os
|
||
import sys
|
||
import signal
|
||
|
||
# 项目根加入 sys.path（脚本在 scripts/ 下运行，保证可 import web_server/config）
|
||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||
|
||
# 单接口超时保护：adb 相关接口可能慢（连接/查询设备），给足时间
|
||
signal.alarm(300)
|
||
|
||
from web_server import app # noqa: E402 （导入即完成 app 装配，与生产一致）
|
||
|
||
|
||
# 流式/二进制接口：无法用普通 GET 断言，跳过（已有单独验证路径）
|
||
_SKIP_PREFIXES = (
|
||
"/api/screen/stream", "/api/screen/thumb",
|
||
"/api/uiauto/screenshot", "/api/device/screenshot",
|
||
)
|
||
# 关键业务 POST：用真实参数验证（这些覆盖了核心链路）
|
||
_KEY_POSTS = [
|
||
("一键亮屏", "/api/device/screen_all", {"mode": "on"}),
|
||
("一键息屏", "/api/device/screen_all", {"mode": "off"}),
|
||
("剪贴板注入", "/api/tools/clipboard/set",
|
||
{"serials": ["100.100.10.11:5555"], "text": "回归测试"}),
|
||
("测试步骤(wait)", "/api/steps/test",
|
||
{"serial": "100.100.10.11:5555",
|
||
"step": {"type": "wait", "label": "等待", "params": {"max": 1, "min": 1, "probability": 100}}}),
|
||
("adb 终端执行", "/api/adb/cmd", {"cmd": "adb devices"}),
|
||
("adb 红线拦截", "/api/adb/cmd", {"cmd": "adb kill-server"}),
|
||
("应用版本查询", "/api/tools/appver", {"pkg": "jp.co.cyberagent.stf"}),
|
||
]
|
||
# 红线接口：期望被拒绝（不是 ok）
|
||
_EXPECT_REJECT = {"/api/adb/cmd"}
|
||
|
||
|
||
def _fill_path_params(rule):
|
||
"""把路由路径参数 <xxx> 替换为测试值（serial 用真实设备）。"""
|
||
url = rule
|
||
url = url.replace("<serial>", "100.100.10.11:5555")
|
||
url = url.replace("<apk_id>", "x")
|
||
url = url.replace("<action_id>", "x")
|
||
url = url.replace("<device_id>", "x")
|
||
url = url.replace("<uid>", "1")
|
||
return url
|
||
|
||
|
||
def main():
|
||
c = app.test_client()
|
||
# 登录 + CSRF
|
||
r = c.post("/login", data={"username": "admin", "password": "admin123"})
|
||
if r.status_code != 302:
|
||
print("❌ 登录失败（请检查 admin 账号）")
|
||
sys.exit(1)
|
||
token = c.get("/api/csrf").get_json()["token"]
|
||
H = {"X-CSRF-Token": token}
|
||
|
||
fails = []
|
||
warns = []
|
||
passed = 0
|
||
|
||
# ========== 1. GET 全量 ==========
|
||
rules = sorted({r.rule for r in app.url_map.iter_rules()
|
||
if r.rule.startswith("/api") and "GET" in r.methods})
|
||
for rule in rules:
|
||
if rule.startswith(_SKIP_PREFIXES):
|
||
continue
|
||
url = _fill_path_params(rule)
|
||
try:
|
||
r = c.get(url)
|
||
if r.status_code == 500:
|
||
fails.append(f"GET {rule} -> HTTP 500（代码缺陷）")
|
||
elif r.status_code >= 502:
|
||
warns.append(f"GET {rule} -> HTTP {r.status_code}（业务错误，如未配置/设备离线）")
|
||
else:
|
||
passed += 1
|
||
except Exception as e:
|
||
fails.append(f"GET {rule} -> {type(e).__name__}: {e}")
|
||
print(f"GET 路由: {len(rules)} 个（跳过 {len([r for r in rules if r.startswith(_SKIP_PREFIXES)])} 个流式/二进制）")
|
||
|
||
# ========== 1.5 未登录访问（登录重定向链路） ==========
|
||
try:
|
||
r = c.get("/")
|
||
if r.status_code == 302 and "login" in (r.headers.get("Location") or ""):
|
||
passed += 1
|
||
elif r.status_code == 200:
|
||
passed += 1 # 已登录态也可能 200（测试客户端无 cookie 时走 302）
|
||
else:
|
||
fails.append(f"未登录访问 / -> HTTP {r.status_code}（应 302 到登录页）")
|
||
except Exception as e:
|
||
fails.append(f"未登录访问 / -> {type(e).__name__}: {e}")
|
||
|
||
# ========== 2. 写操作空 body 探测（抓 import 遗漏/引用错误） ==========
|
||
write_rules = sorted({r.rule for r in app.url_map.iter_rules()
|
||
if r.rule.startswith("/api")
|
||
and (r.methods & {"POST", "PUT", "DELETE"})})
|
||
for rule in write_rules:
|
||
url = _fill_path_params(rule)
|
||
try:
|
||
methods = {m for r2 in app.url_map.iter_rules() if r2.rule == rule for m in r2.methods}
|
||
if "DELETE" in methods:
|
||
r = c.delete(url, headers=H)
|
||
elif "PUT" in methods:
|
||
r = c.put(url, json={}, headers=H)
|
||
else:
|
||
r = c.post(url, json={}, headers=H)
|
||
# 空 body 期望 400（参数校验）；403=权限/404=路径参数、500=代码缺陷
|
||
if r.status_code == 500:
|
||
fails.append(f"写操作 {rule} -> HTTP 500（空 body 应 400，500=代码缺陷）")
|
||
elif r.status_code >= 502:
|
||
warns.append(f"写操作 {rule} -> HTTP {r.status_code}（业务错误，如未配置/设备离线）")
|
||
else:
|
||
passed += 1
|
||
except Exception as e:
|
||
fails.append(f"写操作 {rule} -> {type(e).__name__}: {e}")
|
||
print(f"写操作路由: {len(write_rules)} 个（空 body 探测）")
|
||
|
||
# ========== 3. 关键业务 POST（真实参数） ==========
|
||
for name, url, body in _KEY_POSTS:
|
||
try:
|
||
r = c.post(url, json=body, headers=H)
|
||
j = r.get_json() if r.is_json else None
|
||
is_reject = url in _EXPECT_REJECT and "kill" in (body.get("cmd") or "")
|
||
good = (not j.get("ok")) if is_reject else bool(j and j.get("ok") is not False)
|
||
if good:
|
||
passed += 1
|
||
print(f" ✅ {name}: {r.status_code}")
|
||
else:
|
||
fails.append(f"关键 POST {name} -> HTTP {r.status_code} {str(j)[:80]}")
|
||
except Exception as e:
|
||
fails.append(f"关键 POST {name} -> {type(e).__name__}: {e}")
|
||
|
||
# ========== 汇总 ==========
|
||
print("-" * 50)
|
||
if warns:
|
||
print(f"⚠ 业务性提示 {len(warns)} 个（非代码缺陷）：")
|
||
for w in warns:
|
||
print(" ", w)
|
||
if fails:
|
||
print(f"❌ 发现 {len(fails)} 个代码缺陷：")
|
||
for f in fails:
|
||
print(" ", f)
|
||
sys.exit(1)
|
||
print(f"✅ 全部通过（{passed} 项检查，含 {len(rules)} GET + {len(write_rules)} 写操作探测 + {len(_KEY_POSTS)} 关键业务）")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|