tests/test_workflow.py

"""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())