tests/test_platform.py
Rama master · réplica de lectura del repositorio original.
import os
import tempfile
import threading
import time
import unittest
from dataclasses import replace
from pathlib import Path
from unittest.mock import patch
from peach import settings_file
from peach.media import FilesystemBackend, MediaOffline, MediaUnavailable
from peach.platform import (
MOUNTS_ENV,
UNMAPPED_ROOT,
is_unmapped,
is_windows_path,
location_mounts,
location_roots,
resolve_location,
resolve_root,
root_online,
root_status,
system_volume,
translate_ledger_path,
translate_roots,
within_root,
)
from peach.repository import MediaAsset
POSIX_ONLY = unittest.skipIf(os.name == "nt", "路径翻译只在非 Windows 上发生")
#: 测试显式声明数据库路径前缀，不依赖本机 config.toml。
LOCATIONS = {"local": (r"R:\media",), "115": ("B:/",), "pikpak": ("A:/",)}
def _roots(table):
"""测试可传单个字符串；设置层使用与数据库路径前缀按序对应的元组。"""
return {key: ((value,) if value else ()) if isinstance(value, str) else tuple(value)
for key, value in table.items()}
def with_media(mounts, locations=None):
"""把设置层的 `[media.locations]` / `[media.mounts]` 钉死成给定值。"""
config = settings_file.load_config(environ={}, strict=False)
fixed = replace(
config, mounts=_roots(mounts),
locations=_roots(LOCATIONS if locations is None else locations),
)
return patch.object(settings_file, "active", lambda: fixed)
class LocationResolutionTests(unittest.TestCase):
"""按数据库路径前缀判定来源，两个平台的行为一致。"""
def test_windows_path_shape_is_recognised(self):
self.assertTrue(is_windows_path(r"R:\Media\one.mp4"))
self.assertTrue(is_windows_path("B:/one.mp4"))
self.assertFalse(is_windows_path("/Volumes/Media/media/one.mp4"))
self.assertFalse(is_windows_path("relative/one.mp4"))
def test_path_under_a_declared_root_resolves_to_that_location(self):
with with_media({}):
self.assertEqual(
resolve_location(r"R:\media\创作者\one.mp4"),
("local", ("创作者", "one.mp4")),
)
self.assertEqual(resolve_location("B:/x/one.mp4"), ("115", ("x", "one.mp4")))
def test_declared_root_matching_ignores_case(self):
"""数据库中 `R:\\Media` 与配置中的 `R:\\media` 对应同一目录，匹配时忽略大小写。"""
with with_media({}):
self.assertEqual(resolve_location(r"R:\MEDIA\one.mp4")[0], "local")
def test_the_longest_declared_root_wins(self):
with with_media({}, {"whole": "R:/", "local": r"R:\media"}):
self.assertEqual(resolve_location(r"R:\media\a.mp4"), ("local", ("a.mp4",)))
self.assertEqual(
resolve_location(r"R:\Resources\a.mp4"), ("whole", ("Resources", "a.mp4")))
def test_path_outside_every_declared_root_has_no_location(self):
with with_media({}):
self.assertEqual(resolve_location(r"Z:\secret\one.mp4"), (None, ()))
# 同一盘符下仍按目录区分：`R:\Resources` 不在 `R:\media` 中。
self.assertIsNone(resolve_location(r"R:\Resources\Intake\one.mp4")[0])
def test_declared_roots_are_read_from_the_settings_layer(self):
with with_media({}):
self.assertEqual(location_roots().get("local"), (r"R:\media",))
self.assertNotIn("online", location_roots())
def test_a_source_with_two_roots_resolves_each_to_its_own_index(self):
roots = {"local": (r"R:\media", r"S:\more"), "115": ("B:/",)}
self.assertEqual(resolve_root(r"R:\media\a.mp4", roots), ("local", 0, ("a.mp4",)))
self.assertEqual(resolve_root(r"S:\more\x\b.mp4", roots), ("local", 1, ("x", "b.mp4")))
self.assertEqual(resolve_root("B:/c.mp4", roots), ("115", 0, ("c.mp4",)))
self.assertEqual(resolve_root(r"Z:\nope", roots), (None, -1, ()))
@POSIX_ONLY
def test_each_root_translates_to_the_mount_at_the_same_position(self):
with with_media({"local": ["/Volumes/A", "/Volumes/B"]},
{"local": [r"R:\media", r"R:\media2"]}):
self.assertEqual(translate_ledger_path(r"R:\media\one.mp4"), Path("/Volumes/A/one.mp4"))
self.assertEqual(translate_ledger_path(r"R:\media2\two.mp4"), Path("/Volumes/B/two.mp4"))
# 部分前缀没有对应的本机目录时按无法访问处理，不转换到其他目录。
with with_media({"local": ["/Volumes/A"]}, {"local": [r"R:\media", r"R:\media2"]}):
self.assertTrue(is_unmapped(translate_ledger_path(r"R:\media2\two.mp4")))
class MountSourceTests(unittest.TestCase):
def test_mixed_mount_mapping_keeps_each_roots_position(self):
declared = {'local': ('R:\\one', 'S:\\two')}
with tempfile.TemporaryDirectory() as temporary:
mounted = Path(temporary).resolve()
with with_media({'local': ('', str(mounted))}, declared), patch.dict(os.environ, {MOUNTS_ENV: ''}):
self.assertEqual(location_mounts()['local'][1], mounted)
first = translate_ledger_path('R:\\one\\a.mp4')
if os.name == 'nt':
self.assertEqual(first, Path('R:\\one\\a.mp4'))
else:
self.assertTrue(is_unmapped(first))
self.assertEqual(translate_ledger_path('S:\\two\\b.mp4'), mounted / 'b.mp4')
"""挂载表按来源 ID 取自设置文件；`PEACH_MEDIA_MOUNTS` 仍然压过它。"""
def test_empty_source_uses_the_settings_parser_shape(self):
with with_media({"local": "/mnt/res", "115": "", "pikpak": ""}):
with patch.dict("os.environ", {MOUNTS_ENV: ""}):
self.assertEqual(settings_file.active().mounts["115"], ())
self.assertEqual(location_mounts(), {"local": (Path("/mnt/res"),)})
def test_settings_file_supplies_the_base_mapping(self):
with with_media({"local": "/mnt/res"}):
with patch.dict("os.environ", {MOUNTS_ENV: ""}):
self.assertEqual(location_mounts(), {"local": (Path("/mnt/res"),)})
def test_environment_still_wins_over_the_settings_file(self):
with with_media({"local": "/mnt/res", "115": "/mnt/115"}):
with patch.dict("os.environ", {MOUNTS_ENV: "local=/mnt/other"}):
self.assertEqual(
location_mounts(),
{"local": (Path("/mnt/other"),), "115": (Path("/mnt/115"),)})
def test_a_fresh_machine_has_no_mounts_at_all(self):
"""内建默认为空：没挂的来源按脱盘处理，绝不猜一个本机路径。"""
with with_media({}):
with patch.dict("os.environ", {MOUNTS_ENV: ""}):
self.assertEqual(location_mounts(), {})
class TranslationTests(unittest.TestCase):
@POSIX_ONLY
def test_declared_root_is_rebased_onto_the_local_mount(self):
with with_media({"local": "/mnt/res/media", "115": "/mnt/115"}):
with patch.dict("os.environ", {MOUNTS_ENV: ""}):
self.assertEqual(
translate_ledger_path(r"R:\media\创作者\one.mp4"),
Path("/mnt/res/media/创作者/one.mp4"),
)
self.assertEqual(
translate_ledger_path("B:/x/one.mp4"), Path("/mnt/115/x/one.mp4"))
@POSIX_ONLY
def test_unmapped_source_never_becomes_a_relative_path(self):
"""没有本机挂载目录的来源转换到 UNMAPPED_ROOT，避免读取当前目录下的同名文件。"""
with with_media({"local": "/mnt/res/media"}):
with patch.dict("os.environ", {MOUNTS_ENV: ""}):
path = translate_ledger_path(r"Z:\secret\one.mp4")
self.assertTrue(path.is_absolute())
self.assertTrue(is_unmapped(path))
self.assertTrue(str(path).startswith(str(UNMAPPED_ROOT)))
@POSIX_ONLY
def test_declared_but_unmounted_source_is_unmapped_too(self):
"""已配置路径前缀但没有本机挂载目录时，仍标为无法访问，不返回相对路径。"""
with with_media({}):
with patch.dict("os.environ", {MOUNTS_ENV: ""}):
self.assertTrue(is_unmapped(translate_ledger_path(r"R:\media\one.mp4")))
@POSIX_ONLY
def test_unmapped_roots_are_dropped_from_authorisation(self):
with with_media({"local": "/mnt/res/media", "pikpak": "", "115": ""}):
with patch.dict("os.environ", {MOUNTS_ENV: ""}):
roots = translate_roots((r"R:\media", "B:/", "A:/"))
self.assertEqual(roots, (Path("/mnt/res/media"),))
def test_posix_path_passes_through_untouched(self):
self.assertEqual(translate_ledger_path("/tmp/one.mp4"), Path("/tmp/one.mp4"))
def test_system_volume_is_the_disk_gate_target(self):
volume = system_volume()
self.assertTrue(volume.is_absolute())
self.assertTrue(volume.is_dir())
self.assertEqual(volume, Path(volume.anchor))
class SystemVolumeTests(unittest.TestCase):
@unittest.skipUnless(os.name == 'nt', 'Windows 系统目录 API')
def test_system_directory_can_be_on_another_drive(self):
import ctypes
def windows_directory(buffer, size):
buffer.value = 'Z:\\Windows'
return len(buffer.value)
with patch.object(ctypes.windll.kernel32, 'GetWindowsDirectoryW', side_effect=windows_directory):
self.assertEqual(system_volume(), Path('Z:/'))
@unittest.skipUnless(os.name == 'nt', 'Windows 系统目录 API')
def test_missing_system_directory_is_reported(self):
import ctypes
with patch.object(ctypes.windll.kernel32, 'GetWindowsDirectoryW', return_value=0):
with self.assertRaises(OSError):
system_volume()
class RootBoundaryTests(unittest.TestCase):
def test_within_root_accepts_exact_and_nested(self):
root = Path("/mnt/res/media")
self.assertTrue(within_root(root, root))
self.assertTrue(within_root(root / "a" / "b.mp4", root))
def test_within_root_rejects_sibling_and_shorter_paths(self):
root = Path("/mnt/res/media")
self.assertFalse(within_root(Path("/mnt/res/other/b.mp4"), root))
self.assertFalse(within_root(Path("/mnt"), root))
def test_within_root_survives_case_only_differences_on_insensitive_mounts(self):
"""账本写 `R:\\Media`，授权根是 `.../media`；大小写不敏感的挂载必须仍然放行。"""
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp).resolve() / "media"
root.mkdir()
shouting = Path(tmp).resolve() / "MEDIA" / "one.mp4"
insensitive = (Path(tmp).resolve() / "MEDIA").is_dir()
self.assertEqual(within_root(shouting, root), insensitive)
@unittest.skipUnless(os.name == "nt", "盘符只在 Windows 上有意义")
def test_windows_roots_answer_from_the_path_text_without_touching_the_drives(self):
"""网盘盘符每次 stat 都是一趟网络往返：大小写不同和跨盘符都不该去问盘。"""
with patch.object(Path, "samefile", side_effect=AssertionError("不该 stat 盘符")):
self.assertTrue(within_root(Path("B:/Media/a.mp4"), Path("b:/media")))
self.assertFalse(within_root(Path("A:/media/a.mp4"), Path("B:/")))
self.assertFalse(within_root(Path("B:/other/a.mp4"), Path("C:/media")))
def test_root_online_needs_a_readable_mount_not_just_a_directory(self):
with tempfile.TemporaryDirectory() as tmp:
self.assertTrue(root_online(Path(tmp)))
self.assertFalse(root_online(Path(tmp)))
self.assertFalse(root_online(UNMAPPED_ROOT))
class OfflineModeTests(unittest.TestCase):
"""脱盘是来源级判定：一个来源不在，不影响其他来源上的资产。"""
def _backend(self, tmp: Path) -> FilesystemBackend:
return FilesystemBackend([tmp], tmp / "snapshots")
@POSIX_ONLY
def test_asset_on_an_unmapped_source_reports_offline(self):
with tempfile.TemporaryDirectory() as tmp:
backend = self._backend(Path(tmp).resolve())
with with_media({"local": "/mnt/res/media"}):
with self.assertRaises(MediaOffline) as caught:
backend.file_for(MediaAsset(1, r"Z:\video\one.mp4", None))
self.assertEqual(caught.exception.source, "Z:")
self.assertEqual(caught.exception.asset_id, 1)
@POSIX_ONLY
def test_asset_on_a_detached_but_mapped_source_reports_offline(self):
with tempfile.TemporaryDirectory() as tmp:
detached = Path(tmp).resolve() / "gone"
backend = FilesystemBackend([detached], Path(tmp).resolve() / "snapshots")
with with_media({"local": str(detached)}):
with self.assertRaises(MediaOffline):
backend.file_for(MediaAsset(2, r"R:\media\video\one.mp4", None))
@POSIX_ONLY
def test_missing_file_on_a_live_source_is_not_offline(self):
"""单个文件丢失和整个来源不在必须分开报，否则前端会误显示脱盘模式。"""
with tempfile.TemporaryDirectory() as tmp:
live = Path(tmp).resolve()
backend = FilesystemBackend([live], live / "snapshots")
with with_media({"local": str(live)}):
with self.assertRaises(MediaUnavailable) as caught:
backend.file_for(MediaAsset(3, r"R:\media\nope.mp4", None))
self.assertNotIsInstance(caught.exception, MediaOffline)
def test_source_status_reports_each_root_separately(self):
with tempfile.TemporaryDirectory() as tmp:
live = Path(tmp).resolve()
dead = live / "gone"
status = FilesystemBackend([live, dead], live / "snapshots").source_status()
self.assertEqual([row["online"] for row in status], [True, False])
class MountReachabilityTests(unittest.TestCase):
def test_an_offline_root_does_not_disable_other_roots_of_the_same_source(self):
from peach.mount_reachability import MountReachability, MountRoot
monitor = MountReachability([
MountRoot('local', '本地磁盘', 'D:/JAV', self.root, 'root-one', 'library-one'),
MountRoot('local', '本地磁盘', 'E:/JAV', None, 'root-two', 'library-one'),
], probe=lambda _path: 'ok', retry_delay=0)
self.addCleanup(monitor.stop)
monitor._states[self.root].state = 'ok'
row = monitor.sources()[0]
self.assertTrue(row['online'])
self.assertEqual([(root['root_id'], root['online']) for root in row['roots']],
[('root-one', True), ('root-two', False)])
self.assertEqual(monitor.summary()['state'], 'degraded')
def setUp(self):
from peach.mount_reachability import MountReachability, MountRoot
self.tmp = tempfile.TemporaryDirectory()
self.addCleanup(self.tmp.cleanup)
self.root = Path(self.tmp.name).resolve()
self.now = 0.0
self.calls = []
self.answer = "ok"
def probe(path):
self.calls.append(path)
return self.answer
self.monitor = MountReachability([MountRoot("local", "本地磁盘", "R:/media", self.root)],
probe=probe, clock=lambda: self.now, retry_delay=0)
self.addCleanup(self.monitor.stop)
def wait_state(self, expected):
deadline = time.monotonic() + 2
while time.monotonic() < deadline:
self.monitor.tick()
if self.monitor.summary()["sources"][0]["state"] == expected:
return
time.sleep(0.005)
self.fail(f"没有取得状态 {expected}")
def test_directory_probe_distinguishes_empty_missing_file_and_permission(self):
self.assertEqual(root_status(self.root), "ok")
self.assertEqual(root_status(self.root / "missing"), "missing")
file = self.root / "file"
file.touch()
self.assertEqual(root_status(file), "not_directory")
for error, expected in ((PermissionError(), "permission_denied"), (OSError(), "unavailable")):
with patch("peach.platform.os.scandir", side_effect=error):
self.assertEqual(root_status(self.root), expected)
def test_disconnected_network_mount_is_unreachable_not_missing_or_denied(self):
# Windows 把网络路径断开（53/67）映射成 FileNotFoundError，设备未就绪（21）映射成 PermissionError。
for error_type, winerror in ((FileNotFoundError, 53), (FileNotFoundError, 67), (PermissionError, 21)):
error = error_type(2, "network")
error.winerror = winerror
with self.subTest(winerror=winerror), patch("peach.platform.os.scandir", side_effect=error):
self.assertEqual(root_status(self.root), "unavailable")
def test_failure_while_reading_the_first_entry_is_not_online(self):
class Entries:
def __enter__(self): return self
def __exit__(self, *args): pass
def __iter__(self): return self
def __next__(self): raise PermissionError
with patch("peach.platform.os.scandir", return_value=Entries()):
self.assertEqual(root_status(self.root), "permission_denied")
self.assertFalse(root_online(self.root))
def test_healthy_and_failed_probes_use_different_intervals_and_recover(self):
self.wait_state("ok")
self.assertEqual(len(self.calls), 1)
self.now = 599
self.monitor.tick()
self.assertEqual(len(self.calls), 1)
self.answer = "permission_denied"
self.now = 600
self.wait_state("permission_denied")
self.assertEqual(len(self.calls), 3)
self.assertEqual(self.monitor.summary()["warnings"], ["本地磁盘：没有权限读取"])
self.now = 659
self.monitor.tick()
self.assertEqual(len(self.calls), 3)
self.answer = "ok"
self.now = 660
self.wait_state("ok")
self.assertEqual(len(self.calls), 4)
self.assertEqual(self.monitor.summary()["warnings"], [])
def test_a_successful_retry_does_not_report_a_failure(self):
answers = iter(["missing", "ok"])
self.monitor.probe = lambda path: next(answers)
self.wait_state("ok")
self.assertEqual(self.monitor.summary()["warnings"], [])
def test_first_probe_does_not_depend_on_the_clock_origin(self):
self.now = -10000
self.wait_state("ok")
self.assertEqual(len(self.calls), 1)
def test_pending_probe_is_unknown_and_timeout_keeps_one_worker(self):
entered, release = threading.Event(), threading.Event()
self.addCleanup(release.set)
def blocked(path):
self.calls.append(path)
entered.set()
release.wait(2)
return "ok"
self.monitor.probe = blocked
self.monitor.tick()
self.assertTrue(entered.wait(1))
self.assertIsNone(self.monitor.sources()[0]["online"])
self.now = 6
self.monitor.tick()
self.assertEqual(self.monitor.summary()["sources"][0]["state"], "timeout")
for self.now in (60, 600, 6000):
self.monitor.tick()
self.assertEqual(len(self.calls), 1)
self.assertEqual(self.monitor.summary()["warnings"], [])
release.set()
self.wait_state("ok")
def test_slow_roots_have_a_global_limit_and_stop_does_not_wait_for_them(self):
from peach.mount_reachability import MountReachability, MountRoot, MAX_PENDING
release = threading.Event()
self.addCleanup(release.set)
def blocked(path):
self.calls.append(path)
release.wait(2)
return "ok"
self.monitor = MountReachability([
MountRoot("local", "本地磁盘", str(at), self.root / str(at)) for at in range(12)
], probe=blocked)
self.addCleanup(self.monitor.stop)
self.monitor.tick()
self.monitor.tick()
deadline = time.monotonic() + 1
while len(self.calls) < MAX_PENDING and time.monotonic() < deadline:
time.sleep(0.005)
self.assertEqual(len(self.calls), MAX_PENDING)
self.monitor.stop()
self.assertFalse(release.is_set())
self.monitor.tick()
self.assertEqual(len(self.calls), MAX_PENDING)
def test_summary_hides_paths_and_all_roots_contribute_to_source_health(self):
from peach.mount_reachability import MountReachability, MountRoot
self.monitor = MountReachability([
MountRoot("local", "本地磁盘", "R:/private", self.root),
MountRoot("local", "本地磁盘", "S:/private", None),
])
self.assertFalse(self.monitor.sources()[0]["online"])
row = self.monitor.summary()["sources"][0]
self.assertEqual(row["state"], "unmapped")
self.assertNotIn("roots", row)
self.assertNotIn("declared", row)
self.assertNotIn("resolved", row)
self.assertEqual(row["message"], "本地磁盘：未配置挂载点")
if __name__ == "__main__":
unittest.main()