aigc-home/calliope-backend/tests/test_workflow_test.py

"""「测试工作流」（HEL-643 批次 2）端到端：真实队列 → fake ComfyUI → 产物隔离。

测试任务与普通任务区分（jobs.origin）、dry-run 拒绝、幂等复用、
保留策略与级联清理，全部对着 fake ComfyUI 跑真实链路。
"""

from __future__ import annotations

import time
from pathlib import Path

from calliope.config import settings
from calliope.queue.workflow_test import workflow_test_dir
from tests.test_studio_generation import (
    IMAGE_WORKFLOW,
    PNG_1PX,
    _real_run,
)


def _wait_terminal(client, job_id: int, timeout: float = 25.0) -> dict:
    deadline = time.time() + timeout
    while time.time() < deadline:
        r = client.get(f"/api/jobs/{job_id}")
        assert r.status_code == 200, r.text
        job = r.json()
        if job["status"] not in ("pending", "running"):
            return job
        time.sleep(0.4)
    raise AssertionError(f"test job {job_id} did not finish in {timeout}s")


def _create_workflow(client, workflow: dict, name: str = "可测工作流") -> int:
    r = client.post(
        "/api/workflows",
        json={"name": name, "workflow_json": workflow, "usage": "text2image"},
    )
    assert r.status_code == 200, r.text
    return r.json()["id"]


def test_passing_test_downloads_real_output_to_scratch_dir(client, fake_comfy):
    with _real_run(client, fake_comfy):
        wf_id = _create_workflow(client, IMAGE_WORKFLOW, name="通过用例")
        r = client.post(
            f"/api/workflows/{wf_id}/test", json={"input_values": {"1": "测试提示词"}}
        )
        assert r.status_code == 200, r.text
        body = r.json()
        assert body["reused"] is False
        job = _wait_terminal(client, body["job"]["id"])
        assert job["status"] == "done", job.get("error")
        assert job["origin"] == "workflow_test"
        assert job["payload"]["source"] == "workflow_test"
        assert job["phase"] == "finished"
        # 真实拿到可下载输出才算通过：产物落在独立 scratch 目录
        assert len(job["output_paths"]) >= 1
        out = Path(job["output_paths"][0])
        assert out.is_file() and out.read_bytes() == PNG_1PX
        assert workflow_test_dir(wf_id) in out.parents
        # 提交给 fake ComfyUI 的是补丁后的真实提示词
        queued = fake_comfy.prompts[-1]["prompt"]
        assert queued["1"]["inputs"]["text"] == "测试提示词"


def test_tasks_list_shows_origin_but_playground_history_hides_it(client, fake_comfy):
    with _real_run(client, fake_comfy):
        wf_id = _create_workflow(client, IMAGE_WORKFLOW, name="可见性用例")
        r = client.post(f"/api/workflows/{wf_id}/test", json={"input_values": {"1": "p"}})
        job_id = r.json()["job"]["id"]
        _wait_terminal(client, job_id)

        listed = client.get("/api/jobs?limit=50").json()
        mine = [j for j in listed if j["id"] == job_id]
        assert mine and mine[0]["origin"] == "workflow_test"

        pg = client.get("/api/playground/jobs").json()
        assert all(j["id"] != job_id for j in pg), "test jobs must stay out of studio history"


def test_failed_prompt_reports_chinese_reason(client, fake_comfy):
    with _real_run(client, fake_comfy):
        wf_id = _create_workflow(client, IMAGE_WORKFLOW, name="排队被拒用例")
        fake_comfy.fail_prompt = True
        r = client.post(f"/api/workflows/{wf_id}/test", json={"input_values": {"1": "p"}})
        job = _wait_terminal(client, r.json()["job"]["id"])
        assert job["status"] == "failed"
        assert "拒绝执行" in (job["error"] or "")


def test_no_output_does_not_pass(client, fake_comfy):
    with _real_run(client, fake_comfy):
        wf_id = _create_workflow(client, IMAGE_WORKFLOW, name="无输出用例")
        fake_comfy.drop_outputs = True
        r = client.post(f"/api/workflows/{wf_id}/test", json={"input_values": {"1": "p"}})
        job = _wait_terminal(client, r.json()["job"]["id"])
        assert job["status"] == "failed"
        assert "没有返回任何图片或视频" in (job["error"] or "")


def test_dry_run_setting_refuses_real_test(client, fake_comfy):
    with _real_run(client, fake_comfy):
        settings.dry_run = True  # conftest default; _real_run clears it — set again
        try:
            wf_id = _create_workflow(client, IMAGE_WORKFLOW, name="dry 拒绝用例")
            r = client.post(f"/api/workflows/{wf_id}/test", json={"input_values": {"1": "p"}})
            assert r.status_code == 400
            assert "Dry-run" in r.json()["detail"]
        finally:
            settings.dry_run = False


