Spaces:
Running
Running
File size: 8,497 Bytes
7cc81cb | 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 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 | from __future__ import annotations
import base64
from pathlib import Path
import pytest
from fastapi.testclient import TestClient
from app.container import build_container
from app.core.exceptions import TemplateValidationError
from app.models.media import InputMedia, MediaSource, ResolvedRequest
from app.templates.executor import OPERATION_BINDINGS
from app.templates.loader import TemplateLoader
from app.templates.registry import TemplateRegistry
from app.templates.validator import TemplateValidator
from main import create_app
EXPECTED_CATEGORIES = {
"branding",
"conversion",
"faceless",
"lyrics",
"motivation",
"podcast",
"social",
"subtitles",
"utility",
"youtube",
}
def test_builtin_templates_are_loaded_dynamically(settings) -> None:
registry = build_container(settings).template_registry
assert registry.count == 71
assert set(registry.categories()) == EXPECTED_CATEGORIES
assert registry.get("instagram_reel").version == 1
assert registry.get("instagram_reel@latest").version == 1
assert registry.get("instagram_reel@1").name == "Instagram Reel"
def test_parameter_substitution_preserves_declared_types(settings) -> None:
registry = build_container(settings).template_registry
prepared = registry.prepare("youtube_shorts@1", {"crf": 19, "max_duration": 42.5})
trim_step = prepared.pipeline[0]
compress_step = prepared.pipeline[-1]
assert trim_step.parameters["duration"] == 42.5
assert isinstance(trim_step.parameters["duration"], float)
assert compress_step.parameters["crf"] == 19
assert isinstance(compress_step.parameters["crf"], int)
def test_template_registry_keeps_old_versions(tmp_path: Path) -> None:
root = tmp_path / "templates"
root.mkdir()
(root / "versions.yaml").write_text(
"""
templates:
- id: sample
name: Sample One
category: custom
description: First stable workflow.
author: Tests
version: 1
tags: [test]
estimated_runtime: fast
supported_inputs: [video]
supported_outputs: [source]
parameters: {}
pipeline: [{operation: download}]
output: {format: source}
examples: []
- id: sample
name: Sample Two
category: custom
description: Second stable workflow.
author: Tests
version: 2
tags: [test]
estimated_runtime: fast
supported_inputs: [video]
supported_outputs: [source]
parameters: {}
pipeline: [{operation: download}]
output: {format: source}
examples: []
""",
encoding="utf-8",
)
validator = TemplateValidator(set(OPERATION_BINDINGS))
registry = TemplateRegistry(TemplateLoader(root, validator), validator)
assert registry.get("sample@1").name == "Sample One"
assert registry.get("sample@2").name == "Sample Two"
assert registry.get("sample@latest").version == 2
assert registry.get("sample").version == 2
def test_invalid_yaml_operation_is_never_registered(tmp_path: Path) -> None:
root = tmp_path / "templates"
root.mkdir()
(root / "invalid.yaml").write_text(
"""
id: invalid
name: Invalid
category: custom
description: Invalid operation must fail loading.
author: Tests
version: 1
tags: [test]
estimated_runtime: fast
supported_inputs: [video]
supported_outputs: [mp4]
parameters: {}
pipeline: [{operation: shell_command}]
output: {format: mp4}
examples: []
""",
encoding="utf-8",
)
validator = TemplateValidator(set(OPERATION_BINDINGS))
with pytest.raises(TemplateValidationError, match="unsupported operation"):
TemplateRegistry(TemplateLoader(root, validator), validator)
def test_invalid_yaml_syntax_is_never_loaded(tmp_path: Path) -> None:
root = tmp_path / "templates"
root.mkdir()
(root / "broken.yaml").write_text("id: broken\npipeline: [\n", encoding="utf-8")
validator = TemplateValidator(set(OPERATION_BINDINGS))
with pytest.raises(TemplateValidationError, match="syntax"):
TemplateLoader(root, validator).load()
def test_required_and_typed_parameters_are_enforced(tmp_path: Path) -> None:
root = tmp_path / "templates"
root.mkdir()
(root / "required.yaml").write_text(
"""
id: required_sample
name: Required Sample
category: custom
description: Exercise strict runtime parameter validation.
author: Tests
version: 1
tags: [test]
estimated_runtime: fast
supported_inputs: [video]
supported_outputs: [mp4]
parameters:
width: {type: integer, required: true, minimum: 2}
pipeline: [{operation: resize, width: "{{ width }}", height: 720}]
output: {format: mp4}
examples: []
""",
encoding="utf-8",
)
validator = TemplateValidator(set(OPERATION_BINDINGS))
registry = TemplateRegistry(TemplateLoader(root, validator), validator)
with pytest.raises(TemplateValidationError, match="Required"):
registry.prepare("required_sample", {})
with pytest.raises(TemplateValidationError, match="must be integer"):
registry.prepare("required_sample", {"width": "1080"})
assert (
registry.prepare("required_sample", {"width": 1080}).pipeline[0].parameters["width"] == 1080
)
async def test_template_executor_calls_existing_operation(settings, tmp_path, monkeypatch) -> None:
container = build_container(settings)
source = tmp_path / "source.wav"
source.write_bytes(b"RIFF-test-audio")
async def fake_probe(inputs):
return [
{
"filename": media.filename,
"mime_type": media.mime_type,
"size": media.size,
}
for media in inputs
]
async def fake_ffmpeg(args, *, operation, timeout=None):
output = Path(args[-1])
output.parent.mkdir(parents=True, exist_ok=True)
output.write_bytes(b"ID3-template-output")
monkeypatch.setattr(container.processor, "probe_inputs", fake_probe)
monkeypatch.setattr(container.ffmpeg, "run", fake_ffmpeg)
resolved = ResolvedRequest(
request_id="5fdbe750-4cb7-4f87-aa5c-3df50c3a6629",
inputs=[
InputMedia(
source=MediaSource.MULTIPART,
filename=source.name,
mime_type="audio/wav",
temp_path=source,
size=source.stat().st_size,
)
],
)
response = await container.template_executor.execute(resolved, "mp3@1", {})
assert response.success is True
assert response.download_url is not None
assert response.metadata["template"]["id"] == "mp3"
assert response.metadata["operations"] == ["convert_audio"]
published = container.cleanup.resolve_download(
resolved.request_id, Path(response.download_url).name
)
assert published.read_bytes() == b"ID3-template-output"
def test_template_rest_endpoints_and_nested_input(settings, monkeypatch) -> None:
application = create_app(settings)
container = application.state.container
async def fake_probe(inputs):
return [
{
"filename": media.filename,
"mime_type": media.mime_type,
"size": media.size,
}
for media in inputs
]
async def fake_ffmpeg(args, *, operation, timeout=None):
output = Path(args[-1])
output.parent.mkdir(parents=True, exist_ok=True)
output.write_bytes(b"ID3-rest-template")
monkeypatch.setattr(container.processor, "probe_inputs", fake_probe)
monkeypatch.setattr(container.ffmpeg, "run", fake_ffmpeg)
with TestClient(application) as client:
listing = client.get("/v1/templates")
categories = client.get("/v1/templates/categories")
details = client.get("/v1/templates/instagram_reel@1")
execution = client.post(
"/v1/templates/run",
json={
"template": "mp3@latest",
"input": {
"base64": base64.b64encode(b"RIFF-rest-audio").decode(),
"filename": "audio.wav",
"mime_type": "audio/wav",
},
"parameters": {},
},
)
assert listing.status_code == 200
assert listing.json()["metadata"]["count"] == 71
assert set(categories.json()["metadata"]["categories"]) == EXPECTED_CATEGORIES
assert details.json()["metadata"]["template"]["version"] == 1
assert execution.status_code == 200
assert execution.json()["metadata"]["template"]["id"] == "mp3"
|