154 lines
6.9 KiB
Python
154 lines
6.9 KiB
Python
import json
|
|
import subprocess
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
from catalog.app import Catalog, ProbeResult, cache_key, inspect_media, playlist_url
|
|
|
|
|
|
class CatalogTests(unittest.TestCase):
|
|
def test_playlist_url_encodes_each_path_segment(self) -> None:
|
|
self.assertEqual(
|
|
playlist_url(Path("中文 media") / "clip #1.mp4"),
|
|
"/hls/%E4%B8%AD%E6%96%87%20media/clip%20%231.mp4/master.m3u8",
|
|
)
|
|
|
|
def test_cache_key_changes_when_media_changes(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temporary_directory:
|
|
source = Path(temporary_directory) / "clip.mp4"
|
|
source.write_bytes(b"a")
|
|
first = cache_key(Path("clip.mp4"), source.stat())
|
|
source.write_bytes(b"updated")
|
|
second = cache_key(Path("clip.mp4"), source.stat())
|
|
self.assertNotEqual(first, second)
|
|
|
|
def test_refresh_publishes_validated_list_and_cached_cover(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temporary_directory:
|
|
root = Path(temporary_directory)
|
|
media_dir = root / "media"
|
|
cache_dir = root / "cache"
|
|
media_dir.mkdir()
|
|
source = media_dir / "clip.mp4"
|
|
source.write_bytes(b"video")
|
|
catalog = Catalog(media_dir, cache_dir, auto_refresh=False)
|
|
probe_payload = json.dumps(
|
|
{
|
|
"format": {"duration": "20"},
|
|
"streams": [
|
|
{"index": 0, "codec_type": "video", "codec_name": "h264", "disposition": {"attached_pic": 0}},
|
|
{"index": 1, "codec_type": "audio", "codec_name": "aac", "disposition": {"attached_pic": 0}},
|
|
],
|
|
}
|
|
)
|
|
|
|
def fake_run(command: list[str], **_kwargs: object) -> subprocess.CompletedProcess[str]:
|
|
if command[0] == "ffprobe":
|
|
return subprocess.CompletedProcess(command, 0, probe_payload, "")
|
|
Path(command[-1]).write_bytes(b"jpeg")
|
|
return subprocess.CompletedProcess(command, 0, "", "")
|
|
|
|
self.assertFalse(catalog.ready())
|
|
with patch("catalog.app.subprocess.run", side_effect=fake_run) as run:
|
|
catalog.refresh_now()
|
|
|
|
self.assertTrue(catalog.ready())
|
|
self.assertTrue((cache_dir / "current").is_symlink())
|
|
items = json.loads(catalog.media_payload())
|
|
self.assertEqual(len(items), 1)
|
|
self.assertEqual(items[0]["path"], "clip.mp4")
|
|
self.assertEqual(items[0]["playlistUrl"], "/hls/clip.mp4/master.m3u8")
|
|
cover_name = items[0]["coverUrl"].rsplit("/", 1)[-1]
|
|
self.assertEqual(catalog.cover_payload(cover_name), b"jpeg")
|
|
self.assertEqual(run.call_count, 2)
|
|
|
|
with patch("catalog.app.subprocess.run") as run:
|
|
self.assertEqual(json.loads(catalog.media_payload()), items)
|
|
self.assertEqual(catalog.cover_payload(cover_name), b"jpeg")
|
|
run.assert_not_called()
|
|
|
|
def test_refresh_excludes_invalid_media(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temporary_directory:
|
|
root = Path(temporary_directory)
|
|
media_dir = root / "media"
|
|
cache_dir = root / "cache"
|
|
media_dir.mkdir()
|
|
(media_dir / "valid.mp4").write_bytes(b"valid")
|
|
(media_dir / "broken.mp4").write_bytes(b"broken")
|
|
catalog = Catalog(media_dir, cache_dir, auto_refresh=False)
|
|
|
|
def fake_probe(source: Path) -> ProbeResult | None:
|
|
return ProbeResult(10.0, None) if source.name == "valid.mp4" else None
|
|
|
|
with patch("catalog.app.inspect_media", side_effect=fake_probe), patch.object(
|
|
Catalog, "_generate_cover", return_value=True
|
|
):
|
|
catalog.refresh_now()
|
|
|
|
items = json.loads(catalog.media_payload())
|
|
self.assertEqual([item["path"] for item in items], ["valid.mp4"])
|
|
status = catalog.status()
|
|
self.assertEqual(status["validFiles"], 1)
|
|
self.assertEqual(status["skippedFiles"], 1)
|
|
|
|
def test_failed_refresh_keeps_previous_snapshot(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temporary_directory:
|
|
root = Path(temporary_directory)
|
|
media_dir = root / "media"
|
|
cache_dir = root / "cache"
|
|
media_dir.mkdir()
|
|
(media_dir / "clip.mp4").write_bytes(b"video")
|
|
catalog = Catalog(media_dir, cache_dir, auto_refresh=False)
|
|
|
|
with patch("catalog.app.inspect_media", return_value=ProbeResult(10.0, None)), patch.object(
|
|
Catalog, "_generate_cover", return_value=True
|
|
):
|
|
catalog.refresh_now()
|
|
previous_payload = catalog.media_payload()
|
|
|
|
with patch.object(Catalog, "_build_snapshot", side_effect=RuntimeError("disk full")):
|
|
catalog.refresh_now()
|
|
|
|
self.assertTrue(catalog.ready())
|
|
self.assertEqual(catalog.media_payload(), previous_payload)
|
|
self.assertEqual(catalog.status()["state"], "ready")
|
|
self.assertEqual(catalog.status()["lastError"], "disk full")
|
|
|
|
def test_existing_snapshot_is_loaded_without_subprocesses(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temporary_directory:
|
|
root = Path(temporary_directory)
|
|
media_dir = root / "media"
|
|
cache_dir = root / "cache"
|
|
media_dir.mkdir()
|
|
(media_dir / "clip.mp4").write_bytes(b"video")
|
|
builder = Catalog(media_dir, cache_dir, auto_refresh=False)
|
|
with patch("catalog.app.inspect_media", return_value=ProbeResult(10.0, None)), patch.object(
|
|
Catalog, "_generate_cover", return_value=True
|
|
):
|
|
builder.refresh_now()
|
|
|
|
with patch("catalog.app.subprocess.run") as run:
|
|
catalog = Catalog(media_dir, cache_dir, auto_refresh=False)
|
|
payload = catalog.media_payload()
|
|
run.assert_not_called()
|
|
self.assertTrue(catalog.ready())
|
|
self.assertEqual(len(json.loads(payload)), 1)
|
|
|
|
def test_inspect_media_rejects_unsupported_streams(self) -> None:
|
|
payload = {
|
|
"format": {"duration": "12.5"},
|
|
"streams": [
|
|
{"index": 0, "codec_type": "video", "codec_name": "h264", "disposition": {"attached_pic": 0}},
|
|
{"index": 1, "codec_type": "audio", "codec_name": "aac", "disposition": {"attached_pic": 0}},
|
|
{"index": 2, "codec_type": "subtitle", "codec_name": "mov_text", "disposition": {"attached_pic": 0}},
|
|
],
|
|
}
|
|
completed = subprocess.CompletedProcess([], 0, json.dumps(payload), "")
|
|
with patch("catalog.app.subprocess.run", return_value=completed):
|
|
self.assertIsNone(inspect_media(Path("clip.mp4")))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|