diff --git a/scripts/regression_test.py b/scripts/regression_test.py new file mode 100644 index 0000000..b922df2 --- /dev/null +++ b/scripts/regression_test.py @@ -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): + """把路由路径参数 替换为测试值(serial 用真实设备)。""" + url = rule + url = url.replace("", "100.100.10.11:5555") + url = url.replace("", "x") + url = url.replace("", "x") + url = url.replace("", "x") + url = url.replace("", "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() diff --git a/web/monitor.py b/web/monitor.py index 31d09dd..653ec78 100644 --- a/web/monitor.py +++ b/web/monitor.py @@ -11,7 +11,8 @@ from flask_login import login_required from core import device_pool from core.adb_helper import (screenshot, list_installed_apps, adb_connect, adb_connect_light, _ADB_LOCK, _adb) -from core.device_worker import get_all_worker_status +from core.device_worker import (get_all_worker_status, clear_worker_error, + clear_all_worker_errors) from web import context from web.auth import perm_required, PERM_DEVICES from web.common import _merged_device_list, _device_screen_state