- backend: project_stages/stage_runs tables; /api/workbench router (overview, run agent with confirmed-upstream context only, edit draft, confirm, restore, script import, in-flow image/video generation via the shared queue with source=workbench); project stats gain real stage_confirmed/stage_status/active_jobs - frontend: /workbench picker + /workbench/[id] five-stage page (stepper, per-stage editors, style direction chooser, storyboard table, asset prompt cards with in-flow generation, shot video prompts with player, agent + task side rail); nav 创作工作台 now opens it; home entries go through the workbench (导入剧本 skips 剧本创作); project cards show real five-stage progress; i18n keys in all seven dictionaries - tests: 9 backend workbench tests (flow handoff, edit/confirm/restore, import, generation, Chinese failure states) Co-authored-by: multica-agent <github@multica.ai>
164 lines
5.3 KiB
Python
164 lines
5.3 KiB
Python
from __future__ import annotations
|
|
|
|
import tempfile
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
from calliope.config import Settings
|
|
from calliope.main import create_app
|
|
from calliope.db import migrate_db
|
|
|
|
|
|
@pytest.fixture
|
|
def client(monkeypatch):
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
import calliope.config as config_module
|
|
import asyncio
|
|
|
|
s = config_module.settings
|
|
prev = {
|
|
"data_dir": s.data_dir,
|
|
"assets_dir": s.assets_dir,
|
|
"dry_run": s.dry_run,
|
|
}
|
|
s.data_dir = Path(tmpdir)
|
|
s.assets_dir = Path(tmpdir) / "assets"
|
|
s.dry_run = True
|
|
monkeypatch.setattr(config_module.Settings, "save_config_file", lambda self: None)
|
|
asyncio.run(migrate_db(s.db_path))
|
|
app = create_app()
|
|
try:
|
|
with TestClient(app) as c:
|
|
yield c
|
|
finally:
|
|
for k, v in prev.items():
|
|
setattr(s, k, v)
|
|
|
|
|
|
def test_health(client):
|
|
r = client.get("/api/health")
|
|
assert r.status_code == 200
|
|
assert r.json()["status"] == "ok"
|
|
|
|
|
|
def test_health_version_matches_package_metadata(client):
|
|
"""PR #47: the served version must come from package metadata
|
|
(single-sourced via calliope.__version__), never a stale hard-coded
|
|
literal — 1.3.2 shipped reporting 1.2.1, 1.4.1 reported 1.4.0."""
|
|
import tomllib
|
|
from pathlib import Path
|
|
|
|
pyproject = Path(__file__).resolve().parents[1] / "pyproject.toml"
|
|
data = tomllib.loads(pyproject.read_text(encoding="utf-8"))
|
|
# dynamic = ["version"] means pyproject no longer carries the literal;
|
|
# the truth lives in src/calliope/__init__.py.
|
|
from calliope import __version__
|
|
|
|
assert data["project"].get("version") is None, (
|
|
"pyproject pins a literal version — single-source it via "
|
|
"calliope.__version__ (PR #47)"
|
|
)
|
|
|
|
r = client.get("/api/health")
|
|
assert r.json()["version"] == __version__
|
|
|
|
|
|
def test_create_and_list_project(client):
|
|
r = client.post("/api/projects", json={"title": "Test", "idea": "idea"})
|
|
assert r.status_code == 200
|
|
data = r.json()
|
|
assert data["title"] == "Test"
|
|
assert data["idea"] == "idea"
|
|
pid = data["id"]
|
|
|
|
r = client.get("/api/projects")
|
|
assert r.status_code == 200
|
|
assert any(p["id"] == pid for p in r.json())
|
|
|
|
|
|
def test_list_projects_includes_stats(client):
|
|
r = client.post("/api/projects", json={"title": "ListStats"})
|
|
pid = r.json()["id"]
|
|
r = client.get("/api/projects")
|
|
assert r.status_code == 200
|
|
item = next(p for p in r.json() if p["id"] == pid)
|
|
assert item["stats"] == {
|
|
"scene_count": 0,
|
|
"character_count": 0,
|
|
"asset_ready_count": 0,
|
|
"asset_total_count": 0,
|
|
# Five-stage workbench progress (HEL-614): fresh project = not started.
|
|
"stage_confirmed": 0,
|
|
"stage_status": {},
|
|
"active_jobs": 0,
|
|
}
|
|
|
|
|
|
def test_get_project_with_stats(client):
|
|
r = client.post("/api/projects", json={"title": "Stats"})
|
|
pid = r.json()["id"]
|
|
r = client.get(f"/api/projects/{pid}")
|
|
assert r.status_code == 200
|
|
data = r.json()
|
|
assert data["stats"]["scene_count"] == 0
|
|
|
|
|
|
def test_update_project(client):
|
|
r = client.post("/api/projects", json={"title": "Old"})
|
|
pid = r.json()["id"]
|
|
r = client.patch(f"/api/projects/{pid}", json={"title": "New"})
|
|
assert r.status_code == 200
|
|
assert r.json()["title"] == "New"
|
|
|
|
|
|
def test_update_project_cover(client):
|
|
r = client.post("/api/projects", json={"title": "Cover"})
|
|
pid = r.json()["id"]
|
|
assert r.json()["cover_path"] is None
|
|
|
|
r = client.patch(f"/api/projects/{pid}", json={"cover_path": "/tmp/cover.png"})
|
|
assert r.status_code == 200
|
|
assert r.json()["cover_path"] == "/tmp/cover.png"
|
|
|
|
r = client.get(f"/api/projects/{pid}")
|
|
assert r.json()["cover_path"] == "/tmp/cover.png"
|
|
r = client.get("/api/projects")
|
|
assert next(p for p in r.json() if p["id"] == pid)["cover_path"] == "/tmp/cover.png"
|
|
|
|
# Explicit null clears the cover again.
|
|
r = client.patch(f"/api/projects/{pid}", json={"cover_path": None})
|
|
assert r.status_code == 200
|
|
assert r.json()["cover_path"] is None
|
|
|
|
# Unrelated patches leave the (cleared) cover untouched.
|
|
r = client.patch(f"/api/projects/{pid}", json={"title": "Cover 2"})
|
|
assert r.json()["title"] == "Cover 2"
|
|
assert r.json()["cover_path"] is None
|
|
|
|
|
|
def test_create_project_accepts_name_alias(client):
|
|
"""LLM agents and external API clients often emit `name` for the required
|
|
title field — the schema aliases it instead of returning an opaque 422."""
|
|
r = client.post("/api/projects", json={"name": "Aliased", "idea": "idea"})
|
|
assert r.status_code == 200
|
|
assert r.json()["title"] == "Aliased"
|
|
|
|
|
|
def test_update_project_accepts_name_alias(client):
|
|
r = client.post("/api/projects", json={"title": "Old"})
|
|
pid = r.json()["id"]
|
|
r = client.patch(f"/api/projects/{pid}", json={"name": "Renamed"})
|
|
assert r.status_code == 200
|
|
assert r.json()["title"] == "Renamed"
|
|
|
|
|
|
def test_delete_project(client):
|
|
r = client.post("/api/projects", json={"title": "Delete"})
|
|
pid = r.json()["id"]
|
|
r = client.delete(f"/api/projects/{pid}")
|
|
assert r.status_code == 200
|
|
r = client.get(f"/api/projects/{pid}")
|
|
assert r.status_code == 404
|