tests/test_media.py
Rama master · réplica de lectura del repositorio original.
import os
import subprocess
import tempfile
import sys
import unittest
from pathlib import Path
from unittest.mock import patch
import peach.media as media_module
from peach.ffmpeg import BinaryChoice, FFmpegResolver, binary_diagnostics
from peach.media import (
FilesystemBackend, MediaEngine, MediaUnavailable, normalized_path,
resolve_case_insensitive,
)
from peach.repository import MediaAsset
class MediaEngineTests(unittest.TestCase):
def test_tool_diagnostics_distinguishes_software_av1_and_hardware_only(self):
for decoder, available in (('av1', False), ('libdav1d', True)):
results = [subprocess.CompletedProcess([], 0, b'ffmpeg version test\n', b''),
subprocess.CompletedProcess([], 0, f' V..... {decoder} AV1 decoder\n'.encode(), b'')]
with patch('peach.ffmpeg.subprocess.run', side_effect=results):
result = binary_diagnostics(BinaryChoice(Path('ffmpeg'), 'test'), decoders=True)
self.assertTrue(result['usable'])
self.assertEqual(result['av1_software'], available)
def test_tool_diagnostics_probe_each_binary_file_once_until_it_changes(self):
with tempfile.TemporaryDirectory() as tmp:
binary = Path(tmp).resolve() / 'ffmpeg.exe'
binary.write_bytes(b'one')
choice = BinaryChoice(binary, 'test')
version = subprocess.CompletedProcess([], 0, b'ffmpeg version test\n', b'')
with patch('peach.ffmpeg.subprocess.run', return_value=version) as run:
self.assertTrue(binary_diagnostics(choice)['usable'])
self.assertTrue(binary_diagnostics(choice)['usable'])
self.assertEqual(run.call_count, 1)
binary.write_bytes(b'upgraded')
self.assertTrue(binary_diagnostics(choice)['usable'])
self.assertEqual(run.call_count, 2)
with patch('peach.ffmpeg.subprocess.run', side_effect=subprocess.TimeoutExpired('ffmpeg', 5)) as run:
self.assertFalse(binary_diagnostics(BinaryChoice(binary, 'other'))['usable'])
self.assertFalse(binary_diagnostics(BinaryChoice(binary, 'other'))['usable'])
self.assertEqual(run.call_count, 2)
def test_tool_diagnostics_reports_execution_failure(self):
with patch('peach.ffmpeg.subprocess.run', side_effect=subprocess.TimeoutExpired('ffmpeg', 5)):
result = binary_diagnostics(BinaryChoice(Path('ffmpeg'), 'test'), decoders=True)
self.assertTrue(result['available'])
self.assertFalse(result['usable'])
self.assertEqual(result['error'], 'TimeoutExpired')
# 临时目录一律先 resolve 再喂给被测代码。媒体解析本身就以 resolve 后的真实路径
# 判断是否越过根目录，而 CI runner 的临时目录都是别名：macOS 的 /var 是
# /private/var 的软链，Windows runner 的 %TEMP% 是 C:\Users\RUNNER~1 这种 8.3
# 短名，展开后是 runneradmin。拿未 resolve 的路径做断言，本机全绿、CI 全红。
def test_filesystem_backend_enforces_root(self):
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp).resolve()
inside = root / "media" / "one.mp4"
outside = root.parent / "outside.mp4"
inside.parent.mkdir()
inside.write_bytes(b"test")
backend = FilesystemBackend([root], root / "snapshots")
self.assertEqual(backend.file_for(MediaAsset(1, str(inside), None)), inside)
with self.assertRaises(MediaUnavailable):
backend.file_for(MediaAsset(2, str(outside), None))
def test_engine_resolves_through_the_filesystem_only(self):
"""媒体解析没有第二条来源。
2026-09-01 关掉 Stash adapter 之前，这里还有一层 backend 契约。它在生产里
一次都没有生效过：`media_binding` 表是空的，`external_id("stash")` 永远返回
None，而 `stream_candidates` 本身没有任何调用方。见 ADR-0021。
"""
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp).resolve()
media = root / "one.mp4"
media.write_bytes(b"test")
repository = unittest.mock.Mock()
repository.media_asset.return_value = MediaAsset(1, str(media), None)
engine = MediaEngine(repository, FilesystemBackend([root], root / "snapshots"))
self.assertEqual(engine.file_for(1), media)
def test_engine_raises_for_unknown_asset(self):
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
repository = unittest.mock.Mock()
repository.media_asset.return_value = None
engine = MediaEngine(
repository,
FilesystemBackend([root], root / "snapshots"),
)
with self.assertRaises(Exception) as raised:
engine.file_for(404)
self.assertEqual(type(raised.exception).__name__, "MediaNotFound")
def test_ffmpeg_resolver_prefers_explicit_environment(self):
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
binary = root / "ffmpeg.exe"
binary.write_bytes(b"test")
resolver = FFmpegResolver(root)
with patch.dict("os.environ", {"PEACH_FFMPEG": str(binary)}, clear=False):
choice = resolver.ffmpeg()
self.assertEqual(choice.source, "environment")
def test_ffmpeg_resolver_uses_managed_bundle(self):
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
binary = root / "bin" / ("ffmpeg.exe" if os.name == "nt" else "ffmpeg")
binary.parent.mkdir()
binary.write_bytes(b"test")
with patch.dict("os.environ", {"PEACH_FFMPEG": ""}, clear=False), \
patch("peach.ffmpeg.shutil.which", return_value=None):
choice = FFmpegResolver(root).ffmpeg()
self.assertEqual(choice.source, "peach-managed")
def test_frozen_application_uses_both_bundled_tools_without_path(self):
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp).resolve()
bundle = root / "app" / "tools" / "ffmpeg" / "bin"
bundle.mkdir(parents=True)
suffix = ".exe" if os.name == "nt" else ""
for name in ("ffmpeg", "ffprobe"):
(bundle / (name + suffix)).write_bytes(b"tool")
with patch.object(sys, "frozen", True, create=True), \
patch.object(sys, "executable", str(root / "app" / "Peach.exe")), \
patch.dict(os.environ, {"PEACH_FFMPEG": "", "PEACH_FFPROBE": ""}), \
patch("peach.ffmpeg.shutil.which", return_value=None):
resolver = FFmpegResolver(root / "data")
for choice in (resolver.ffmpeg(), resolver.ffprobe()):
self.assertEqual(choice.source, "peach-bundled")
self.assertEqual(choice.path.parent, bundle)
def test_source_environment_does_not_use_tools_beside_python(self):
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp).resolve()
tool = root / "tools" / "ffmpeg" / "bin" / ("ffmpeg.exe" if os.name == "nt" else "ffmpeg")
tool.parent.mkdir(parents=True)
tool.write_bytes(b"tool")
with patch.object(sys, "frozen", False, create=True), \
patch.object(sys, "executable", str(root / "python.exe")), \
patch.dict(os.environ, {"PEACH_FFMPEG": ""}), \
patch("peach.ffmpeg.shutil.which", return_value=None):
self.assertIsNone(FFmpegResolver(root / "data").ffmpeg())
def test_offline_windows_root_does_not_block_startup(self):
with patch.object(Path, "resolve", side_effect=OSError("offline")):
path = normalized_path(Path("B:/"))
self.assertTrue(path.is_absolute())
@unittest.skipUnless(os.name == "nt", "只有 Windows 的盘符路径跳过 resolve")
def test_windows_ledger_paths_are_not_resolved_through_the_redirector(self):
"""PikPak 的 A: 是 WinFsp 网络驱动器：`resolve()` 一条文件路径要 7 秒，还换成 UNC 形态，
之后每次 stat 再花 7 到 14 秒；盘符路径本身就是绝对路径，只需要折掉 `..`。"""
with patch.object(Path, "resolve", side_effect=AssertionError("resolve() 走了网络重定向器")):
path = normalized_path(r"A:\番号\..\番号\clip.mp4")
self.assertEqual(path, Path(r"A:\番号\clip.mp4"))
def test_filesystem_backend_matches_case_insensitively_on_sensitive_mounts(self):
"""CloudDrive 大小写敏感：账本 `abw-118.mp4` 对磁盘 `ABW-118.mp4` 必须救回。"""
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
real = root / "ABW-118.MP4"
real.write_bytes(b"test")
backend = FilesystemBackend([root], root / "snapshots")
opened = backend.file_for(MediaAsset(1, str(root / "abw-118.mp4"), None))
self.assertTrue(opened.is_file())
self.assertEqual(opened.name.casefold(), "abw-118.mp4")
# 完全缺失的名字仍应拒绝
with self.assertRaises(MediaUnavailable):
backend.file_for(MediaAsset(2, str(root / "nope.mp4"), None))
def test_batch_path_resolution_translates_ledger_drive_before_case_matching(self):
"""sheets/probe 不经过 FilesystemBackend，也必须先翻译账本盘符。"""
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
real = root / "ABW-118.MP4"
real.write_bytes(b"test")
with patch(
"peach.media.translate_ledger_path",
return_value=root / "abw-118.mp4",
):
opened = Path(resolve_case_insensitive(r"A:\\ABW-118.mp4"))
self.assertTrue(opened.is_file())
self.assertTrue(opened.parent.samefile(root))
def test_case_insensitive_miss_expires_after_the_short_ttl(self):
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
with patch("peach.media.monotonic", return_value=0):
self.assertIsNone(
media_module._case_insensitive_match(str(root), "abw-118.mp4")
)
real = root / "ABW-118.MP4"
real.write_bytes(b"test")
with patch("peach.media.monotonic", return_value=6):
self.assertEqual(
media_module._case_insensitive_match(str(root), "abw-118.mp4"),
real.name,
)
def test_remote_mp4_defaults_to_range_and_only_segments_on_request(self):
"""默认 HLS 会让 HEVC 静默黑屏：`-c copy` 把 HEVC 装进 MPEG-TS，而 Chromium 的
MSE 不支持 TS 里的 HEVC（实测 `video/mp2t; codecs="hvc1…"` 为 false），数据进得了
缓冲却一帧都解不出，也没有 error 事件。同一浏览器直接 Range 播同一文件能出帧。
见 ADR-0016。
"""
repository = unittest.mock.Mock()
repository.media_asset.return_value = MediaAsset(
1, r"B:\video\one.mp4", None, "115", "one.mp4", 31.5, 100,
)
engine = MediaEngine(
repository, FilesystemBackend([Path("B:/")], Path("snapshots")),
)
self.assertEqual(engine.stream_plan(1).protocol, "range")
self.assertEqual(engine.stream_plan(1, mode="auto").protocol, "range")
plan = engine.stream_plan(1, mode="hls")
self.assertEqual(plan.protocol, "hls")
self.assertEqual(plan.segment_seconds, 6)
def test_hls_stays_unavailable_where_it_never_applied(self):
"""按需模式不是万能开关：本地来源和时长未知的片源仍然只能走 Range。"""
repository = unittest.mock.Mock()
repository.media_asset.return_value = MediaAsset(
1, r"R:\video\one.mp4", None, "local", "one.mp4", 31.5, 100,
)
engine = MediaEngine(repository, FilesystemBackend([Path("R:/")], Path("snapshots")))
self.assertEqual(engine.stream_plan(1, mode="hls").protocol, "range")
repository.media_asset.return_value = MediaAsset(
1, r"B:\video\one.mp4", None, "115", "one.mp4", None, 100,
)
engine = MediaEngine(repository, FilesystemBackend([Path("B:/")], Path("snapshots")))
self.assertEqual(engine.stream_plan(1, mode="hls").protocol, "range")
def test_local_or_unknown_duration_keeps_range_plan(self):
repository = unittest.mock.Mock()
repository.media_asset.return_value = MediaAsset(
1, r"R:\video\one.mp4", None, "local", "one.mp4", None, 100,
)
engine = MediaEngine(
repository, FilesystemBackend([Path("R:/")], Path("snapshots")),
)
self.assertEqual(engine.stream_plan(1).protocol, "range")
if __name__ == "__main__":
unittest.main()