Spaces:
Running
Running
File size: 3,625 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 | 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()))
|