From 6d11145a39274bd27c896c5630e06326f2cd275f Mon Sep 17 00:00:00 2001 From: ThearchyHelios Date: Mon, 5 Oct 2026 08:43:44 +0800 Subject: [PATCH] =?UTF-8?q?=E5=AE=9A=E6=97=B6=E4=BB=BB=E5=8A=A1=E6=8E=A5?= =?UTF-8?q?=E5=8F=A3=EF=BC=9A=E5=A2=9E=E5=88=A0=E6=94=B9=E6=9F=A5=E3=80=81?= =?UTF-8?q?=E9=A2=84=E8=A7=88=E3=80=81=E5=8F=98=E9=87=8F=E6=B8=85=E5=8D=95?= =?UTF-8?q?=E3=80=81=E7=AB=8B=E5=8D=B3=E8=BF=90=E8=A1=8C=E3=80=81=E8=A7=A6?= =?UTF-8?q?=E5=8F=91=E5=8E=86=E5=8F=B2=E5=92=8C=E6=B5=8B=E8=AF=95=20Webhoo?= =?UTF-8?q?k?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 按前后端约定的 Schedule / Fire 形状返回；Webhook 地址和签名密钥加密存放，接口只给主机和末 4 位、有没有密钥 - 创建、修改时校验名称、cron、时区、工作流存在、正式运行要有发布版本（没有就说明原因、建议改探索运行）、输入 模板和入口字段；不对回 422 {detail（中文）, field}，请求体自己解析，连 JSON 写坏了也回中文 - PATCH 带 version，条件 UPDATE 做乐观锁，对不上回 409；webhook.url 省略表示不改、空串或 null 表示删除，secret 省略不改、空串去掉；只停用、改名不重新校验工作流——工作流撤了发布也要能先把任务停下来 - 预览给下几次触发（UTC）和按下一次触发渲染的输入，带工作流时只查字段名、不查缺的必填项（用户还在填） - 立即运行和到点触发同一套规则（重叠跳过、正式运行只从当前发布版本发起），运行记在点按钮的人名下；触发时间 取到秒，同一秒连点交回同一条，不会发起两次 - 触发历史按创建时间倒序翻页，Fire 带运行现在的状态和推送记录；删除任务连同触发历史和推送记录，发起过的运行留着 --- backend/app/api/schedules.py | 520 +++++++++++++++++++++++++++- backend/tests/test_schedules_api.py | 464 +++++++++++++++++++++++++ 2 files changed, 981 insertions(+), 3 deletions(-) create mode 100644 backend/tests/test_schedules_api.py diff --git a/backend/app/api/schedules.py b/backend/app/api/schedules.py index 718eb8b8..74273857 100644 --- a/backend/app/api/schedules.py +++ b/backend/app/api/schedules.py @@ -1,9 +1,523 @@ -"""定时任务：按周期自动发起工作流运行。 +"""定时任务：按周期自动发起工作流运行，结束、失败、停在审批时推送 Webhook。 -目前只登记了路由前缀，接口由定时任务的工作包补上；后台的调度循环在 app/scheduler 里。 +形状按前后端约定（Schedule、Fire 两种对象，见 _schedule_out、_fire_outs）。Webhook 地址和签名密钥加密存放，接口 +只给主机和末 4 位（url_hint）、有没有密钥（has_secret）。请求体自己解析、自己校验：错误一律回中文 detail，表单 +据 field 把报错落到那一格（422 {detail, code, field}，同 api/coded.py）。后台的调度循环在 app/scheduler 里。 """ from __future__ import annotations -from fastapi import APIRouter +from datetime import datetime +from typing import Any + +from fastapi import APIRouter, Depends, Header, HTTPException, Query, Request, Response +from sqlalchemy import and_, delete, func, select, update +from sqlalchemy.ext.asyncio import AsyncSession + +from app.api.coded import CodedHTTPException +from app.api.runs import actor_of +from app.core.crypto import decrypt, encrypt +from app.db.base import get_session, utcnow +from app.db.models import Run, Schedule, ScheduleFire, WebhookDelivery, Workflow, WorkflowVersion +from app.scheduler import cron, service, webhook router = APIRouter(prefix="/api/schedules", tags=["schedules"]) + +#: 请求不合规（422）的机器码。field 是出错的那一项：name、workflow_id、cron、timezone、run_class、input_template、 +#: enabled、overlap、misfire_grace_s、webhook、version、limit、before +SCHEDULE_INVALID = "schedule_invalid" +_GONE = "定时任务不存在，可能已被删除" +#: 错过的触发在多少秒内补发：默认 1 小时，最长 7 天 +DEFAULT_GRACE_S = 3600 +MAX_GRACE_S = 7 * 24 * 3600 +#: 预览最多给几次 +MAX_PREVIEW = 20 + + +def _invalid(detail: str, field: str | None = None) -> CodedHTTPException: + return CodedHTTPException(422, detail, SCHEDULE_INVALID, field=field) + + +def _iso(moment: datetime | None) -> str | None: + return service.as_utc(moment).isoformat() if moment is not None else None + + +async def _body(request: Request) -> dict[str, Any]: + try: + data = await request.json() + except ValueError: + data = None + if not isinstance(data, dict): + raise _invalid("请求内容需要是 JSON 对象") + return data + + +# -------------------------------------------------------------------------- +# 逐项校验：返回规范化之后的值，不对抛 422（带 field） +# -------------------------------------------------------------------------- + + +def _name(value: Any) -> str: + text = value.strip() if isinstance(value, str) else "" + if not text: + raise _invalid("请填写定时任务的名称", "name") + if len(text) > 200: + raise _invalid("定时任务的名称不能超过 200 个字符", "name") + return text + + +def _workflow_id(value: Any) -> str: + if not isinstance(value, str) or not value.strip(): + raise _invalid("请选择要定时运行的工作流", "workflow_id") + return value.strip() + + +def _cron(value: Any) -> str: + try: + return cron.parse(value if isinstance(value, str) else "").expr + except cron.CronError as e: + raise _invalid(str(e), "cron") from e + + +def _timezone(value: Any) -> str: + try: + return cron.timezone_name(value if isinstance(value, str) else "") + except cron.TimezoneError as e: + raise _invalid(str(e), "timezone") from e + + +def _run_class(value: Any) -> str: + if value not in service.RUN_CLASSES: + raise _invalid(f"运行类别需要是 formal（正式运行）或 exploratory（探索运行），当前为「{value}」", "run_class") + return value + + +def _template(value: Any) -> dict[str, str]: + try: + return service.check_template(value) + except service.TemplateError as e: + raise _invalid(str(e), "input_template") from e + + +def _enabled(value: Any) -> bool: + if not isinstance(value, bool): + raise _invalid(f"启用状态需要是 true 或 false，当前为「{value}」", "enabled") + return value + + +def _overlap(value: Any) -> str: + if value != "skip": + raise _invalid(f"重叠处理目前只支持 skip（上一次运行还没结束时跳过这次触发），当前为「{value}」", "overlap") + return value + + +def _grace(value: Any) -> int: + if isinstance(value, bool) or not isinstance(value, int) or not 0 <= value <= MAX_GRACE_S: + raise _invalid(f"补发宽限需要是 0 到 {MAX_GRACE_S} 之间的整数秒数（最长 7 天），当前为「{value}」", + "misfire_grace_s") + return value + + +def _format(value: Any) -> str: + if value not in webhook.FORMATS: + raise _invalid(f"不支持的 Webhook 格式「{value}」，可选的格式：{'、'.join(webhook.FORMATS)}", "webhook") + return value + + +def _events(value: Any) -> list[str]: + if not isinstance(value, list) or any(e not in webhook.EVENTS for e in value): + raise _invalid("Webhook 的通知事件只能从 finished（运行完成）、failed（运行失败）、awaiting_approval（等待审批）" + f"中选择，当前为「{value}」", "webhook") + return [e for e in webhook.EVENTS if e in value] + + +def _url(value: Any) -> str: + try: + return webhook.check_url(value) + except ValueError as e: + raise _invalid(str(e), "webhook") from e + + +def _secret(value: Any) -> str | None: + if value is None or (isinstance(value, str) and not value.strip()): + return None + if not isinstance(value, str): + raise _invalid("Webhook 签名密钥需要是文本", "webhook") + return value.strip() + + +def _blank(value: Any) -> bool: + return value is None or (isinstance(value, str) and not value.strip()) + + +def _new_webhook(value: Any) -> dict[str, Any] | None: + """创建时的 webhook：null 或 url 为空表示不配；否则校验格式、地址、密钥、事件（事件省略表示三种都要）。""" + if value is None: + return None + if not isinstance(value, dict): + raise _invalid("Webhook 设置需要是一个对象，或者为空表示不推送", "webhook") + if _blank(value.get("url")): + return None + return {"format": _format(value.get("format", "generic")), "url": _url(value.get("url")), + "secret": _secret(value.get("secret")), "events": _events(value.get("events", list(webhook.EVENTS)))} + + +def _webhook_columns(hook: dict[str, Any] | None) -> dict[str, Any]: + if hook is None: + return {"webhook_format": None, "webhook_url": None, "webhook_secret": None, "webhook_events": []} + return {"webhook_format": hook["format"], "webhook_url": encrypt(hook["url"]), + "webhook_secret": encrypt(hook["secret"]) if hook["secret"] else None, "webhook_events": hook["events"]} + + +def _patched_webhook(row: Schedule, value: Any) -> dict[str, Any]: + """PATCH 的 webhook → 要写的列。null 或 url 写空串：删除整个 Webhook；url 省略：地址不变；secret 省略：密钥不变、 + 写空串：去掉密钥。""" + if value is None: + return _webhook_columns(None) + if not isinstance(value, dict): + raise _invalid("Webhook 设置需要是一个对象，或者为空表示不推送", "webhook") + if "url" in value and _blank(value.get("url")): + return _webhook_columns(None) + has_url = bool(row.webhook_url) + if "url" not in value and not has_url: + raise _invalid("请填写 Webhook 地址", "webhook") + out: dict[str, Any] = { + "webhook_format": _format(value.get("format", row.webhook_format if has_url else "generic")), + "webhook_events": _events(value.get("events", list(row.webhook_events or []) if has_url + else list(webhook.EVENTS))), + } + if "url" in value: + out["webhook_url"] = encrypt(_url(value.get("url"))) + if "secret" in value: + secret = _secret(value.get("secret")) + out["webhook_secret"] = encrypt(secret) if secret else None + return out + + +async def _check_workflow(session: AsyncSession, workflow_id: str, run_class: str, template: dict[str, str]) -> Workflow: + """工作流要在；正式运行要有发布版本；模板里的字段名对得上入口字段，必填项都有（按这次要用的那张图查）。""" + workflow = await session.get(Workflow, workflow_id) + if workflow is None: + raise _invalid("工作流不存在，可能已被删除", "workflow_id") + graph: Any = workflow.graph + if run_class == "formal": + if not workflow.published_version: + raise _invalid(f"「{workflow.name}」还没有发布版本，正式运行只能从发布版本发起。" + "请先在编排页发布，或改为探索运行", "run_class") + graph = await _published_graph(session, workflow) or graph + try: + service.check_fields(template, service.input_fields(graph), workflow.name) + except service.TemplateError as e: + raise _invalid(str(e), "input_template") from e + return workflow + + +async def _published_graph(session: AsyncSession, workflow: Workflow) -> Any: + return await session.scalar(select(WorkflowVersion.graph).where( + WorkflowVersion.workflow_id == workflow.id, WorkflowVersion.version == workflow.published_version)) + + +# -------------------------------------------------------------------------- +# 对外的形状 +# -------------------------------------------------------------------------- + + +def _description(row: Schedule) -> str: + try: + return cron.describe(row.cron, row.timezone) + except cron.CronError: + return f"按 cron 表达式 {row.cron}（{row.timezone}）" + + +def _webhook_out(row: Schedule) -> dict[str, Any] | None: + if not row.webhook_url: + return None + plain = decrypt(row.webhook_url) + return { + "format": row.webhook_format, + # 解不开（加密密钥换过）时不猜：照实说，请人重新填 + "url_hint": webhook.url_hint(plain) if plain else "（地址无法解密，请重新填写）", + "has_secret": bool(row.webhook_secret), + "events": [e for e in webhook.EVENTS if e in (row.webhook_events or [])], + } + + +async def _fire_outs(session: AsyncSession, fires: list[ScheduleFire]) -> list[dict[str, Any]]: + """Fire 对象：run_status 是运行现在的状态，notified 是这次触发的推送记录 [{event, status, at}]（按时间先后）。""" + run_ids = [f.run_id for f in fires if f.run_id] + statuses = dict((await session.execute(select(Run.id, Run.status).where(Run.id.in_(run_ids)))).all()) \ + if run_ids else {} + notified: dict[str, list[dict[str, Any]]] = {} + if fires: + rows = (await session.execute( + select(WebhookDelivery).where(WebhookDelivery.fire_id.in_([f.id for f in fires])) + .order_by(WebhookDelivery.created_at) + )).scalars() + for d in rows: + notified.setdefault(d.fire_id, []).append({"event": d.event, "status": d.status, "at": _iso(d.created_at)}) + return [{ + "id": f.id, + "fire_at": _iso(f.fire_at), + "kind": f.kind, + "status": f.status, + "run_id": f.run_id, + "run_status": statuses.get(f.run_id) if f.run_id else None, + "reason": f.reason, + "notified": notified.get(f.id, []), + "created_at": _iso(f.created_at), + } for f in fires] + + +async def _schedule_outs(session: AsyncSession, rows: list[Schedule]) -> list[dict[str, Any]]: + ids = [r.id for r in rows] + names: dict[str, str] = {} + last: dict[str, dict[str, Any]] = {} + if rows: + names = dict((await session.execute( + select(Workflow.id, Workflow.name).where(Workflow.id.in_({r.workflow_id for r in rows})))).all()) + newest = (select(ScheduleFire.schedule_id, func.max(ScheduleFire.created_at).label("at")) + .where(ScheduleFire.schedule_id.in_(ids)).group_by(ScheduleFire.schedule_id).subquery()) + fires = list((await session.execute(select(ScheduleFire).join(newest, and_( + ScheduleFire.schedule_id == newest.c.schedule_id, ScheduleFire.created_at == newest.c.at)))).scalars()) + for out, fire in zip(await _fire_outs(session, fires), fires): + last.setdefault(fire.schedule_id, out) + return [{ + "id": r.id, + "name": r.name, + "workflow_id": r.workflow_id, + # 工作流还在就给它现在的名字；删掉之后用保存时记下的那一份 + "workflow_name": names.get(r.workflow_id) or r.workflow_name, + "cron": r.cron, + "timezone": r.timezone, + "run_class": r.run_class, + "input_template": dict(r.input_template or {}), + "enabled": r.enabled, + "overlap": r.overlap, + "misfire_grace_s": r.misfire_grace_s, + "webhook": _webhook_out(r), + "next_run_at": _iso(r.next_run_at), + "last_fire": last.get(r.id), + "description_text": _description(r), + "created_by": r.created_by, + "created_at": _iso(r.created_at), + "updated_at": _iso(r.updated_at), + "version": r.version, + } for r in rows] + + +async def _schedule_out(session: AsyncSession, row: Schedule) -> dict[str, Any]: + return (await _schedule_outs(session, [row]))[0] + + +async def _get(session: AsyncSession, schedule_id: str) -> Schedule: + row = await session.get(Schedule, schedule_id) + if row is None: + raise HTTPException(404, _GONE) + return row + + +# -------------------------------------------------------------------------- +# 固定路径（要排在 /{schedule_id} 前面） +# -------------------------------------------------------------------------- + + +@router.get("/variables") +async def list_variables(tz: str | None = Query(default=None, alias="timezone")) -> list[dict[str, str]]: + """输入模板能用的变量：[{name, label, example}]。示例按现在这一刻、timezone（默认 Asia/Shanghai）触发算。""" + zone = _timezone(tz) if tz is not None else service.DEFAULT_TIMEZONE + return service.variables(tz=zone) + + +@router.post("/preview") +async def preview(request: Request, session: AsyncSession = Depends(get_session)) -> dict[str, Any]: + """{cron, timezone, input_template?, workflow_id?, n?=5} → {description_text, next: [ISO（UTC）…], inputs}。 + + inputs 是按下一次触发渲染的输入。带了 workflow_id 时对照入口字段的名字；必填项没填不在这里报——用户还在填。 + """ + body = await _body(request) + cron_expr, tz = _cron(body.get("cron")), _timezone(body.get("timezone")) + template = _template(body.get("input_template")) + n = body.get("n", 5) + n = min(max(n, 1), MAX_PREVIEW) if isinstance(n, int) and not isinstance(n, bool) else 5 + workflow_id = body.get("workflow_id") + if isinstance(workflow_id, str) and workflow_id.strip(): + workflow = await session.get(Workflow, workflow_id.strip()) + if workflow is not None: + # 还不知道按哪一类运行：发布版本和画布的字段都认 + published = await _published_graph(session, workflow) if workflow.published_version else None + fields = service.input_fields(workflow.graph) + service.input_fields(published) + try: + service.check_fields(template, fields, workflow.name, require=False) + except service.TemplateError as e: + raise _invalid(str(e), "input_template") from e + upcoming = cron.preview(cron_expr, tz, n) + first = upcoming[0] if upcoming else utcnow() + return { + "description_text": cron.describe(cron_expr, tz), + "next": [_iso(m) for m in upcoming], + "inputs": service.render_inputs(template, service.template_values(first, tz, cron_expr)), + } + + +# -------------------------------------------------------------------------- +# 增删改查 +# -------------------------------------------------------------------------- + + +@router.get("") +async def list_schedules(session: AsyncSession = Depends(get_session)) -> dict[str, Any]: + rows = list((await session.execute(select(Schedule).order_by(Schedule.created_at.desc()))).scalars()) + return {"items": await _schedule_outs(session, rows)} + + +@router.post("", status_code=201) +async def create_schedule(request: Request, session: AsyncSession = Depends(get_session), + x_actor: str | None = Header(default=None)) -> dict[str, Any]: + body = await _body(request) + name = _name(body.get("name")) + workflow_id = _workflow_id(body.get("workflow_id")) + cron_expr, tz = _cron(body.get("cron")), _timezone(body.get("timezone")) + run_class = _run_class(body.get("run_class", "formal")) + template = _template(body.get("input_template")) + enabled = _enabled(body.get("enabled", True)) + overlap = _overlap(body.get("overlap", "skip")) + grace = _grace(body.get("misfire_grace_s", DEFAULT_GRACE_S)) + hook = _new_webhook(body.get("webhook")) + workflow = await _check_workflow(session, workflow_id, run_class, template) + row = Schedule(name=name, workflow_id=workflow_id, workflow_name=workflow.name, cron=cron_expr, timezone=tz, + run_class=run_class, input_template=template, enabled=enabled, overlap=overlap, + misfire_grace_s=grace, next_run_at=service.next_run(cron_expr, tz, enabled), + created_by=actor_of(x_actor), **_webhook_columns(hook)) + session.add(row) + await session.commit() + await session.refresh(row) + return await _schedule_out(session, row) + + +@router.get("/{schedule_id}") +async def get_schedule(schedule_id: str, session: AsyncSession = Depends(get_session)) -> dict[str, Any]: + return await _schedule_out(session, await _get(session, schedule_id)) + + +@router.patch("/{schedule_id}") +async def update_schedule(schedule_id: str, request: Request, + session: AsyncSession = Depends(get_session)) -> dict[str, Any]: + """改任意可改的项，带上读到的 version；对不上回 409。只读的项（id、next_run_at、last_fire……）带了也不理。 + + 改了 cron、时区或启用状态就重算下一次触发（从现在算，不补停用期间错过的）。改了工作流、运行类别或输入模板 + 就按改后的组合重新校验；只停用、改名不查工作流——工作流后来撤了发布，也要能先把任务停下来。 + """ + body = await _body(request) + row = await _get(session, schedule_id) + version = body.get("version") + if isinstance(version, bool) or not isinstance(version, int): + raise _invalid("保存时需要带上读到的版本号 version，请刷新后重试", "version") + if version != row.version: + raise HTTPException(409, f"这个定时任务已被修改（当前是第 {row.version} 版，提交的是第 {version} 版），" + "请刷新后再保存") + values: dict[str, Any] = {} + if "name" in body: + values["name"] = _name(body["name"]) + if "workflow_id" in body: + values["workflow_id"] = _workflow_id(body["workflow_id"]) + if "cron" in body: + values["cron"] = _cron(body["cron"]) + if "timezone" in body: + values["timezone"] = _timezone(body["timezone"]) + if "run_class" in body: + values["run_class"] = _run_class(body["run_class"]) + if "input_template" in body: + values["input_template"] = _template(body["input_template"]) + if "enabled" in body: + values["enabled"] = _enabled(body["enabled"]) + if "overlap" in body: + values["overlap"] = _overlap(body["overlap"]) + if "misfire_grace_s" in body: + values["misfire_grace_s"] = _grace(body["misfire_grace_s"]) + if "webhook" in body: + values |= _patched_webhook(row, body["webhook"]) + + def merged(key: str) -> Any: + return values.get(key, getattr(row, key)) + + if {"workflow_id", "run_class", "input_template"} & values.keys(): + workflow = await _check_workflow(session, merged("workflow_id"), merged("run_class"), merged("input_template")) + values["workflow_name"] = workflow.name + if {"cron", "timezone", "enabled"} & values.keys(): + values["next_run_at"] = service.next_run(merged("cron"), merged("timezone"), merged("enabled")) + # 条件 UPDATE 才是真的乐观锁：两个人同时读到第 3 版、同时保存，只有先到的那个改得动 + result = await session.execute( + update(Schedule).where(Schedule.id == schedule_id, Schedule.version == version) + .values(**values, version=version + 1).execution_options(synchronize_session=False)) + if result.rowcount != 1: + await session.rollback() + raise HTTPException(409, "这个定时任务刚被别人修改过，请刷新后再保存") + await session.commit() + session.expire_all() + return await _schedule_out(session, await _get(session, schedule_id)) + + +@router.delete("/{schedule_id}", status_code=204) +async def delete_schedule(schedule_id: str, session: AsyncSession = Depends(get_session)) -> Response: + """删除任务，连同它的触发历史和推送记录。它发起过的运行留着（运行上的来源 id 也留着）：那是追溯凭证。""" + await _get(session, schedule_id) + # 外键是 ON DELETE CASCADE；这里照样按顺序删一遍，不依赖连接上开没开外键约束 + await session.execute(delete(WebhookDelivery).where(WebhookDelivery.schedule_id == schedule_id)) + await session.execute(delete(ScheduleFire).where(ScheduleFire.schedule_id == schedule_id)) + await session.execute(delete(Schedule).where(Schedule.id == schedule_id)) + await session.commit() + return Response(status_code=204) + + +# -------------------------------------------------------------------------- +# 立即运行、触发历史、测试 Webhook +# -------------------------------------------------------------------------- + + +@router.post("/{schedule_id}/run") +async def run_now(schedule_id: str, session: AsyncSession = Depends(get_session), + x_actor: str | None = Header(default=None)) -> dict[str, Any]: + """立即触发一次（kind=manual），停用的任务也可以。规则和到点触发相同：上一次运行没结束就跳过，正式运行只从当前 + 发布版本发起。运行记在点按钮的人名下，没有署名时记在「定时任务：<名字>」名下。返回 {fire, run_id}。 + + 触发时间取到秒：连点两下落在同一秒的，交回同一条触发记录，不会发起两次。""" + await _get(session, schedule_id) + fire = await service.fire_once(schedule_id, utcnow().replace(microsecond=0), "manual", actor=actor_of(x_actor)) + if fire is None: + raise HTTPException(404, _GONE) + return {"fire": (await _fire_outs(session, [fire]))[0], "run_id": fire.run_id} + + +@router.get("/{schedule_id}/fires") +async def list_fires(schedule_id: str, limit: str | None = None, before: str | None = None, + session: AsyncSession = Depends(get_session)) -> dict[str, Any]: + """触发历史，新的在前：{items: [Fire], next_before}。翻页把 next_before 原样作为 before 传回来；没有更多时为空。""" + await _get(session, schedule_id) + size = 20 + if limit is not None: + if not limit.strip().isdigit() or not 1 <= int(limit) <= 100: + raise _invalid(f"每页条数需要是 1 到 100 之间的整数，当前为「{limit}」", "limit") + size = int(limit) + stmt = select(ScheduleFire).where(ScheduleFire.schedule_id == schedule_id) + if before is not None: + try: + cursor = datetime.fromisoformat(before.strip().replace("Z", "+00:00")) + except ValueError as e: + raise _invalid(f"翻页位置需要是接口返回的 next_before，当前为「{before}」", "before") from e + stmt = stmt.where(ScheduleFire.created_at < service.as_utc(cursor)) + rows = list((await session.execute( + stmt.order_by(ScheduleFire.created_at.desc(), ScheduleFire.id.desc()).limit(size + 1))).scalars()) + items = rows[:size] + return {"items": await _fire_outs(session, items), + "next_before": _iso(items[-1].created_at) if len(rows) > size else None} + + +@router.post("/{schedule_id}/webhook/test") +async def test_webhook(schedule_id: str, session: AsyncSession = Depends(get_session)) -> dict[str, Any]: + """按保存的 Webhook 发一条测试消息（不重试）：{ok, http_status, message}。失败的说明里只有主机名。""" + await _get(session, schedule_id) + try: + return await webhook.send_test(schedule_id) + except KeyError as e: + raise HTTPException(404, _GONE) from e + except ValueError as e: + raise _invalid(str(e), "webhook") from e diff --git a/backend/tests/test_schedules_api.py b/backend/tests/test_schedules_api.py new file mode 100644 index 00000000..a4e36688 --- /dev/null +++ b/backend/tests/test_schedules_api.py @@ -0,0 +1,464 @@ +"""定时任务接口（api/schedules.py），按前后端约定的形状： + +- Schedule：{id, name, workflow_id, workflow_name, cron, timezone, run_class, input_template, enabled, overlap, + misfire_grace_s, webhook: {format, url_hint, has_secret, events} | null, next_run_at, last_fire, description_text, + created_by, created_at, updated_at, version}；Webhook 地址和密钥加密存放，接口只给主机和末 4 位 +- Fire：{id, fire_at, kind, status, run_id, run_status, reason, notified: [{event, status, at}], created_at} +- 创建、修改时校验 cron、时区、工作流存在、正式运行要有发布版本、输入模板；不对回 422 {detail（中文）, field} +- PATCH 带 version，对不上回 409；webhook.url 省略表示不改，空串表示删除 Webhook +- 预览下几次触发和渲染后的输入；变量清单；立即运行（同一套重叠规则）；触发历史翻页；测试 Webhook + +工作流、任务名、地址、密钥都是虚构的。 +""" +from __future__ import annotations + +import asyncio +import json +import uuid +from datetime import datetime, timedelta, timezone +from urllib.parse import quote +from zoneinfo import ZoneInfo + +import httpx +import pytest +from httpx import ASGITransport, AsyncClient +from sqlalchemy import delete, func, select + +from app.core.crypto import decrypt +from app.db.base import SessionLocal +from app.db.models import Run, Schedule, ScheduleFire, WebhookDelivery, Workflow, WorkflowVersion +from app.api import schedules as schedules_api +from app.engine.runner import run_manager +from app.main import app +from app.scheduler import webhook + +UTC = timezone.utc +HOOK = "https://open.feishu.cn/open-apis/bot/v2/hook/7c6b5a49-dddd-eeee-ffff-1234567890ab" +SECRET = "SECdemo-api-not-real-7" +SCHEDULE_KEYS = {"id", "name", "workflow_id", "workflow_name", "cron", "timezone", "run_class", "input_template", + "enabled", "overlap", "misfire_grace_s", "webhook", "next_run_at", "last_fire", "description_text", + "created_by", "created_at", "updated_at", "version"} +FIRE_KEYS = {"id", "fire_at", "kind", "status", "run_id", "run_status", "reason", "notified", "created_at"} + + +@pytest.fixture(autouse=True) +async def engine_up(): + await run_manager.setup() + async with SessionLocal() as session: + await session.execute(delete(Schedule)) + await session.commit() + yield + await run_manager.shutdown() + + +@pytest.fixture +async def client(): + async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c: + yield c + + +def node(nid, ntype, **config): + return {"id": nid, "type": ntype, "position": {"x": 0, "y": 0}, "data": {"label": nid, "config": config}} + + +def graph(*fields: dict) -> dict: + return {"nodes": [node("start", "input", fields=list(fields)), + node("out", "output", fields=[{"name": "r", "value": "完成 {{ input.stat_date }}"}])], + "edges": [{"source": "start", "target": "out"}]} + + +DATE_FIELD = {"name": "stat_date", "required": True} + + +async def workflow(*, published: bool = True, fields=(DATE_FIELD,)) -> tuple[str, str]: + name = f"门店周报-{uuid.uuid4().hex[:6]}" + async with SessionLocal() as session: + wf = Workflow(name=name, graph=graph(*fields), status="published" if published else "draft", + published_version=1 if published else None) + session.add(wf) + await session.flush() + session.add(WorkflowVersion(workflow_id=wf.id, version=1, graph=graph(*fields))) + await session.commit() + return wf.id, name + + +def payload(wf_id: str, **over) -> dict: + base = {"name": "周一早报", "workflow_id": wf_id, "cron": "0 9 * * 1", "timezone": "Asia/Shanghai", + "run_class": "formal", "input_template": {"stat_date": "{{ schedule.yesterday }}"}, "enabled": True, + "overlap": "skip", "misfire_grace_s": 3600, + "webhook": {"format": "feishu", "url": HOOK, "secret": SECRET, "events": ["finished", "failed"]}} + base.update(over) + return base + + +async def create(client, wf_id: str, **over) -> dict: + res = await client.post("/api/schedules", json=payload(wf_id, **over), + headers={"X-Actor": quote("李小明")}) + assert res.status_code == 201, res.text + return res.json() + + +async def settle(run_id: str) -> None: + for _ in range(400): + async with SessionLocal() as session: + row = await session.get(Run, run_id) + if row.status in ("succeeded", "failed", "cancelled") and row.manifest_seq is not None: + await run_manager.wait_idle(run_id) + return + await asyncio.sleep(0.03) + raise AssertionError(f"运行 {run_id} 停在 {row.status}") + + +# -------------------------------------------------------------------------- +# 增删改查 +# -------------------------------------------------------------------------- + + +async def test_create_returns_the_agreed_shape_and_hides_the_address(client): + wf, wf_name = await workflow() + res = await client.post("/api/schedules", json=payload(wf), headers={"X-Actor": quote("李小明")}) + assert res.status_code == 201 + body = res.json() + assert set(body) == SCHEDULE_KEYS + assert body["workflow_name"] == wf_name and body["created_by"] == "李小明" and body["version"] == 1 + assert body["webhook"] == {"format": "feishu", "url_hint": "open.feishu.cn/…90ab", "has_secret": True, + "events": ["finished", "failed"]} + assert body["description_text"] == "每周一 09:00（Asia/Shanghai）" + assert body["last_fire"] is None and body["overlap"] == "skip" and body["misfire_grace_s"] == 3600 + next_at = datetime.fromisoformat(body["next_run_at"]) + assert next_at.utcoffset().total_seconds() == 0 and next_at.astimezone(UTC).hour == 1 + assert next_at > datetime.now(UTC) + # 地址、密钥一个字都不回到浏览器 + assert "7c6b5a49" not in res.text and SECRET not in res.text and "/hook/" not in res.text + async with SessionLocal() as session: + row = await session.get(Schedule, body["id"]) + assert row.webhook_url.startswith("enc::") and decrypt(row.webhook_url) == HOOK + assert row.webhook_secret.startswith("enc::") and decrypt(row.webhook_secret) == SECRET + + +async def test_list_and_get(client): + wf, _ = await workflow() + made = await create(client, wf) + listed = (await client.get("/api/schedules")).json() + assert set(listed) == {"items"} and [s["id"] for s in listed["items"]] == [made["id"]] + got = await client.get(f"/api/schedules/{made['id']}") + assert got.status_code == 200 and got.json() == made + + +async def test_patch_updates_fields_and_bumps_the_version(client): + wf, _ = await workflow() + made = await create(client, wf) + res = await client.patch(f"/api/schedules/{made['id']}", + json={"name": "每日早报", "cron": "30 8 * * *", "version": 1}) + assert res.status_code == 200, res.text + body = res.json() + assert (body["name"], body["cron"], body["version"]) == ("每日早报", "30 8 * * *", 2) + assert body["description_text"] == "每天 08:30（Asia/Shanghai）" + assert datetime.fromisoformat(body["next_run_at"]).astimezone(UTC).minute == 30 + # 停用：没有下一次 + paused = (await client.patch(f"/api/schedules/{made['id']}", json={"enabled": False, "version": 2})).json() + assert paused["enabled"] is False and paused["next_run_at"] is None and paused["version"] == 3 + resumed = (await client.patch(f"/api/schedules/{made['id']}", json={"enabled": True, "version": 3})).json() + assert resumed["next_run_at"] is not None + + +async def test_stale_version_is_a_conflict(client): + wf, _ = await workflow() + made = await create(client, wf) + assert (await client.patch(f"/api/schedules/{made['id']}", json={"name": "甲", "version": 1})).status_code == 200 + res = await client.patch(f"/api/schedules/{made['id']}", json={"name": "乙", "version": 1}) + assert res.status_code == 409 and "刷新" in res.json()["detail"] + missing = await client.patch(f"/api/schedules/{made['id']}", json={"name": "丙"}) + assert missing.status_code == 422 and missing.json()["field"] == "version" + assert (await client.get(f"/api/schedules/{made['id']}")).json()["name"] == "甲" + + +async def test_patch_webhook_keeps_or_removes_the_address(client): + wf, _ = await workflow() + made = await create(client, wf) + sid = made["id"] + # 不带 url：地址不变，只改事件和格式 + res = await client.patch(f"/api/schedules/{sid}", json={ + "webhook": {"format": "dingtalk", "events": ["awaiting_approval"]}, "version": 1}) + assert res.status_code == 200, res.text + assert res.json()["webhook"] == {"format": "dingtalk", "url_hint": "open.feishu.cn/…90ab", "has_secret": True, + "events": ["awaiting_approval"]} + async with SessionLocal() as session: + assert decrypt((await session.get(Schedule, sid)).webhook_url) == HOOK + # 密钥写空串：去掉签名 + res = await client.patch(f"/api/schedules/{sid}", json={"webhook": {"secret": ""}, "version": 2}) + assert res.json()["webhook"]["has_secret"] is False + # url 写空串：删除整个 Webhook + res = await client.patch(f"/api/schedules/{sid}", json={"webhook": {"url": ""}, "version": 3}) + assert res.json()["webhook"] is None + async with SessionLocal() as session: + row = await session.get(Schedule, sid) + assert (row.webhook_url, row.webhook_secret, row.webhook_format) == (None, None, None) + # 没有地址时只给格式不行 + res = await client.patch(f"/api/schedules/{sid}", json={"webhook": {"format": "wecom"}, "version": 4}) + assert res.status_code == 422 and res.json()["field"] == "webhook" + # 重新配上 + res = await client.patch(f"/api/schedules/{sid}", json={ + "webhook": {"format": "wecom", "url": "https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=abc-0000-9f9f"}, + "version": 4}) + assert res.json()["webhook"] == {"format": "wecom", "url_hint": "qyapi.weixin.qq.com/…9f9f", "has_secret": False, + "events": ["finished", "failed", "awaiting_approval"]} + + +async def test_webhook_null_removes_it(client): + wf, _ = await workflow() + made = await create(client, wf) + res = await client.patch(f"/api/schedules/{made['id']}", json={"webhook": None, "version": 1}) + assert res.json()["webhook"] is None + + +async def test_delete(client): + wf, _ = await workflow() + made = await create(client, wf) + sid = made["id"] + ran = (await client.post(f"/api/schedules/{sid}/run")).json() + await settle(ran["run_id"]) + res = await client.delete(f"/api/schedules/{sid}") + assert res.status_code == 204 + gone = await client.get(f"/api/schedules/{sid}") + assert gone.status_code == 404 and gone.json()["detail"] == "定时任务不存在，可能已被删除" + async with SessionLocal() as session: + assert await session.scalar(select(func.count()).select_from(ScheduleFire) + .where(ScheduleFire.schedule_id == sid)) == 0 + # 运行记录留着，来源 id 也留着：它是追溯凭证 + assert (await session.get(Run, ran["run_id"])).schedule_id == sid + + +@pytest.mark.parametrize("method, path", [ + ("get", ""), ("patch", ""), ("delete", ""), ("post", "/run"), ("get", "/fires"), ("post", "/webhook/test")]) +async def test_unknown_schedule_is_404_in_chinese(client, method, path): + kwargs = {"json": {"version": 1}} if method == "patch" else {} + res = await getattr(client, method)(f"/api/schedules/nope{uuid.uuid4().hex[:6]}{path}", **kwargs) + assert res.status_code == 404 and res.json()["detail"] == "定时任务不存在，可能已被删除" + + +# -------------------------------------------------------------------------- +# 校验：422 {detail, field} +# -------------------------------------------------------------------------- + + +@pytest.mark.parametrize("over, field, fragment", [ + ({"name": " "}, "name", "名称"), + ({"cron": "0 25 * * *"}, "cron", "「时」"), + ({"cron": "0 0 30 2 *"}, "cron", "永远不会触发"), + ({"timezone": "Mars/Olympus"}, "timezone", "时区"), + ({"run_class": "weekly"}, "run_class", "正式运行"), + ({"input_template": {"stat_date": "{{ schedule.yesterdy }}"}}, "input_template", "schedule.yesterdy"), + ({"input_template": {"stat_dat": "{{ schedule.yesterday }}"}}, "input_template", "stat_dat"), + ({"input_template": {}}, "input_template", "必填"), + ({"misfire_grace_s": -1}, "misfire_grace_s", "宽限"), + ({"misfire_grace_s": True}, "misfire_grace_s", "宽限"), + ({"overlap": "queue"}, "overlap", "skip"), + ({"enabled": "yes"}, "enabled", "启用"), + ({"webhook": {"format": "slack", "url": HOOK}}, "webhook", "格式"), + ({"webhook": {"format": "feishu", "url": "open.feishu.cn/hook"}}, "webhook", "Webhook 地址"), + ({"webhook": {"format": "feishu", "url": HOOK, "events": ["done"]}}, "webhook", "通知事件"), + ({"workflow_id": ""}, "workflow_id", "工作流"), +]) +async def test_invalid_create_is_422_with_field(client, over, field, fragment): + wf, _ = await workflow() + res = await client.post("/api/schedules", json=payload(wf, **over)) + assert res.status_code == 422, res.text + body = res.json() + assert body["field"] == field and fragment in body["detail"], body + + +async def test_missing_workflow_and_unpublished_formal(client): + res = await client.post("/api/schedules", json=payload(f"gone{uuid.uuid4().hex[:6]}")) + assert res.status_code == 422 and res.json()["field"] == "workflow_id" and "不存在" in res.json()["detail"] + wf, name = await workflow(published=False) + res = await client.post("/api/schedules", json=payload(wf)) + assert res.status_code == 422 and res.json()["field"] == "run_class" + assert f"「{name}」还没有发布版本" in res.json()["detail"] + # 探索运行不要求发布版本 + res = await client.post("/api/schedules", json=payload(wf, run_class="exploratory")) + assert res.status_code == 201 + + +async def test_non_json_body_is_422_in_chinese(client): + res = await client.post("/api/schedules", content=b"not json", headers={"Content-Type": "application/json"}) + assert res.status_code == 422 and "JSON" in res.json()["detail"] + res = await client.post("/api/schedules", json=["列表"]) + assert res.status_code == 422 and "JSON" in res.json()["detail"] + + +async def test_webhook_with_empty_url_on_create_means_none(client): + wf, _ = await workflow() + made = await create(client, wf, webhook={"format": "feishu", "url": "", "events": ["failed"]}) + assert made["webhook"] is None + + +# -------------------------------------------------------------------------- +# 预览、变量 +# -------------------------------------------------------------------------- + + +async def test_preview(client): + res = await client.post("/api/schedules/preview", json={ + "cron": "0 9 * * *", "timezone": "Asia/Shanghai", "n": 3, + "input_template": {"stat_date": "{{ schedule.yesterday }}", "since": "{{ schedule.last_fire_time }}"}}) + assert res.status_code == 200, res.text + body = res.json() + assert set(body) == {"description_text", "next", "inputs"} + assert body["description_text"] == "每天 09:00（Asia/Shanghai）" + moments = [datetime.fromisoformat(t) for t in body["next"]] + assert len(moments) == 3 and all(m.utcoffset().total_seconds() == 0 for m in moments) + assert all((b - a).total_seconds() == 86400 for a, b in zip(moments, moments[1:])) + first_local = moments[0].astimezone(ZoneInfo("Asia/Shanghai")) + assert body["inputs"]["stat_date"] == (first_local.date() - timedelta(days=1)).isoformat() + assert body["inputs"]["since"].endswith("09:00:00+08:00") + + +async def test_preview_defaults_to_five(client): + body = (await client.post("/api/schedules/preview", json={"cron": "*/10 * * * *", "timezone": "UTC"})).json() + assert len(body["next"]) == 5 and body["inputs"] == {} + + +@pytest.mark.parametrize("given, field", [ + ({"cron": "bad", "timezone": "UTC"}, "cron"), + ({"cron": "0 9 * * *", "timezone": "Nowhere/City"}, "timezone"), + ({"cron": "0 9 * * *", "timezone": "UTC", "input_template": {"d": "{{ schedule.today"}}, "input_template"), +]) +async def test_preview_errors(client, given, field): + res = await client.post("/api/schedules/preview", json=given) + assert res.status_code == 422 and res.json()["field"] == field + assert any("一" <= ch <= "鿿" for ch in res.json()["detail"]) + + +async def test_preview_checks_field_names_against_the_workflow_but_not_missing_ones(client): + wf, _ = await workflow(fields=(DATE_FIELD, {"name": "region"})) + ok = await client.post("/api/schedules/preview", json={ + "cron": "0 9 * * *", "timezone": "UTC", "workflow_id": wf, "input_template": {"region": "华东"}}) + assert ok.status_code == 200 + bad = await client.post("/api/schedules/preview", json={ + "cron": "0 9 * * *", "timezone": "UTC", "workflow_id": wf, "input_template": {"regoin": "华东"}}) + assert bad.status_code == 422 and bad.json()["field"] == "input_template" and "regoin" in bad.json()["detail"] + + +async def test_variables(client): + res = await client.get("/api/schedules/variables") + assert res.status_code == 200 + items = res.json() + assert len(items) == 10 and all(set(i) == {"name", "label", "example"} for i in items) + assert {"name": "schedule.yesterday", "label": "昨天（YYYY-MM-DD）"}.items() <= next( + i for i in items if i["name"] == "schedule.yesterday").items() + assert (await client.get("/api/schedules/variables", params={"timezone": "Nowhere/City"})).status_code == 422 + + +# -------------------------------------------------------------------------- +# 立即运行、触发历史 +# -------------------------------------------------------------------------- + + +async def test_run_now_and_overlap(client, monkeypatch): + wf, _ = await workflow() + made = await create(client, wf, webhook=None) + res = await client.post(f"/api/schedules/{made['id']}/run", headers={"X-Actor": quote("王芳")}) + assert res.status_code == 200, res.text + body = res.json() + assert set(body) == {"fire", "run_id"} and set(body["fire"]) == FIRE_KEYS + fire = body["fire"] + assert (fire["kind"], fire["status"], fire["run_id"]) == ("manual", "started", body["run_id"]) + assert fire["run_status"] in ("queued", "running", "succeeded") + async with SessionLocal() as session: + run = await session.get(Run, body["run_id"]) + assert run.schedule_id == made["id"] and run.started_by == "王芳" and run.run_class == "formal" + assert set(run.input) == {"stat_date"} + # 上一次还没结束（这里把它改成停在审批）：同一套重叠规则 + await settle(body["run_id"]) + async with SessionLocal() as session: + await session.execute(Run.__table__.update().where(Run.id == body["run_id"]).values(status="interrupted")) + await session.commit() + # 换到下一秒再点（同一秒内连点交回同一条，见下一个用例） + monkeypatch.setattr(schedules_api, "utcnow", lambda: datetime.now(UTC) + timedelta(seconds=2)) + second = (await client.post(f"/api/schedules/{made['id']}/run")).json() + assert (second["fire"]["status"], second["run_id"]) == ("skipped_overlap", None) + assert body["run_id"][:8] in second["fire"]["reason"] + # 运行列表能按定时任务筛 + listed = (await client.get("/api/runs", params={"schedule_id": made["id"]})).json() + assert [r["id"] for r in listed] == [body["run_id"]] + + +async def test_double_click_in_the_same_second_returns_the_same_fire(client, monkeypatch): + wf, _ = await workflow() + made = await create(client, wf, webhook=None) + moment = datetime.now(UTC).replace(microsecond=0) + monkeypatch.setattr(schedules_api, "utcnow", lambda: moment) + first, second = [(await client.post(f"/api/schedules/{made['id']}/run")).json() for _ in range(2)] + assert first["fire"]["id"] == second["fire"]["id"] and first["run_id"] == second["run_id"] + async with SessionLocal() as session: + assert await session.scalar(select(func.count()).select_from(Run).where(Run.schedule_id == made["id"])) == 1 + await settle(first["run_id"]) + + +async def test_run_now_refused_for_unpublished_formal(client): + wf, name = await workflow() + made = await create(client, wf, webhook=None) + async with SessionLocal() as session: + await session.execute(Workflow.__table__.update().where(Workflow.id == wf).values(published_version=None)) + await session.commit() + body = (await client.post(f"/api/schedules/{made['id']}/run")).json() + assert (body["fire"]["status"], body["run_id"]) == ("failed_to_start", None) + assert body["fire"]["reason"] == f"「{name}」还没有发布版本，请先在编排页发布" + + +async def test_fire_history_pages_and_last_fire(client): + wf, _ = await workflow() + made = await create(client, wf, webhook=None) + sid = made["id"] + async with SessionLocal() as session: + for i in range(5): + session.add(ScheduleFire(schedule_id=sid, fire_at=datetime(2026, 9, 1 + i, 1, 0, tzinfo=UTC), + kind="scheduled", status="skipped_overlap", reason=f"第 {i} 次", + created_at=datetime(2026, 9, 1 + i, 1, 0, 1, tzinfo=UTC))) + await session.commit() + newest = (await session.execute(select(ScheduleFire).where(ScheduleFire.schedule_id == sid) + .order_by(ScheduleFire.created_at.desc()))).scalars().first() + session.add(WebhookDelivery(schedule_id=sid, fire_id=newest.id, event="failed", status="ok", attempts=1, + http_status=200, created_at=datetime(2026, 9, 5, 1, 0, 5, tzinfo=UTC))) + await session.commit() + page1 = (await client.get(f"/api/schedules/{sid}/fires", params={"limit": 2})).json() + assert [f["reason"] for f in page1["items"]] == ["第 4 次", "第 3 次"] and page1["next_before"] + assert page1["items"][0]["notified"] == [{"event": "failed", "status": "ok", "at": "2026-09-05T01:00:05+00:00"}] + assert page1["items"][0]["fire_at"] == "2026-09-05T01:00:00+00:00" + page2 = (await client.get(f"/api/schedules/{sid}/fires", + params={"limit": 2, "before": page1["next_before"]})).json() + assert [f["reason"] for f in page2["items"]] == ["第 2 次", "第 1 次"] and page2["next_before"] + page3 = (await client.get(f"/api/schedules/{sid}/fires", + params={"limit": 2, "before": page2["next_before"]})).json() + assert [f["reason"] for f in page3["items"]] == ["第 0 次"] and page3["next_before"] is None + last = (await client.get(f"/api/schedules/{sid}")).json()["last_fire"] + assert last["reason"] == "第 4 次" and set(last) == FIRE_KEYS + bad = await client.get(f"/api/schedules/{sid}/fires", params={"limit": "many"}) + assert bad.status_code == 422 and bad.json()["field"] == "limit" + bad = await client.get(f"/api/schedules/{sid}/fires", params={"before": "昨天"}) + assert bad.status_code == 422 and bad.json()["field"] == "before" + + +# -------------------------------------------------------------------------- +# 测试 Webhook +# -------------------------------------------------------------------------- + + +async def test_webhook_test_endpoint(client, monkeypatch): + seen: list[httpx.Request] = [] + + def handler(request: httpx.Request) -> httpx.Response: + seen.append(request) + return httpx.Response(200, json={"code": 0, "msg": "success"}) + + monkeypatch.setattr(webhook, "_transport", httpx.MockTransport(handler)) + wf, _ = await workflow() + made = await create(client, wf) + res = await client.post(f"/api/schedules/{made['id']}/webhook/test") + assert res.status_code == 200 + assert res.json() == {"ok": True, "http_status": 200, "message": "测试消息已送达（HTTP 200）"} + assert str(seen[0].url) == HOOK and "sign" in json.loads(seen[0].content) + bare = await create(client, wf, webhook=None) + res = await client.post(f"/api/schedules/{bare['id']}/webhook/test") + assert res.status_code == 422 and res.json()["field"] == "webhook" and "Webhook" in res.json()["detail"]