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

"""HEL-636: the two-step import closes the loop — analyze → confirm → save,
user corrections stick (reanalyze never overwrites them), pending workflows
cannot generate, and legacy rows migrate losslessly."""

from __future__ import annotations

import json
from pathlib import Path

from calliope.config import settings
from calliope.db import get_db, migrate_db

FIXTURES = Path(__file__).parent / "fixtures" / "comfy_native"


def _load(name: str) -> dict:
    return json.loads((FIXTURES / name).read_text(encoding="utf-8"))


AMBIGUOUS = {
    "1": {"class_type": "PrimitiveStringMultiline", "inputs": {"value": "p"}},
    "5": {"class_type": "LoadImage", "inputs": {"image": "x.png"}},
    "7": {
        "class_type": "SomeVideoSampler",
        "inputs": {"prompt": ["1", 0], "image": ["5", 0]},
    },
    "9": {"class_type": "SaveVideo", "inputs": {"images": ["7", 0]}},
}


def test_analyze_returns_usage_evidence_and_problem(client):
    wf = _load("qwen_image_t2i_API.json")
    r = client.post("/api/workflows/analyze", json={"workflow_json": wf})
    assert r.status_code == 200
    body = r.json()
    assert body["usage"] == "text2image"
    assert body["usage_ambiguous"] is False and body["needs_mapping"] is False
    assert any(o["kind"] == "image" for o in body["usage_evidence"]["outputs"])
    assert body["usable_usage"] == [
        "text2image", "image2image", "text2video", "image2video", "ref2video",
    ]

    r = client.post("/api/workflows/analyze", json={"workflow_json": AMBIGUOUS})
    body = r.json()
    assert body["usage"] is None and body["usage_ambiguous"] is True
    assert "首帧还是参考图" in body["usage_problem"]
    assert body["usage_evidence"]["image_slots"][0]["nodeId"] == "5"

    bare = {"1": {"class_type": "Note", "inputs": {}}}
    r = client.post("/api/workflows/analyze", json={"workflow_json": bare})
    body = r.json()
    assert "未在图中找到输出节点" in body["usage_problem"]


def test_create_stores_auto_usage_and_kind(client):
    r = client.post(
        "/api/workflows",
        json={"name": "自动·文生图", "workflow_json": _load("qwen_image_t2i_API.json")},
    )
    assert r.status_code == 200, r.text
    wf = r.json()
    assert wf["usage"] == "text2image" and wf["usage_source"] == "auto"
    assert wf["kind"] == "image" and wf["is_enabled"] is True

    r = client.post(
        "/api/workflows",
        json={"name": "自动·参考生视频", "workflow_json": _load("h3_ref2va_05_API.json")},
    )
    wf = r.json()
    assert wf["usage"] == "ref2video" and wf["kind"] == "video"


def test_ambiguous_import_stays_pending_and_cannot_generate(client):
    r = client.post("/api/workflows", json={"name": "歧义", "workflow_json": AMBIGUOUS})
    assert r.status_code == 200
    wf = r.json()
    assert wf["usage"] is None and wf["is_enabled"] is False

    gen = client.post(
        "/api/playground/generate",
        json={"workflow_id": wf["id"], "input_values": {"1": "x"}},
    )
    assert gen.status_code == 400
    assert "已停用" in gen.json()["detail"]
    assert client.get("/api/playground/jobs").json() == []


def test_confirm_page_correction_roundtrip(client):
    """analyze → user picks usage + fixes the slot role → create carries both."""
    analyzed = client.post("/api/workflows/analyze", json={"workflow_json": AMBIGUOUS}).json()
    assert analyzed["usage"] is None

    slot = analyzed["usage_evidence"]["image_slots"][0]
    schema = [
        {**i, "role": "first_frame", "manual": True} if i["nodeId"] == slot["nodeId"] else i
        for i in analyzed["inputs"]
    ]
    r = client.post(
        "/api/workflows",
        json={
            "name": "确认·图生视频",
            "workflow_json": AMBIGUOUS,
            "usage": "image2video",
            "usage_source": "user",
            "input_schema": schema,
        },
    )
    assert r.status_code == 200, r.text
    wf = r.json()
    assert wf["usage"] == "image2video" and wf["usage_source"] == "user"
    assert wf["kind"] == "video" and wf["is_enabled"] is True
    # the manual role sticks on read (and through re-derivation)
    entry = next(i for i in wf["input_schema"] if i["nodeId"] == slot["nodeId"])
    assert entry["role"] == "first_frame"

    # reanalyze refreshes the auto schema but never the user's usage
    r = client.post(f"/api/workflows/{wf['id']}/reanalyze")
    assert r.status_code == 200
    fresh = r.json()
    assert fresh["usage"] == "image2video" and fresh["usage_source"] == "user"
    entry = next(i for i in fresh["input_schema"] if i["nodeId"] == slot["nodeId"])
    assert entry["role"] == "first_frame" and entry.get("manual") is True


