102 lines
3.6 KiB
Python
102 lines
3.6 KiB
Python
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from app.providers.base import DownloadOptions, MediaMetadata
|
|
from app.providers.mediathek import MediathekProvider, is_stream_manifest
|
|
from app.providers.youtube import parse_yt_dlp_progress
|
|
from app.services.downloader import copy_tree_contents
|
|
|
|
|
|
class FakeStream:
|
|
def __init__(self, lines: list[bytes]):
|
|
self.lines = lines
|
|
|
|
async def readline(self) -> bytes:
|
|
if self.lines:
|
|
return self.lines.pop(0)
|
|
return b""
|
|
|
|
|
|
class FakeProcess:
|
|
def __init__(self, output_path: Path):
|
|
self.returncode = 0
|
|
self.stdout = FakeStream([b"[download] 100% of 12.00MiB\n"])
|
|
self.stderr = FakeStream([])
|
|
self.output_path = output_path
|
|
|
|
async def wait(self) -> int:
|
|
self.output_path.write_bytes(b"real media bytes")
|
|
return self.returncode
|
|
|
|
|
|
def test_copy_tree_contents_copies_files(tmp_path: Path):
|
|
src = tmp_path / 'src'
|
|
src.mkdir()
|
|
(src / 'movie.mp4').write_bytes(b'data')
|
|
(src / 'meta.info.json').write_text('{}')
|
|
dest = tmp_path / 'dest'
|
|
|
|
copied = copy_tree_contents(src, dest)
|
|
|
|
assert sorted(p.name for p in copied) == ['meta.info.json', 'movie.mp4']
|
|
assert (dest / 'movie.mp4').read_bytes() == b'data'
|
|
|
|
|
|
def test_copy_tree_contents_rejects_top_level_symlink(tmp_path: Path):
|
|
src = tmp_path / "src"
|
|
src.mkdir()
|
|
(tmp_path / "secret.txt").write_text("secret")
|
|
(src / "leak.txt").symlink_to(tmp_path / "secret.txt")
|
|
|
|
with pytest.raises(Exception, match="symlink"):
|
|
copy_tree_contents(src, tmp_path / "dest")
|
|
|
|
|
|
def test_copy_tree_contents_rejects_nested_symlink(tmp_path: Path):
|
|
src = tmp_path / "src"
|
|
nested = src / "nested"
|
|
nested.mkdir(parents=True)
|
|
(tmp_path / "secret.txt").write_text("secret")
|
|
(nested / "leak.txt").symlink_to(tmp_path / "secret.txt")
|
|
|
|
with pytest.raises(Exception, match="symlink"):
|
|
copy_tree_contents(src, tmp_path / "dest")
|
|
|
|
|
|
def test_parse_yt_dlp_progress_lines():
|
|
assert parse_yt_dlp_progress('[download] 42.7% of 10.00MiB at 1.00MiB/s ETA 00:05') == pytest.approx(0.427)
|
|
assert parse_yt_dlp_progress('[download] 100% of 10.00MiB') == 1.0
|
|
assert parse_yt_dlp_progress('[info] unrelated') is None
|
|
|
|
|
|
def test_stream_manifest_detection():
|
|
assert is_stream_manifest('https://example.org/master.m3u8')
|
|
assert is_stream_manifest('https://example.org/manifest.mpd')
|
|
assert is_stream_manifest('https://example.org/video', 'application/vnd.apple.mpegurl')
|
|
assert not is_stream_manifest('https://example.org/video.mp4', 'video/mp4')
|
|
|
|
|
|
async def test_m3u8_download_uses_ytdlp_not_manifest_file(monkeypatch, tmp_path: Path):
|
|
provider = MediathekProvider()
|
|
manifest_url = 'https://cdn.example.org/master.m3u8'
|
|
|
|
async def fake_probe(url: str) -> MediaMetadata:
|
|
return MediaMetadata(provider='mediathek', title='master.m3u8', external_id=manifest_url)
|
|
|
|
async def fake_create_subprocess_exec(*args, **kwargs):
|
|
assert args[0].endswith('yt-dlp')
|
|
assert '--proxy' not in args
|
|
assert '--remux-video' in args
|
|
output_template = Path(args[args.index('--output') + 1])
|
|
return FakeProcess(output_template.parent / 'downloaded.mp4')
|
|
|
|
monkeypatch.setattr(provider, 'probe', fake_probe)
|
|
monkeypatch.setattr('app.providers.mediathek.shutil.which', lambda name: '/usr/bin/yt-dlp' if name == 'yt-dlp' else None)
|
|
monkeypatch.setattr('app.providers.mediathek.asyncio.create_subprocess_exec', fake_create_subprocess_exec)
|
|
|
|
result = await provider.download(manifest_url, tmp_path, DownloadOptions())
|
|
|
|
assert [p.name for p in result.output_files] == ['downloaded.mp4']
|
|
assert not (tmp_path / 'master.m3u8').exists()
|