"""No-download verification of full workflow ordering and rollback boundaries.""" from __future__ import annotations from pathlib import Path from typing import Any, Callable from config import Settings from core.executor import InferenceCommand from core.workflow import WorkflowService from routes.schemas import WorkflowRequest from utils.files import OutputManager class FakeAdapter: def __init__(self, name: str) -> None: self.name = name @staticmethod def _write(kwargs: dict[str, Any]) -> Path: path = kwargs.get("output_path") or kwargs.get("subtitle_path") path.write_bytes(b"generated") return path def generate(self, **kwargs: Any) -> Path: return self._write(kwargs) def synthesize(self, **kwargs: Any) -> Path: return self._write(kwargs) def transcribe(self, **kwargs: Any) -> dict[str, Any]: self._write(kwargs) return {"text": "hello", "language": "en"} class FakeTasks: def __init__(self) -> None: self.calls: list[str] = [] async def run_exclusive( self, request_id: str, label: str, action: Callable[[], Any] ) -> Any: assert label == "workflow" return action() def invoke_direct(self, command: InferenceCommand) -> Any: self.calls.append(command.model_name) adapter = FakeAdapter(command.model_name) method = getattr(adapter, command.method_name) return method(**command.arguments) async def test_workflow_runs_in_required_order(tmp_path: Path) -> None: settings = Settings( output_folder=tmp_path / "output", tmp_folder=tmp_path / "tmp", static_folder=tmp_path / "static", device="cpu", ) outputs = OutputManager(settings) outputs.initialize() tasks = FakeTasks() workflow = WorkflowService(settings, tasks, outputs) # type: ignore[arg-type] payload = WorkflowRequest( title="Episode One", script="Hello from the gateway.", image_prompt="A futuristic city", video_prompt="Camera slowly zooms", ) assets = await workflow.run(payload, "request-1") assert tasks.calls == ["flux", "kokoro", "musicgen", "wan", "whisper"] assert all(path.is_file() for path in assets.values())