Spaces:
Running on Zero
Running on Zero
File size: 2,286 Bytes
36333c5 eb808a5 36333c5 eb808a5 36333c5 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 | """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())
|