209 lines
6.4 KiB
Python
209 lines
6.4 KiB
Python
# -*- 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()
|