auto_control/scripts/regression_test.py

"""全量路由回归测试：启动后遍历所有 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__))))

# 测试模式：禁用 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",
)
# 关键业务 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()