Spaces:
Running
Running
| from __future__ import annotations | |
| import base64 | |
| from uuid import uuid4 | |
| import pytest | |
| from starlette.requests import Request | |
| from app.container import build_container | |
| from app.core.exceptions import InputError | |
| from app.models.media import MediaSource | |
| def json_request(payload: bytes) -> Request: | |
| sent = False | |
| async def receive(): | |
| nonlocal sent | |
| if sent: | |
| return {"type": "http.disconnect"} | |
| sent = True | |
| return {"type": "http.request", "body": payload, "more_body": False} | |
| request = Request( | |
| { | |
| "type": "http", | |
| "method": "POST", | |
| "path": "/v1/probe", | |
| "headers": [(b"content-type", b"application/json")], | |
| "query_string": b"", | |
| }, | |
| receive, | |
| ) | |
| request.state.request_id = str(uuid4()) | |
| return request | |
| async def test_resolves_json_base64(settings) -> None: | |
| container = build_container(settings) | |
| encoded = base64.b64encode(b"ID3-not-real-audio").decode() | |
| request = json_request( | |
| ('{"base64":"%s","filename":"sample.mp3","format":"wav"}' % encoded).encode() | |
| ) | |
| resolved = await container.resolver.resolve(request) | |
| assert resolved.primary.source is MediaSource.JSON_BASE64 | |
| assert resolved.primary.filename == "sample.mp3" | |
| assert resolved.primary.temp_path.read_bytes() == b"ID3-not-real-audio" | |
| assert resolved.params["filename"] == "sample.mp3" | |
| assert resolved.params["format"] == "wav" | |
| async def test_resolves_n8n_binary_property(settings) -> None: | |
| container = build_container(settings) | |
| encoded = base64.b64encode(b"audio").decode() | |
| payload = ( | |
| '{"binary":{"audio":{"data":"%s","fileName":"voice.mp3",' | |
| '"mimeType":"audio/mpeg"}}}' % encoded | |
| ).encode() | |
| resolved = await container.resolver.resolve(json_request(payload)) | |
| assert resolved.primary.source is MediaSource.N8N_BINARY | |
| assert resolved.primary.filename == "voice.mp3" | |
| assert resolved.primary.temp_path.read_bytes() == b"audio" | |
| async def test_resolves_nested_template_input(settings) -> None: | |
| container = build_container(settings) | |
| encoded = base64.b64encode(b"RIFF-template-audio").decode() | |
| request = json_request( | |
| ( | |
| '{"template":"mp3","input":{"base64":"%s",' | |
| '"filename":"source.wav","mime_type":"audio/wav"},"parameters":{}}' % encoded | |
| ).encode() | |
| ) | |
| resolved = await container.resolver.resolve(request) | |
| assert resolved.primary.source is MediaSource.JSON_BASE64 | |
| assert resolved.primary.filename == "source.wav" | |
| assert resolved.params["template"] == "mp3" | |
| assert resolved.params["parameters"] == {} | |
| async def test_resolve_payload_copies_managed_temp_file(settings) -> None: | |
| settings.output_dir.mkdir(parents=True) | |
| source = settings.output_dir / "previous" / "clip.mp3" | |
| source.parent.mkdir() | |
| source.write_bytes(b"ID3-managed-media") | |
| container = build_container(settings) | |
| resolved = await container.resolver.resolve_payload({"temp_path": str(source)}, str(uuid4())) | |
| assert resolved.primary.source is MediaSource.LOCAL_PATH | |
| assert resolved.primary.temp_path != source | |
| assert resolved.primary.temp_path.read_bytes() == source.read_bytes() | |
| async def test_resolve_payload_rejects_unmanaged_path(settings, tmp_path) -> None: | |
| source = tmp_path / "outside.mp3" | |
| source.write_bytes(b"ID3-unmanaged-media") | |
| container = build_container(settings) | |
| with pytest.raises(InputError, match="TEMP_DIR or OUTPUT_DIR"): | |
| await container.resolver.resolve_payload({"temp_path": str(source)}, str(uuid4())) | |