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"