455 lines
17 KiB
Python
455 lines
17 KiB
Python
"""五段创作工作台 (HEL-614) — stage lifecycle, context handoff (confirmed
|
|
drafts only), edit/confirm/restore, script import, in-flow generation via
|
|
the shared queue, and honest failure states. LLM is faked by monkeypatching
|
|
the workbench router's LLMClient; ComfyUI runs in dry_run placeholder mode."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import time
|
|
|
|
import httpx
|
|
import pytest
|
|
|
|
from calliope.config import settings
|
|
from calliope.routers import workbench as wb
|
|
|
|
IMAGE_WORKFLOW = {
|
|
"1": {
|
|
"class_type": "CLIPTextEncode",
|
|
"inputs": {"text": "默认提示词"},
|
|
"_meta": {"title": "(Input:prompt) 正向提示词"},
|
|
},
|
|
"9": {
|
|
"class_type": "SaveImage",
|
|
"inputs": {"images": "x"},
|
|
"_meta": {"title": "(Output:image) 图片输出"},
|
|
},
|
|
}
|
|
|
|
SCRIPT_TEXT = (
|
|
"《夜行便利店》\n第一场·内景·便利店·凌晨两点：店员小夏整理货架，风铃轻响，"
|
|
"一位穿米色风衣的女孩推门而入，径直走到关东煮前。\n小夏：还是老样子？\n女孩点头。"
|
|
)
|
|
|
|
STYLE_REPLY = json.dumps(
|
|
{
|
|
"directions": [
|
|
{
|
|
"id": "style1",
|
|
"name": "霓虹赛博",
|
|
"description": "青紫霓虹、湿润反光、浅景深",
|
|
"prompt": "cyberpunk convenience store, neon rim light, wet floor reflections",
|
|
"negative_prompt": "blurry, lowres",
|
|
},
|
|
{
|
|
"id": "style2",
|
|
"name": "暖调生活流",
|
|
"description": "暖黄顶光、柔焦、生活质感",
|
|
"prompt": "warm cozy interior, soft rim light, film grain",
|
|
"negative_prompt": "",
|
|
},
|
|
]
|
|
},
|
|
ensure_ascii=False,
|
|
)
|
|
|
|
STORYBOARD_REPLY = json.dumps(
|
|
{
|
|
"shots": [
|
|
{
|
|
"id": "shot1",
|
|
"scene": "1",
|
|
"shot_size": "全景",
|
|
"camera": "固定",
|
|
"description": "凌晨便利店全貌，货架整齐，灯光惨白",
|
|
"dialog": "",
|
|
"duration_sec": 3,
|
|
},
|
|
{
|
|
"id": "shot2",
|
|
"scene": "1",
|
|
"shot_size": "特写",
|
|
"camera": "推",
|
|
"description": "风铃轻响，女孩的手推开玻璃门",
|
|
"dialog": "小夏：还是老样子？",
|
|
"duration_sec": 2,
|
|
},
|
|
]
|
|
},
|
|
ensure_ascii=False,
|
|
)
|
|
|
|
ASSETS_REPLY = json.dumps(
|
|
{
|
|
"assets": [
|
|
{
|
|
"id": "asset1",
|
|
"kind": "character",
|
|
"name": "小夏",
|
|
"description": "夜班店员，白色制服",
|
|
"prompt": "convenience store clerk, white uniform, name tag, half body",
|
|
},
|
|
{
|
|
"id": "asset2",
|
|
"kind": "location",
|
|
"name": "便利店",
|
|
"description": "深夜便利店内部",
|
|
"prompt": "japanese convenience store interior at night, bright shelves",
|
|
},
|
|
]
|
|
},
|
|
ensure_ascii=False,
|
|
)
|
|
|
|
SHOT_REPLY = json.dumps(
|
|
{
|
|
"shots": [
|
|
{
|
|
"id": "vshot1",
|
|
"shot_no": "S1",
|
|
"description": "开场全景",
|
|
"prompt": "slow push in on convenience store shelves at night, cinematic",
|
|
"duration_sec": 4,
|
|
},
|
|
{
|
|
"id": "vshot2",
|
|
"shot_no": "S2",
|
|
"description": "风铃特写",
|
|
"prompt": "close up wind chime, door opens, warm light spill",
|
|
"duration_sec": 3,
|
|
},
|
|
]
|
|
},
|
|
ensure_ascii=False,
|
|
)
|
|
|
|
|
|
class FakeLLM:
|
|
"""Scripted replies; records every chat() call for context assertions."""
|
|
|
|
replies: list[str] = []
|
|
calls: list[list[dict]] = []
|
|
|
|
@classmethod
|
|
def reset(cls, replies: list[str]) -> None:
|
|
cls.replies = list(replies)
|
|
cls.calls = []
|
|
|
|
@classmethod
|
|
def for_role(cls, role: str, *, timeout: float = 120.0) -> "FakeLLM":
|
|
return cls()
|
|
|
|
async def chat(self, messages, temperature=0.7, **kwargs) -> str:
|
|
type(self).calls.append(messages)
|
|
if type(self).replies:
|
|
return type(self).replies.pop(0)
|
|
return json.dumps({"text": SCRIPT_TEXT}, ensure_ascii=False)
|
|
|
|
async def close(self) -> None:
|
|
return None
|
|
|
|
|
|
class RefusingLLM(FakeLLM):
|
|
@classmethod
|
|
def for_role(cls, role: str, *, timeout: float = 120.0) -> "RefusingLLM":
|
|
return cls()
|
|
|
|
async def chat(self, messages, temperature=0.7, **kwargs) -> str:
|
|
type(self).calls.append(messages)
|
|
raise httpx.ConnectError("connection refused")
|
|
|
|
|
|
def _make_project(client, **fields) -> int:
|
|
payload = {"title": fields.get("title", "夜行便利店"), "idea": "凌晨便利店的温暖相遇"}
|
|
r = client.post("/api/projects", json=payload)
|
|
assert r.status_code == 200, r.text
|
|
return r.json()["id"]
|
|
|
|
|
|
def _overview(client, pid: int) -> dict:
|
|
r = client.get(f"/api/workbench/projects/{pid}")
|
|
assert r.status_code == 200, r.text
|
|
return r.json()
|
|
|
|
|
|
def _stage(overview: dict, key: str) -> dict:
|
|
return next(s for s in overview["stages"] if s["stage_key"] == key)
|
|
|
|
|
|
def _wait_stage_status(client, pid: int, key: str, statuses: set[str], timeout: float = 10.0) -> dict:
|
|
deadline = time.time() + timeout
|
|
stage = None
|
|
while time.time() < deadline:
|
|
stage = _stage(_overview(client, pid), key)
|
|
if stage["status"] in statuses:
|
|
return stage
|
|
time.sleep(0.2)
|
|
pytest.fail(f"stage {key} stuck in {stage and stage['status']!r}, want {statuses}")
|
|
|
|
|
|
def _run(client, pid: int, key: str, instruction: str | None = None):
|
|
body = {"instruction": instruction} if instruction else {}
|
|
return client.post(f"/api/workbench/projects/{pid}/stages/{key}/run", json=body)
|
|
|
|
|
|
def _confirm(client, pid: int, key: str, **extra):
|
|
return client.post(f"/api/workbench/projects/{pid}/stages/{key}/confirm", json=extra or None)
|
|
|
|
|
|
def test_assistant_notes_keep_confirmation_and_survive_artifact_edits(client):
|
|
pid = _make_project(client)
|
|
note_url = f"/api/workbench/projects/{pid}/stages/style/note"
|
|
r = client.patch(note_url, json={"text": "建议先比较冷暖两个方向。"})
|
|
assert r.status_code == 200
|
|
assert r.json()["stage"]["status"] == "todo"
|
|
assert r.json()["stage"]["assistant_note"] == "建议先比较冷暖两个方向。"
|
|
assert _confirm(client, pid, "style").status_code == 400
|
|
# An ordinary structured save must keep the separate advice.
|
|
r = client.patch(f"/api/workbench/projects/{pid}/stages/style", json={"content": json.loads(STYLE_REPLY)})
|
|
assert r.status_code == 200
|
|
assert r.json()["stage"]["assistant_note"] == "建议先比较冷暖两个方向。"
|
|
# Notes on confirmed content don't change its status or confirmed snapshot.
|
|
client.post(f"/api/workbench/projects/{pid}/stages/script/import", json={"text": SCRIPT_TEXT})
|
|
before = _stage(_overview(client, pid), "script")
|
|
r = client.patch(f"/api/workbench/projects/{pid}/stages/script/note", json={"text": "下一版建议加强人物动机。"})
|
|
assert r.status_code == 200
|
|
after = r.json()["stage"]
|
|
assert after["status"] == before["status"] == "confirmed"
|
|
assert after["confirmed_content"] == before["confirmed_content"]
|
|
assert after["draft_edited"] == before["draft_edited"]
|
|
assert _confirm(client, pid, "script").status_code == 200
|
|
confirmed = _stage(_overview(client, pid), "script")
|
|
assert "assistant_note" not in confirmed["confirmed_content"]
|
|
client.patch(f"/api/workbench/projects/{pid}/stages/script/note", json={"text": "保留确认后的新建议。"})
|
|
restored = client.post(f"/api/workbench/projects/{pid}/stages/script/restore").json()["stage"]
|
|
assert restored["assistant_note"] == "保留确认后的新建议。"
|
|
assert restored["confirmed_content"] == confirmed["confirmed_content"]
|
|
assert client.patch(note_url, json={"text": ""}).json()["stage"]["assistant_note"] == ""
|
|
assert client.patch(note_url, json={"text": "x" * 200_001}).status_code == 422
|
|
assert client.patch(f"/api/workbench/projects/{pid}/stages/unknown/note", json={"text": "x"}).status_code == 404
|
|
|
|
|
|
# ─────────────────────────────────────────────────────────────────────────
|
|
|
|
|
|
def test_overview_ensures_five_todo_stages(client):
|
|
pid = _make_project(client)
|
|
overview = _overview(client, pid)
|
|
assert [s["stage_key"] for s in overview["stages"]] == [
|
|
"script",
|
|
"style",
|
|
"storyboard",
|
|
"assets",
|
|
"shot",
|
|
]
|
|
assert all(s["status"] == "todo" for s in overview["stages"])
|
|
assert overview["project"]["title"] == "夜行便利店"
|
|
|
|
|
|
def test_run_requires_previous_stage_confirmed(client, monkeypatch):
|
|
monkeypatch.setattr(wb, "LLMClient", FakeLLM)
|
|
FakeLLM.reset([])
|
|
pid = _make_project(client)
|
|
r = _run(client, pid, "style")
|
|
assert r.status_code == 409
|
|
assert "剧本创作" in r.json()["detail"]
|
|
|
|
|
|
def test_five_stage_flow_confirmed_context_handoff(client, monkeypatch):
|
|
"""从创意起步端到端：每段生成→确认；只有已确认内容进入下一段上下文。"""
|
|
monkeypatch.setattr(wb, "LLMClient", FakeLLM)
|
|
FakeLLM.reset(
|
|
[
|
|
json.dumps({"text": SCRIPT_TEXT}, ensure_ascii=False),
|
|
STYLE_REPLY,
|
|
STORYBOARD_REPLY,
|
|
ASSETS_REPLY,
|
|
SHOT_REPLY,
|
|
]
|
|
)
|
|
pid = _make_project(client)
|
|
|
|
# 1. 剧本创作
|
|
assert _run(client, pid, "script").status_code == 200
|
|
stage = _wait_stage_status(client, pid, "script", {"awaiting_confirm"})
|
|
assert stage["content"]["text"] == SCRIPT_TEXT
|
|
assert _confirm(client, pid, "script").status_code == 200
|
|
|
|
# 2. 视觉设计 — 上下文带剧本
|
|
assert _run(client, pid, "style").status_code == 200
|
|
stage = _wait_stage_status(client, pid, "style", {"awaiting_confirm"})
|
|
assert len(stage["content"]["directions"]) == 2
|
|
assert SCRIPT_TEXT[:10] in FakeLLM.calls[-1][1]["content"]
|
|
r = _confirm(client, pid, "style", chosen_style_id="style2")
|
|
assert r.status_code == 200
|
|
assert r.json()["stage"]["chosen_style_id"] == "style2"
|
|
|
|
# 3. 分镜设计 — 上下文带剧本 + 已选风格（暖调生活流，非默认第一套）
|
|
assert _run(client, pid, "storyboard").status_code == 200
|
|
_wait_stage_status(client, pid, "storyboard", {"awaiting_confirm"})
|
|
ctx = FakeLLM.calls[-1][1]["content"]
|
|
assert "暖调生活流" in ctx
|
|
assert SCRIPT_TEXT[:10] in ctx
|
|
assert _confirm(client, pid, "storyboard").status_code == 200
|
|
|
|
# 4. 素材资产 — 上下文带分镜
|
|
assert _run(client, pid, "assets").status_code == 200
|
|
_wait_stage_status(client, pid, "assets", {"awaiting_confirm"})
|
|
ctx = FakeLLM.calls[-1][1]["content"]
|
|
assert "风铃轻响" in ctx
|
|
assert _confirm(client, pid, "assets").status_code == 200
|
|
|
|
# 5. 镜头脚本 — 上下文带素材清单
|
|
assert _run(client, pid, "shot").status_code == 200
|
|
stage = _wait_stage_status(client, pid, "shot", {"awaiting_confirm"})
|
|
ctx = FakeLLM.calls[-1][1]["content"]
|
|
assert "小夏" in ctx
|
|
assert stage["content"]["shots"][0]["prompt"].startswith("slow push")
|
|
|
|
# 进度反映到项目列表（真实阶段数据）
|
|
r = client.get("/api/projects")
|
|
stats = next(p for p in r.json() if p["id"] == pid)["stats"]
|
|
assert stats["stage_confirmed"] == 4
|
|
assert stats["stage_status"]["shot"] == "awaiting_confirm"
|
|
|
|
|
|
def test_run_failure_keeps_confirmed_snapshot_and_reads_chinese(client, monkeypatch):
|
|
monkeypatch.setattr(wb, "LLMClient", RefusingLLM)
|
|
pid = _make_project(client)
|
|
# 先用正常 LLM 确认剧本
|
|
monkeypatch.setattr(wb, "LLMClient", FakeLLM)
|
|
FakeLLM.reset([json.dumps({"text": SCRIPT_TEXT}, ensure_ascii=False)])
|
|
_run(client, pid, "script")
|
|
_wait_stage_status(client, pid, "script", {"awaiting_confirm"})
|
|
_confirm(client, pid, "script")
|
|
# 视觉设计生成失败（连接拒绝）
|
|
monkeypatch.setattr(wb, "LLMClient", RefusingLLM)
|
|
assert _run(client, pid, "style").status_code == 200
|
|
stage = _wait_stage_status(client, pid, "style", {"todo"})
|
|
assert "连接大模型服务失败" in (stage["last_run"]["error"] or "")
|
|
# 剧本阶段不受影响
|
|
assert _stage(_overview(client, pid), "script")["status"] == "confirmed"
|
|
|
|
|
|
def test_edit_after_confirm_and_restore(client, monkeypatch):
|
|
monkeypatch.setattr(wb, "LLMClient", FakeLLM)
|
|
FakeLLM.reset([json.dumps({"text": SCRIPT_TEXT}, ensure_ascii=False)])
|
|
pid = _make_project(client)
|
|
_run(client, pid, "script")
|
|
_wait_stage_status(client, pid, "script", {"awaiting_confirm"})
|
|
_confirm(client, pid, "script")
|
|
|
|
# 编辑已确认阶段 → 变回待确认草稿，已确认快照保留
|
|
edited = {"text": SCRIPT_TEXT + "\n（结尾新增：女孩微笑离开。）"}
|
|
r = client.patch(f"/api/workbench/projects/{pid}/stages/script", json={"content": edited})
|
|
assert r.status_code == 200
|
|
stage = r.json()["stage"]
|
|
assert stage["status"] == "awaiting_confirm"
|
|
assert stage["draft_edited"] is True
|
|
assert stage["confirmed_content"]["text"] == SCRIPT_TEXT
|
|
|
|
# 恢复已确认版本
|
|
r = client.post(f"/api/workbench/projects/{pid}/stages/script/restore")
|
|
assert r.status_code == 200
|
|
stage = r.json()["stage"]
|
|
assert stage["status"] == "confirmed"
|
|
assert stage["content"]["text"] == SCRIPT_TEXT
|
|
|
|
|
|
def test_import_script_confirms_and_unlocks_style(client, monkeypatch):
|
|
monkeypatch.setattr(wb, "LLMClient", FakeLLM)
|
|
FakeLLM.reset([STYLE_REPLY])
|
|
pid = _make_project(client)
|
|
r = client.post(f"/api/workbench/projects/{pid}/stages/script/import", json={"text": SCRIPT_TEXT})
|
|
assert r.status_code == 200
|
|
assert r.json()["stage"]["status"] == "confirmed"
|
|
# 跳过剧本创作，直接从视觉设计继续
|
|
assert _run(client, pid, "style").status_code == 200
|
|
_wait_stage_status(client, pid, "style", {"awaiting_confirm"})
|
|
# 太短的导入被拒绝
|
|
r = client.post(
|
|
f"/api/workbench/projects/{pid}/stages/script/import", json={"text": "太短"}
|
|
)
|
|
assert r.status_code == 400
|
|
assert "太短" in r.json()["detail"]
|
|
|
|
|
|
def test_generate_without_workflow_is_chinese_400(client):
|
|
pid = _make_project(client)
|
|
r = client.post(
|
|
f"/api/workbench/projects/{pid}/generate",
|
|
json={"kind": "image", "stage_key": "assets", "target_id": "asset1", "prompt": "a cat"},
|
|
)
|
|
assert r.status_code == 400
|
|
assert "没有可用的生图工作流" in r.json()["detail"]
|
|
|
|
|
|
def test_generate_image_job_returns_to_project(client):
|
|
pid = _make_project(client)
|
|
# 手动放入素材阶段草稿（带 asset1）
|
|
r = client.patch(
|
|
f"/api/workbench/projects/{pid}/stages/assets",
|
|
json={"content": json.loads(ASSETS_REPLY)},
|
|
)
|
|
assert r.status_code == 200
|
|
wf = client.post(
|
|
"/api/workflows",
|
|
json={"name": "假·文生图", "kind": "image", "workflow_json": IMAGE_WORKFLOW},
|
|
)
|
|
assert wf.status_code == 200, wf.text
|
|
|
|
r = client.post(
|
|
f"/api/workbench/projects/{pid}/generate",
|
|
json={
|
|
"kind": "image",
|
|
"stage_key": "assets",
|
|
"target_id": "asset1",
|
|
"prompt": "convenience store clerk portrait",
|
|
},
|
|
)
|
|
assert r.status_code == 200, r.text
|
|
job = r.json()["job"]
|
|
assert job["payload"]["source"] == "workbench"
|
|
assert job["payload"]["target_id"] == "asset1"
|
|
assert job["payload"]["input_values"]["1"] == "convenience store clerk portrait"
|
|
|
|
# dry_run worker completes with a placeholder; overview enriches the asset
|
|
deadline = time.time() + 40
|
|
job_done = False
|
|
mine = None
|
|
while time.time() < deadline:
|
|
jobs = client.get(f"/api/jobs?project_id={pid}").json()
|
|
mine = next((j for j in jobs if j["id"] == job["id"]), None)
|
|
if mine and mine["status"] in ("done", "failed"):
|
|
job_done = mine["status"] == "done"
|
|
break
|
|
time.sleep(0.4)
|
|
assert job_done, f"workbench job did not finish: {mine}"
|
|
stage = _stage(_overview(client, pid), "assets")
|
|
asset = next(a for a in stage["content"]["assets"] if a["id"] == "asset1")
|
|
assert asset["job"]["status"] == "done"
|
|
assert asset["job"]["output"] and asset["job"]["output"].endswith(".png")
|
|
|
|
|
|
def test_project_list_stats_include_active_jobs(client):
|
|
pid = _make_project(client)
|
|
wf = client.post(
|
|
"/api/workflows",
|
|
json={"name": "假·文生图2", "kind": "image", "workflow_json": IMAGE_WORKFLOW},
|
|
)
|
|
wf_id = wf.json()["id"]
|
|
r = client.post(
|
|
f"/api/workbench/projects/{pid}/generate",
|
|
json={"kind": "image", "stage_key": "assets", "prompt": "x"},
|
|
)
|
|
assert r.status_code == 200
|
|
r = client.get("/api/projects")
|
|
stats = next(p for p in r.json() if p["id"] == pid)["stats"]
|
|
assert stats["active_jobs"] >= 1
|
|
assert stats["stage_confirmed"] == 0
|
|
# a project that never opened the workbench reports no stage rows at all
|
|
pid2 = _make_project(client, title="未开工")
|
|
stats2 = next(p for p in client.get("/api/projects").json() if p["id"] == pid2)["stats"]
|
|
assert stats2["stage_status"] == {} and stats2["stage_confirmed"] == 0
|