feat: 全量路由回归测试脚本 scripts/regression_test.py（GET 全量 + 写操作空 body 探测抓 import 遗漏 + 关键业务 POST，500=缺陷 502/503=业务提示）+ fix: clear_all_worker_errors 缺导入（回归脚本首跑即抓到）

2026-08-19 15:24:11 +08:00
parent 2adc54b512
commit e0c3373dd1
2 changed files with 153 additions and 1 deletions
@@ -0,0 +1,151 @@
+"""全量路由回归测试：启动后遍历所有 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)])} 个流式/二进制）")
+
+    # ========== 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()