butubb
c6fda262db
在 MySQL dev 库上跑全量回归时暴露： 1. 汇总里的 ⚠/❌/✅ 在 Windows GBK 控制台抛 UnicodeEncodeError，而且崩在打印 汇总那一步 —— 探测其实全跑完了，看起来却像脚本挂了（stdout 重设为 UTF-8） 2. `100.100.10.11:5555` 是 Tailscale 时代的地址，设备早换了：路径参数与两个关键 POST 都拿它当目标 → 每次回归要等好几轮 30s adb connect 超时，还误报 "关键 POST 测试步骤(wait) 失败"。改为**运行期从设备池挑一台启用设备** 结果（对 MySQL dev 库）：99 项检查全通过（38 GET + 60 写探测 + 7 关键业务）， 仅 3 条 Tailscale 未配置的业务提示。
234 lines
10 KiB
Python
234 lines
10 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__))))
|
||
|
||
# Windows 控制台默认 GBK：汇总里的 ⚠/❌/✅ 会抛 UnicodeEncodeError，
|
||
# 而且崩在"打印汇总"这一步 —— 看起来像脚本挂了、其实探测已经全跑完
|
||
try:
|
||
sys.stdout.reconfigure(encoding="utf-8", errors="replace")
|
||
sys.stderr.reconfigure(encoding="utf-8", errors="replace")
|
||
except Exception:
|
||
pass
|
||
|
||
# 测试模式：禁用 cron 调度器（否则 test_client 会按 DB 里的任务 cron 真跑任务，
|
||
# 干扰回归/占用设备——实测 ocr关键字任务每 30 分钟触发导致回归超时）
|
||
os.environ["DISABLE_SCHEDULER"] = "1"
|
||
|
||
# 单接口超时保护：adb 相关接口可能慢（连接/查询设备），给足时间。
|
||
# Windows 没有 signal.alarm（会 AttributeError 直接崩），那里就靠调用方自己掐时间。
|
||
if hasattr(signal, "alarm"):
|
||
signal.alarm(300)
|
||
|
||
from web_server import app # noqa: E402 （导入即完成 app 装配，与生产一致）
|
||
|
||
# ================== 安全闸：绝不对生产库跑写操作 ==================
|
||
# 本脚本会发真实写请求（设备池增删、剪贴板注入、亮灭屏…）。跑在连生产库的配置上
|
||
# 就是拿正式数据做实验——直接拒绝启动。
|
||
from core import db_config # noqa: E402
|
||
|
||
if db_config.DEPLOY_ENV == "prod":
|
||
print("=" * 70)
|
||
print("拒绝运行：当前 .env 声明 DEPLOY_ENV=prod，本脚本会发真实写请求。")
|
||
print("回归测试请在 dev 环境跑（或先把 .env 指向 dev 库/临时库）。")
|
||
print("=" * 70)
|
||
sys.exit(3)
|
||
with app.app_context():
|
||
_labeled = db_config.meta_get("deployment_env")
|
||
if _labeled == "prod":
|
||
print("=" * 70)
|
||
print("拒绝运行：目标库 %s 被登记为 prod 环境（app_meta.deployment_env）。"
|
||
% db_config.describe_target())
|
||
print("回归测试会写数据，不能对生产库执行。")
|
||
print("=" * 70)
|
||
sys.exit(3)
|
||
|
||
|
||
# 流式/二进制接口：无法用普通 GET 断言，跳过（已有单独验证路径）
|
||
_SKIP_PREFIXES = (
|
||
"/api/screen/stream", "/api/screen/thumb",
|
||
"/api/uiauto/screenshot", "/api/device/screenshot",
|
||
)
|
||
# 测试用的目标设备：运行期从设备池里挑一台真实的（占位串是池空时的兜底）。
|
||
# 以前这里写死 Tailscale 时代的 100.100.10.11:5555，设备早换了 → 每次回归都要等
|
||
# 好几轮 30s 的 adb connect 超时，还会误报"关键 POST 失败"。
|
||
_TEST_SERIAL = "100.100.10.11:5555"
|
||
_SERIAL_PH = "@SERIAL@"
|
||
|
||
|
||
def _pick_serial():
|
||
"""从设备池里挑一台启用的设备当测试目标（池空则返回占位串）。"""
|
||
try:
|
||
from core.models import Device, db as _db
|
||
with app.app_context():
|
||
d = _db.session.query(Device).filter(Device.enabled.is_(True)).order_by(
|
||
Device.serial).first()
|
||
return d.serial if d else _TEST_SERIAL
|
||
except Exception:
|
||
return _TEST_SERIAL
|
||
|
||
|
||
# 关键业务 POST：用真实参数验证（这些覆盖了核心链路）
|
||
_KEY_POSTS = [
|
||
("一键亮屏", "/api/device/screen_all", {"mode": "on"}),
|
||
("一键息屏", "/api/device/screen_all", {"mode": "off"}),
|
||
("剪贴板注入", "/api/tools/clipboard/set",
|
||
{"serials": [_SERIAL_PH], "text": "回归测试"}),
|
||
("测试步骤(wait)", "/api/steps/test",
|
||
{"serial": _SERIAL_PH,
|
||
"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 _sub_serial(obj):
|
||
"""把请求体里的 @SERIAL@ 占位替换成运行期选定的设备串（支持嵌套 list/dict）。"""
|
||
if isinstance(obj, str):
|
||
return _TEST_SERIAL if obj == _SERIAL_PH else obj
|
||
if isinstance(obj, list):
|
||
return [_sub_serial(x) for x in obj]
|
||
if isinstance(obj, dict):
|
||
return {k: _sub_serial(v) for k, v in obj.items()}
|
||
return obj
|
||
|
||
|
||
def _fill_path_params(rule):
|
||
"""把路由路径参数 <xxx> 替换为测试值（serial 用真实设备）。"""
|
||
url = rule
|
||
url = url.replace("<serial>", _TEST_SERIAL)
|
||
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（真实参数） ==========
|
||
global _TEST_SERIAL
|
||
_TEST_SERIAL = _pick_serial()
|
||
print(f"测试目标设备: {_TEST_SERIAL}")
|
||
for name, url, body in _KEY_POSTS:
|
||
# 占位符换成运行期挑到的真实设备（含嵌套列表）
|
||
body = _sub_serial(body)
|
||
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()
|