From e410d2a6a7841b3b3743e2d6f50ea8b188b4c673 Mon Sep 17 00:00:00 2001 From: ThearchyHelios Date: Tue, 6 Oct 2026 23:56:31 +0800 Subject: [PATCH] =?UTF-8?q?M4=EF=BC=9A=E5=8D=95=E4=B8=AA=E5=B7=A5=E4=BD=9C?= =?UTF-8?q?=E6=B5=81=E6=8E=A5=E5=8F=A3=E7=9A=84=20run=5Fcount=20=E5=92=8C?= =?UTF-8?q?=E5=88=97=E8=A1=A8=E6=8E=A5=E5=8F=A3=E6=8C=89=E5=90=8C=E4=B8=80?= =?UTF-8?q?=E4=B8=AA=E5=87=BD=E6=95=B0=E7=AE=97?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 验收时 GET /api/workflows/{id} 的 run_count 恒为 0，同一时刻列表接口里同一个工作流是 2 次、4 次：只有列表接口按 Run.workflow_id 计数，详情、保存（PATCH）、恢复旧版、新建、复制直接返回数据库对象，落在 WorkflowOut 的默认值 0 上。 计数抽成 _run_counts，列表和返回单个工作流的接口（_workflow_out）都走它，次数一致；新建、复制出来的工作流照实是 0。 --- backend/app/api/workflows.py | 40 +++++++---- backend/tests/test_workflow_run_count.py | 84 ++++++++++++++++++++++++ 2 files changed, 111 insertions(+), 13 deletions(-) create mode 100644 backend/tests/test_workflow_run_count.py diff --git a/backend/app/api/workflows.py b/backend/app/api/workflows.py index 69703cf3..fd25b882 100644 --- a/backend/app/api/workflows.py +++ b/backend/app/api/workflows.py @@ -65,14 +65,28 @@ class WorkflowOut(BaseModel): model_config = {"from_attributes": True} +async def _run_counts(session: AsyncSession, ids: list[str] | None = None) -> dict[str, int]: + """各工作流的运行次数 {工作流 id: 次数}，ids 为 None 时算全部。列表和单个工作流的接口都按它算：以前只有列表 + 计数，详情、保存、恢复的返回落在 WorkflowOut 的默认值 0 上，同一个工作流两处说的次数不一样。""" + stmt = select(Run.workflow_id, func.count(Run.id)).group_by(Run.workflow_id) + if ids is not None: + stmt = stmt.where(Run.workflow_id.in_(ids)) + return dict((await session.execute(stmt)).all()) + + +async def _workflow_out(session: AsyncSession, workflow: Workflow) -> WorkflowOut: + """返回单个工作流的接口都走这里，run_count 和列表一致。""" + item = WorkflowOut.model_validate(workflow) + item.run_count = (await _run_counts(session, [workflow.id])).get(workflow.id, 0) + return item + + @router.get("", response_model=list[WorkflowOut]) async def list_workflows(session: AsyncSession = Depends(get_session)) -> list[WorkflowOut]: rows = list( (await session.execute(select(Workflow).order_by(Workflow.updated_at.desc()))).scalars() ) - counts = dict( - (await session.execute(select(Run.workflow_id, func.count(Run.id)).group_by(Run.workflow_id))).all() - ) + counts = await _run_counts(session) out: list[WorkflowOut] = [] for row in rows: item = WorkflowOut.model_validate(row) @@ -84,7 +98,7 @@ async def list_workflows(session: AsyncSession = Depends(get_session)) -> list[W @router.post("", response_model=WorkflowOut, status_code=201) async def create_workflow( payload: WorkflowIn, session: AsyncSession = Depends(get_session) -) -> Workflow: +) -> WorkflowOut: workflow = Workflow( name=payload.name, description=payload.description, @@ -99,23 +113,23 @@ async def create_workflow( ) await session.commit() await session.refresh(workflow) - return workflow + return await _workflow_out(session, workflow) @router.get("/{workflow_id}", response_model=WorkflowOut) async def get_workflow( workflow_id: str, session: AsyncSession = Depends(get_session) -) -> Workflow: +) -> WorkflowOut: workflow = await session.get(Workflow, workflow_id) if not workflow: raise HTTPException(404, _GONE) - return workflow + return await _workflow_out(session, workflow) @router.patch("/{workflow_id}", response_model=WorkflowOut) async def update_workflow( workflow_id: str, payload: WorkflowPatch, session: AsyncSession = Depends(get_session) -) -> Workflow: +) -> WorkflowOut: workflow = await session.get(Workflow, workflow_id) if not workflow: raise HTTPException(404, _GONE) @@ -146,7 +160,7 @@ async def update_workflow( workflow.status = "draft" await session.commit() await session.refresh(workflow) - return workflow + return await _workflow_out(session, workflow) @router.delete("/{workflow_id}", status_code=204) @@ -172,7 +186,7 @@ async def delete_workflow( @router.post("/{workflow_id}/duplicate", response_model=WorkflowOut, status_code=201) async def duplicate_workflow( workflow_id: str, session: AsyncSession = Depends(get_session) -) -> Workflow: +) -> WorkflowOut: source = await session.get(Workflow, workflow_id) if not source: raise HTTPException(404, _GONE) @@ -185,7 +199,7 @@ async def duplicate_workflow( session.add(copy) await session.commit() await session.refresh(copy) - return copy + return await _workflow_out(session, copy) class VersionOut(BaseModel): @@ -267,7 +281,7 @@ async def get_version( @router.post("/{workflow_id}/versions/{version}/restore", response_model=WorkflowOut) async def restore_version( workflow_id: str, version: int, session: AsyncSession = Depends(get_session) -) -> Workflow: +) -> WorkflowOut: workflow = await session.get(Workflow, workflow_id) if not workflow: raise HTTPException(404, _GONE) @@ -298,7 +312,7 @@ async def restore_version( workflow.status = "draft" await session.commit() await session.refresh(workflow) - return workflow + return await _workflow_out(session, workflow) class PublishIn(BaseModel): diff --git a/backend/tests/test_workflow_run_count.py b/backend/tests/test_workflow_run_count.py new file mode 100644 index 00000000..e6ad7071 --- /dev/null +++ b/backend/tests/test_workflow_run_count.py @@ -0,0 +1,84 @@ +"""单个工作流接口的 run_count 和列表接口按同一个函数算。 + +2026-10 验收：GET /api/workflows/{id} 的 run_count 恒为 0，同一时刻列表接口（GET /api/workflows）里这个工作流是 +2 次、运行列表按 workflow_id 筛出来也是 2 条——只有列表接口按 Run.workflow_id 计数，详情、保存（PATCH）、恢复旧版 +直接返回数据库对象，WorkflowOut.run_count 落在默认值 0 上。现在凡是返回工作流的接口都按同一个函数算，次数和列表一致。 + +工作流名、图、运行都是虚构的。 +""" +from __future__ import annotations + +import pytest +from httpx import ASGITransport, AsyncClient + +from app.db.base import SessionLocal +from app.db.models import Run +from app.main import app + +GRAPH = { + "nodes": [ + {"id": "start", "type": "input", "position": {"x": 0, "y": 0}, + "data": {"label": "输入", "config": {"fields": [{"name": "week"}]}}}, + {"id": "out", "type": "output", "position": {"x": 300, "y": 0}, + "data": {"label": "成果", "config": {"fields": [{"name": "r", "value": "{{ input.week }}"}]}}}, + ], + "edges": [{"source": "start", "target": "out"}], +} + + +@pytest.fixture +async def client(): + async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c: + yield c + + +async def _add_runs(workflow_id: str | None, n: int, status: str = "succeeded") -> None: + async with SessionLocal() as session: + for _ in range(n): + session.add(Run(workflow_id=workflow_id, workflow_name="灯塔周报", graph=GRAPH, input={}, status=status)) + await session.commit() + + +async def _listed(client, workflow_id: str) -> int: + [row] = [w for w in (await client.get("/api/workflows")).json() if w["id"] == workflow_id] + return row["run_count"] + + +async def test_the_single_workflow_counts_its_runs_like_the_list(client): + made = (await client.post("/api/workflows", json={"name": "灯塔周报", "graph": GRAPH})).json() + other = (await client.post("/api/workflows", json={"name": "潮汐日报", "graph": GRAPH})).json() + await _add_runs(made["id"], 3) + await _add_runs(made["id"], 1, status="failed") + # 别的工作流、没挂在工作流上的运行不算进来 + await _add_runs(other["id"], 2) + await _add_runs(None, 1) + + assert await _listed(client, made["id"]) == 4 + got = await client.get(f"/api/workflows/{made['id']}") + assert got.status_code == 200 and got.json()["run_count"] == 4 + assert (await client.get(f"/api/workflows/{other['id']}")).json()["run_count"] == await _listed(client, other["id"]) == 2 + + +async def test_save_and_restore_answer_with_the_same_count(client): + """保存、恢复旧版的返回也是这个工作流：界面拿它替换手里的那一份，次数不能变回 0。""" + made = (await client.post("/api/workflows", json={"name": "灯塔周报", "graph": GRAPH})).json() + await _add_runs(made["id"], 2) + + renamed = await client.patch(f"/api/workflows/{made['id']}", json={"name": "灯塔周报（新）"}) + assert renamed.status_code == 200 and renamed.json()["run_count"] == 2 + changed = {**GRAPH, "nodes": [*GRAPH["nodes"][:1], {**GRAPH["nodes"][1], "position": {"x": 320, "y": 0}}]} + saved = await client.patch(f"/api/workflows/{made['id']}", json={"graph": changed, "note": "挪了一下"}) + assert saved.status_code == 200 and saved.json()["version"] == 2 and saved.json()["run_count"] == 2 + restored = await client.post(f"/api/workflows/{made['id']}/versions/1/restore") + assert restored.status_code == 200 and restored.json()["run_count"] == 2 + assert await _listed(client, made["id"]) == 2 + + +async def test_new_and_copied_workflows_have_no_runs_yet(client): + """新建、复制出来的是新工作流，还没有运行；复制不带上原来那个的次数。""" + made = (await client.post("/api/workflows", json={"name": "灯塔周报", "graph": GRAPH})).json() + assert made["run_count"] == 0 + await _add_runs(made["id"], 3) + copy = await client.post(f"/api/workflows/{made['id']}/duplicate") + assert copy.status_code == 201 and copy.json()["run_count"] == 0 + assert (await client.get(f"/api/workflows/{made['id']}")).json()["run_count"] == 3