def test_patch_usage_confirms_pending_workflow(client):
    r = client.post("/api/workflows", json={"name": "待确认", "workflow_json": AMBIGUOUS})
    wid = r.json()["id"]
    assert r.json()["is_enabled"] is False

    r = client.patch(f"/api/workflows/{wid}", json={"usage": "ref2video"})
    assert r.status_code == 200
    wf = r.json()
    assert wf["usage"] == "ref2video" and wf["usage_source"] == "user"
    assert wf["kind"] == "video" and wf["is_enabled"] is True

    listed = next(w for w in client.get("/api/workflows").json() if w["id"] == wid)
    assert listed["usage"] == "ref2video"


def test_reanalyze_upgrades_auto_usage_only(client):
    r = client.post(
        "/api/workflows",
        json={"name": "重分析", "workflow_json": _load("qwen_image_t2i_API.json")},
    )
    wid = r.json()["id"]
    r = client.patch(f"/api/workflows/{wid}", json={"usage": "image2image", "usage_source": "user"})
    assert r.json()["usage"] == "image2image"

    r = client.post(f"/api/workflows/{wid}/reanalyze")
    assert r.json()["usage"] == "image2image"  # user decision survives
    assert r.json()["usage_source"] == "user"


def test_legacy_kind_only_callers_still_work(client):
    """Old clients post kind without usage — determinate graphs classify, the
    row is enabled and generation proceeds as before."""
    r = client.post(
        "/api/workflows",
        json={"name": "旧调用", "kind": "image", "workflow_json": _load("qwen_image_t2i_API.json")},
    )
    assert r.status_code == 200
    wf = r.json()
    assert wf["usage"] == "text2image" and wf["is_enabled"] is True


def test_migration_backfills_usage_and_keeps_history(client):
    """A8: pre-HEL-636 rows (direct SQL, no usage) settle on re-migration;
    determinate ones gain usage, ambiguous ones are parked disabled, and
    existing jobs/history stay readable with their original kind."""
    pid = client.post("/api/projects", json={"title": "存量项目", "idea": "x"}).json()["id"]
    conn = get_db(settings.db_path)
    try:
        conn.execute(
            "INSERT INTO workflows (name, kind, workflow_json, input_schema,"
            " output_schema, is_enabled) VALUES (?, 'image', ?, '[]', '[]', 1)",
            ("存量·文生图", json.dumps(_load("qwen_image_t2i_API.json"))),
        )
        conn.execute(
            "INSERT INTO workflows (name, kind, workflow_json, input_schema,"
            " output_schema, is_enabled) VALUES (?, 'video', ?, '[]', '[]', 1)",
            ("存量·歧义", json.dumps(AMBIGUOUS)),
        )
        conn.execute(
            "INSERT INTO jobs (project_id, kind, workflow_id, status, payload_json,"
            " output_paths_json) VALUES (?, 'image', NULL, 'done', '{}', ?)",
            (pid, json.dumps(["/old/result.png"])),
        )
        conn.commit()
    finally:
        conn.close()

    import asyncio

    asyncio.run(migrate_db(settings.db_path))  # idempotent re-run

    wfs = {w["name"]: w for w in client.get("/api/workflows").json()}
    assert wfs["存量·文生图"]["usage"] == "text2image"
    assert wfs["存量·文生图"]["is_enabled"] is True
    assert wfs["存量·歧义"]["usage"] is None
    assert wfs["存量·歧义"]["is_enabled"] is False
    assert wfs["存量·歧义"]["kind"] == "video"  # old kind semantics untouched

    jobs = client.get(f"/api/jobs?project_id={pid}").json()
    assert any(
        j["status"] == "done" and j["kind"] == "image" and j["output_paths"] == ["/old/result.png"]
        for j in jobs
    )

    # a third run changes nothing (idempotent)
    asyncio.run(migrate_db(settings.db_path))
    wfs2 = {w["name"]: w for w in client.get("/api/workflows").json()}
    assert wfs2["存量·文生图"]["usage"] == "text2image"
    assert wfs2["存量·歧义"]["usage"] is None


def test_broken_qwen_import_stays_blocked_but_recorded(client):
    """The damaged Qwen export classifies fine (image2image); its structural
    lint errors still refuse generation — a broken file case, not a usage case."""
    wf_json = _load("qwen21_edit_sage_API.json")
    r = client.post(
        "/api/workflows/analyze", json={"workflow_json": wf_json}
    )
    assert r.json()["usage"] == "image2image"
    r = client.post(
        "/api/workflows", json={"name": "损坏·Qwen", "kind": "image", "workflow_json": wf_json}
    )
    assert r.status_code == 200
    wf = r.json()
    assert wf["usage"] == "image2image" and wf["is_enabled"] is True
    assert any(w["level"] == "error" for w in wf["warnings"])
    gen = client.post(
        "/api/playground/generate",
        json={"workflow_id": wf["id"], "input_values": {"4@prompt": "x"}},
    )
    assert gen.status_code == 400 and "重新导出" in gen.json()["detail"]