Spaces:
Running
Running
File size: 4,966 Bytes
7cc81cb 8a2b510 | 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 | 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()))
async def test_resolves_top_level_n8n_binary_object(settings) -> None:
container = build_container(settings)
encoded = base64.b64encode(b"binary-object").decode()
request = json_request(
('{"data":"%s","fileName":"clip.mp4","mimeType":"video/mp4"}' % encoded).encode()
)
resolved = await container.resolver.resolve(request)
assert resolved.primary.source is MediaSource.N8N_BINARY
assert resolved.primary.filename == "clip.mp4"
assert resolved.primary.mime_type == "video/mp4"
assert resolved.primary.temp_path.read_bytes() == b"binary-object"
async def test_resolves_top_level_url_string(settings) -> None:
container = build_container(settings)
request = json_request(b'"https://example.com/video.mp4"')
async def fake_validate(url):
return None
async def fake_download(url, destination, filename=None):
path = destination / "downloaded.mp4"
path.write_bytes(b"downloaded")
return path, "video/mp4"
container.downloader.validate_url = fake_validate
container.downloader.download = fake_download
container.ytdlp.supports_url = lambda url: False
resolved = await container.resolver.resolve(request)
assert resolved.primary.source is MediaSource.JSON_URL
assert resolved.primary.temp_path.read_bytes() == b"downloaded"
|