akg-factor-bridge/test_xxl_trigger.py

# -*- coding: utf-8 -*-
"""平台触发入口（xxl.py）的单测——子进程与回调都打桩，不跑真链、不发真请求。

运行:  docker compose exec -T akg-factor-bridge python test_xxl_trigger.py
全过输出 "ALL PASS (n cases)"，任一失败退出码 1。
"""
import sys
import time
import traceback

from fastapi import HTTPException

import config
import xxl

RESULTS = []


def case(name):
    def deco(fn):
        RESULTS.append((name, fn))
        return fn
    return deco


def _reset(key="test-key"):
    """每例开跑前清场：配置好 key、清任务表、装默认桩。"""
    config.XXL_TRIGGER_KEY = key
    config.XXL_CALLBACK_MAX_RETRIES = 0
    xxl._jobs.clear()
    xxl._running_id = None
    ran, sent = [], []
    xxl._run_step = lambda cmd: (ran.append(list(cmd)), (0, "ok"))[1]
    xxl._post_callback = lambda url, data: (sent.append((url, data)), "delivered http=200")[1]
    return ran, sent


def _wait_done(task_id, timeout=5.0):
    """等到任务连同结案回调都处理完（closed=True），避免断言撞上后台线程的时间差。"""
    t0 = time.time()
    while time.time() - t0 < timeout:
        if xxl._jobs[task_id].get("closed"):
            return xxl._jobs[task_id]
        time.sleep(0.02)
    raise AssertionError("任务超时未结束")


def _trigger(**kw):
    args = dict(callbackUrl=None, steps="build,plan,push-pool", date=None,
                key="test-key", x_job_key=None)
    args.update(kw)
    return xxl.trigger_daily_build(**args)


@case("没配 XXL_TRIGGER_KEY → 整组端点禁用 (403)，与决策系统同款安全默认")
def _():
    _reset(key="")
    try:
        _trigger(key="whatever")
        raise AssertionError("该 403 没 403")
    except HTTPException as e:
        assert e.status_code == 403, e


@case("key 不对 → 401；对了 → 受理并返回 task_id")
def _():
    ran, _ = _reset()
    try:
        _trigger(key="wrong")
        raise AssertionError("该 401 没 401")
    except HTTPException as e:
        assert e.status_code == 401, e
    r = _trigger()
    assert r["status"] == "accepted" and r["task_id"], r
    _wait_done(r["task_id"])


@case("三步按固定顺序跑完 → success，成功回调是空包")
def _():
    ran, sent = _reset()
    r = _trigger(callbackUrl="http://xxl/callback?logId=1")
    job = _wait_done(r["task_id"])
    assert job["state"] == "success", job
    assert [c[1] for c in ran] == ["build", "plan", "push-pool"], ran
    assert sent and sent[0][0].startswith("http://xxl/callback"), sent
    assert sent[0][1] == b"", sent          # 成功=空包，对端按成功结案


@case("中间一步失败 → 后面的步骤不跑，失败回调带 handleCode=500 与原因")
def _():
    ran, sent = _reset()
    xxl._run_step = lambda cmd: (ran.append(list(cmd)),
                                 (1, "boom") if cmd[1] == "plan" else (0, "ok"))[1]
    r = _trigger(callbackUrl="http://xxl/cb")
    job = _wait_done(r["task_id"])
    assert job["state"] == "failed" and "plan" in job["note"], job
    assert [c[1] for c in ran] == ["build", "plan"], ran      # push-pool 没跑
    body = sent[0][1].decode()
    assert "handleCode=500" in body and "plan" in body, body


@case("同一时刻只允许一条链: 在跑时再触发 → 409")
def _():
    _reset()
    gate = {"open": False}

    def slow(cmd):
        while not gate["open"]:
            time.sleep(0.01)
        return 0, "ok"
    xxl._run_step = slow
    r1 = _trigger(steps="build")
    try:
        _trigger(steps="build")
        raise AssertionError("该 409 没 409")
    except HTTPException as e:
        assert e.status_code == 409, e
    finally:
        gate["open"] = True
    _wait_done(r1["task_id"])
    r2 = _trigger(steps="build")                 # 跑完后可以再触发
    _wait_done(r2["task_id"])


@case("steps 白名单与顺序: 乱序请求被摆正, 未知步骤 400")
def _():
    ran, _ = _reset()
    r = _trigger(steps="push-pool,build")
    _wait_done(r["task_id"])
    assert [c[1] for c in ran] == ["build", "push-pool"], ran   # 顺序固定
    try:
        _trigger(steps="build,发大财")
        raise AssertionError("该 400 没 400")
    except HTTPException as e:
        assert e.status_code == 400, e


@case("date 参数透传到每一步命令")
def _():
    ran, _ = _reset()
    r = _trigger(steps="build,plan", date="2026-08-01")
    _wait_done(r["task_id"])
    for cmd in ran:
        assert cmd[-2:] == ["--date", "2026-08-01"], cmd


@case("无 callbackUrl 也能正常结案 (只记日志, 平台走状态查询)")
def _():
    _, sent = _reset()
    r = _trigger(callbackUrl=None, steps="build")
    job = _wait_done(r["task_id"])
    assert job["state"] == "success" and not sent


@case("状态接口: 能查到 state/steps/tail; 不认识的 task_id 404; 也要带 key")
def _():
    _reset()
    r = _trigger(steps="build")
    _wait_done(r["task_id"])
    st = xxl.job_status(r["task_id"], key="test-key", x_job_key=None)
    assert st["state"] == "success" and st["steps"] == ["build"] and st["tail"]
    try:
        xxl.job_status("no-such-id", key="test-key", x_job_key=None)
        raise AssertionError("该 404 没 404")
    except HTTPException as e:
        assert e.status_code == 404, e
    try:
        xxl.job_status(r["task_id"], key="wrong", x_job_key=None)
        raise AssertionError("该 401 没 401")
    except HTTPException as e:
        assert e.status_code == 401, e


@case("回调连接层失败: 重试次数用尽后放弃, 任务状态不受影响")
def _():
    _reset()
    config.XXL_CALLBACK_MAX_RETRIES = 0          # 失败一次即放弃, 免得测试睡 30 秒

    def dead(url, data):
        raise OSError("connect refused")
    xxl._post_callback = dead
    r = _trigger(callbackUrl="http://xxl/cb", steps="build")
    job = _wait_done(r["task_id"])
    assert job["state"] == "success", job        # 回调失败不改任务结论
    assert any("放弃" in ln for ln in job["tail"]), job["tail"]


# ---------------------------------------------------------------- runner
def main():
    passed, failed = 0, 0
    for name, fn in RESULTS:
        try:
            fn()
            print(f"  PASS  {name}")
            passed += 1
        except Exception:
            print(f"  FAIL  {name}")
            traceback.print_exc()
            failed += 1
    print("-" * 60)
    if failed:
        print(f"FAILED: {failed} / {passed + failed}")
        sys.exit(1)
    print(f"ALL PASS ({passed} cases)")


if __name__ == "__main__":
    main()