def test_repeat_click_reuses_inflight_job(client, fake_comfy):
    with _real_run(client, fake_comfy):
        wf_id = _create_workflow(client, IMAGE_WORKFLOW, name="幂等用例")
        assert client.post("/api/jobs/pause").status_code == 200
        try:
            r1 = client.post(
                f"/api/workflows/{wf_id}/test", json={"input_values": {"1": "一次就好"}}
            )
            assert r1.status_code == 200 and r1.json()["reused"] is False
            r2 = client.post(
                f"/api/workflows/{wf_id}/test", json={"input_values": {"1": "重复点击"}}
            )
            assert r2.status_code == 200
            assert r2.json()["reused"] is True
            assert r2.json()["job"]["id"] == r1.json()["job"]["id"]
        finally:
            client.post("/api/jobs/resume")
        client.post(f"/api/jobs/{r1.json()['job']['id']}/cancel")
        _wait_terminal(client, r1.json()["job"]["id"])


def test_retention_keeps_newest_records_and_prunes_artifacts(client, fake_comfy):
    with _real_run(client, fake_comfy):
        wf_id = _create_workflow(client, IMAGE_WORKFLOW, name="保留策略用例")
        for i in range(7):
            fake_comfy.out_name = f"t{i}.png"
            r = client.post(
                f"/api/workflows/{wf_id}/test",
                json={"input_values": {"1": f"第{i}次"}},
            )
            _wait_terminal(client, r.json()["job"]["id"])
        rows = client.get(f"/api/workflows/{wf_id}/tests").json()
        assert len(rows) <= 5
        kept_ids = {row["id"] for row in rows}
        # 被清掉的旧记录其产物文件也一并删除
        leftovers = [p for p in workflow_test_dir(wf_id).iterdir() if p.suffix == ".png"]
        assert all(p.stem.lstrip("t").isdigit() and int(p.stem[1:]) >= 2 for p in leftovers)
        assert len(kept_ids) == 5


def test_clear_tests_and_workflow_delete_cascade(client, fake_comfy):
    with _real_run(client, fake_comfy):
        wf_id = _create_workflow(client, IMAGE_WORKFLOW, name="清理用例")
        r = client.post(f"/api/workflows/{wf_id}/test", json={"input_values": {"1": "p"}})
        _wait_terminal(client, r.json()["job"]["id"])
        assert workflow_test_dir(wf_id).is_dir()

        r = client.post(f"/api/workflows/{wf_id}/clear-tests")
        assert r.status_code == 200 and r.json()["removed"] >= 1
        assert not workflow_test_dir(wf_id).exists()
        assert client.get(f"/api/workflows/{wf_id}/tests").json() == []

        # 进行中的测试：先拒绝清除，取消后删除工作流级联清理
        assert client.post("/api/jobs/pause").status_code == 200
        try:
            job_id = client.post(
                f"/api/workflows/{wf_id}/test", json={"input_values": {"1": "p"}}
            ).json()["job"]["id"]
            r = client.post(f"/api/workflows/{wf_id}/clear-tests")
            assert r.status_code == 409
        finally:
            client.post("/api/jobs/resume")
        client.post(f"/api/jobs/{job_id}/cancel")
        _wait_terminal(client, job_id)

        # 工作流列表带出「上次测试」实时状态
        listed = client.get("/api/workflows").json()
        mine = [w for w in listed if w["id"] == wf_id][0]
        assert mine["last_test"]["id"] == job_id

        assert client.delete(f"/api/workflows/{wf_id}").status_code == 200
        assert not workflow_test_dir(wf_id).exists()
        assert client.get(f"/api/workflows/{wf_id}/tests").status_code == 404


def test_last_test_summary_live_on_list(client, fake_comfy):
    with _real_run(client, fake_comfy):
        wf_a = _create_workflow(client, IMAGE_WORKFLOW, name="A")
        wf_b = _create_workflow(client, IMAGE_WORKFLOW, name="B")
        r = client.post(f"/api/workflows/{wf_a}/test", json={"input_values": {"1": "p"}})
        done = _wait_terminal(client, r.json()["job"]["id"])
        listed = {w["id"]: w for w in client.get("/api/workflows").json()}
        assert listed[wf_a]["last_test"]["status"] == "done"
        assert listed[wf_a]["last_test"]["output_count"] == len(done["output_paths"])
        assert listed[wf_b]["last_test"] is None