import importlib import os import tempfile import unittest from pathlib import Path from unittest.mock import patch from peach import config, settings_file from peach.settings_file import SettingsFileError class TempTree: """`/peach-data` 加一个并列的 `/app` 当项目根。 临时目录一律先 `.resolve()`:CI runner 的临时路径都是别名(macOS `/var` 软链到 `/private/var`,Windows 短名 `RUNNER~1` 展开成 `runneradmin`),不 resolve 的路径 在开发机上过、在 CI 上红。 """ def __init__(self, with_data_root: bool = True): self._temp = tempfile.TemporaryDirectory() self.root = Path(self._temp.name).resolve() self.project = self.root / "app" self.project.mkdir() self.data_root = self.root / "peach-data" if with_data_root: self.data_root.mkdir() def write(self, text: str) -> Path: path = self.data_root / settings_file.SETTINGS_FILENAME path.parent.mkdir(parents=True, exist_ok=True) path.write_text(text, encoding="utf-8") return path def load(self, environ: dict[str, str] | None = None, **kwargs): return settings_file.load_config( project_root=self.project, environ=environ or {}, **kwargs) def __enter__(self): return self def __exit__(self, *exc): self._temp.cleanup() class ProjectRootTests(unittest.TestCase): def test_project_root_uses_pyinstaller_resource_directory_without_a_src_layer(self): # 锚点必须在当前平台上真的是绝对路径。`C:/repo` 在 POSIX 上是**相对**路径, # `_project_root` 里的 `.resolve()` 会给它前置 cwd,断言于是只在 Windows 成立。 root = Path(tempfile.gettempdir()).resolve() / "repo" source = root / "src" / "peach" / "settings_file.py" self.assertEqual(settings_file._project_root(str(source)), root) # bundle_root 不经过 resolve,原样构造,两个平台比较结果一致。 self.assertEqual( settings_file._project_root(str(source), "C:/bundle/_MEI123"), Path("C:/bundle/_MEI123"), ) class DataRootDiscoveryTests(unittest.TestCase): def test_saved_data_root_locations_reject_invalid_and_cyclic_paths(self): with TempTree() as tree: state = tree.data_root / "state" state.mkdir() locator = state / settings_file.DATA_ROOT_LOCATION for destination in ("", "relative", str(tree.data_root)): with self.subTest(destination=destination): locator.write_text(destination, encoding="utf-8") with self.assertRaises(SettingsFileError): tree.load() def test_a_missing_selected_root_stays_selected(self): with TempTree() as tree: selected = tree.root / "custom-data" state = tree.data_root / "state" state.mkdir() (state / settings_file.DATA_ROOT_LOCATION).write_text(str(selected), encoding="utf-8") loaded = tree.load() self.assertEqual(loaded.data_root, selected) self.assertFalse(loaded.present) def test_environment_variable_wins_over_the_sibling_directory(self): with TempTree() as tree: elsewhere = tree.root / "elsewhere" found = settings_file.discover_data_root( tree.project, {"PEACH_DATA_ROOT": str(elsewhere)}) self.assertEqual(found, (elsewhere, True)) def test_sibling_peach_data_is_found_from_the_project_root(self): with TempTree() as tree: self.assertEqual( settings_file.discover_data_root(tree.project, {}), (tree.data_root, True), ) def test_nested_checkout_still_finds_the_shared_data_root(self): """隔离工作树和打包产物都比项目根深一层以上,向上找必须仍然命中。""" with TempTree() as tree: nested = tree.root / "worktrees" / "task" nested.mkdir(parents=True) self.assertEqual( settings_file.discover_data_root(nested, {}), (tree.data_root, True)) def test_frozen_build_searches_from_the_executable_directory(self): """单文件包的 `_MEIPASS` 是临时目录,起点必须是 EXE 所在目录才找得到 Peach 数据目录。""" with TempTree() as tree: executable = tree.project / "dist" / "Peach" / "Peach.exe" executable.parent.mkdir(parents=True) executable.write_bytes(b"") anchor = settings_file._search_anchor( tree.root / "temp" / "_MEI123", frozen=True, executable=str(executable)) self.assertEqual(anchor, executable.parent) self.assertEqual( settings_file.discover_data_root(anchor, {}), (tree.data_root, True)) def test_source_tree_searches_from_the_project_root(self): with TempTree() as tree: anchor = settings_file._search_anchor( tree.project, frozen=False, executable=str(tree.root / "python.exe")) self.assertEqual(anchor, tree.project) def test_missing_data_root_is_an_explicit_unconfigured_state(self): with TempTree(with_data_root=False) as tree: path, found = settings_file.discover_data_root(tree.project, {}) self.assertFalse(found) # 找不到也要给出落点:`peach init` 按它建目录,页面按它显示提示。 self.assertEqual(path, tree.root / "peach-data") class DefaultsAreGenericTests(unittest.TestCase): """内建默认必须对一个全新用户成立,不能是某台机器的坐标。""" def test_fresh_machine_gets_neutral_defaults(self): with TempTree() as tree: loaded = tree.load() self.assertEqual(loaded.server.host, "127.0.0.1") self.assertEqual(loaded.server.port, 8900) self.assertEqual(loaded.server.mdns_name, "peach") self.assertEqual(loaded.server.review_writer_origin, "") self.assertEqual(loaded.server.review_writer_proxy, "") # 一台新机器只使用用户主动添加的目录。 self.assertEqual(loaded.locations, {}) self.assertEqual(loaded.mounts, {}) self.assertEqual(loaded.replication.smb_host, "") self.assertEqual(loaded.replication.smb_user, "") def test_replication_defaults_off_for_a_single_machine(self): with TempTree() as tree: self.assertFalse(tree.load().replication.enabled) def test_shared_root_defaults_beside_the_data_root(self): with TempTree() as tree: loaded = tree.load() self.assertEqual(loaded.shared_root, tree.root / "peach-sync") self.assertEqual(loaded.smb_share, "peach-sync") def test_directories_default_to_their_own_names_under_the_data_root(self): with TempTree() as tree: loaded = tree.load() for key in settings_file.DIRECTORY_KEYS: self.assertEqual(loaded.directory(key), tree.data_root / key) class PriorityTests(unittest.TestCase): """环境变量 > 设置文件 > 内建默认。三层都要有断言,不然改错顺序没人发现。""" FILE = """ [server] port = 9100 mdns_name = 'from-file' [replication] enabled = true smb_host = 'file-host' """ def test_file_overrides_builtin_defaults(self): with TempTree() as tree: tree.write(self.FILE) loaded = tree.load() self.assertEqual(loaded.server.port, 9100) self.assertEqual(loaded.server.mdns_name, "from-file") self.assertTrue(loaded.replication.enabled) # 文件里没写的项仍然取内建默认,不会因为文件存在就变空。 self.assertEqual(loaded.server.host, "127.0.0.1") def test_environment_overrides_the_file(self): with TempTree() as tree: tree.write(self.FILE) loaded = tree.load({ "PEACH_MDNS_NAME": "from-env", "PEACH_SHARED_SMB_HOST": "env-host", }) self.assertEqual(loaded.server.mdns_name, "from-env") self.assertEqual(loaded.replication.smb_host, "env-host") self.assertEqual(loaded.server.port, 9100) def test_empty_environment_variable_does_not_blank_the_file_value(self): # 空字符串是「没设」,不是「设成空」:LaunchAgent 里留个空变量很常见。 with TempTree() as tree: tree.write(self.FILE) loaded = tree.load({"PEACH_MDNS_NAME": ""}) self.assertEqual(loaded.server.mdns_name, "from-file") def test_absolute_directory_moves_that_data_out_of_the_data_root(self): with TempTree() as tree: other = tree.root / "elsewhere" / "db" tree.write(f"[directories]\ndatabase = '{other.as_posix()}'\n") loaded = tree.load() self.assertEqual(loaded.directory("database"), other) self.assertEqual(loaded.directory("state"), tree.data_root / "state") class BadFileTests(unittest.TestCase): def test_syntax_error_names_the_file_and_the_line(self): with TempTree() as tree: path = tree.write("[server]\nport = = 3\n") with self.assertRaises(SettingsFileError) as caught: tree.load() message = str(caught.exception) self.assertIn(str(path), message) # tomllib 的消息自带行号;没有行号的报错等于让人在几十行里自己找。 self.assertIn("line 2", message) def test_wrong_type_names_the_key(self): with TempTree() as tree: tree.write("[server]\nport = 'eight thousand'\n") with self.assertRaises(SettingsFileError) as caught: tree.load() self.assertIn("server.port", str(caught.exception)) def test_boolean_is_not_accepted_as_a_port(self): with TempTree() as tree: tree.write("[server]\nport = true\n") with self.assertRaises(SettingsFileError): tree.load() def test_unknown_directory_key_is_rejected_instead_of_ignored(self): # 打错一个键名却静默忽略,等于那份数据默默落回默认目录。 with TempTree() as tree: tree.write("[directories]\ndatabses = 'db'\n") with self.assertRaises(SettingsFileError) as caught: tree.load() self.assertIn("databses", str(caught.exception)) def test_phase_one_drive_letter_keys_are_rejected_with_an_upgrade_hint(self): """第一阶段的 `[media] R = ...` 是盘符键;静默忽略等于整台机器安静地脱盘。""" with TempTree() as tree: tree.write("[media]\nR = '/Volumes/Media'\n") with self.assertRaises(SettingsFileError) as caught: tree.load() message = str(caught.exception) self.assertIn("R", message) self.assertIn("media.mounts", message) def test_mount_for_an_undeclared_location_is_rejected(self): # 打错一个来源 ID 同样是安静地脱盘,只是范围小一点。 with TempTree() as tree: tree.write("[media.mounts]\nlocla = '/mnt/res'\n") with self.assertRaises(SettingsFileError) as caught: tree.load() self.assertIn("locla", str(caught.exception)) def test_mounts_are_keyed_by_location_id(self): with TempTree() as tree: tree.write("[media.locations]\nlocal = 'D:\\Videos'\n[media.mounts]\nlocal = '/mnt/res/media'\n") loaded = tree.load() self.assertEqual(loaded.mounts, {"local": ("/mnt/res/media",)}) def test_a_newly_declared_location_can_be_mounted(self): with TempTree() as tree: tree.write( "[media.locations]\nnas = 'N:/'\n[media.mounts]\nnas = '/mnt/nas'\n") loaded = tree.load() self.assertEqual(loaded.locations["nas"], ("N:/",)) self.assertEqual(loaded.mounts, {"nas": ("/mnt/nas",)}) def test_a_declared_locations_table_is_authoritative(self): """`peach init` 的问答只声明 local;内建默认的示例盘符不得从旁边混进来。""" with TempTree() as tree: tree.write("[media.locations]\nlocal = 'D:\\Videos'\n") loaded = tree.load() self.assertEqual(loaded.locations, {"local": (r"D:\Videos",)}) def test_a_location_may_declare_several_roots_with_matching_mounts(self): """两块硬盘都是 `local`:数据库路径前缀与媒体目录均写数组,按顺序对应,保存时保留。""" with TempTree() as tree: tree.write("[media.locations]\nlocal = ['D:\\Videos', 'E:\\Movies']\n" "[media.mounts]\nlocal = ['/mnt/a', '/mnt/b']\n") loaded = tree.load() self.assertEqual(loaded.locations, {"local": (r"D:\Videos", r"E:\Movies")}) self.assertEqual(loaded.mounts, {"local": ("/mnt/a", "/mnt/b")}) rendered = settings_file.render(loaded) self.assertIn("local = ['D:\\Videos', 'E:\\Movies']", rendered) self.assertIn("local = ['/mnt/a', '/mnt/b']", rendered) tree.write(rendered) self.assertEqual(tree.load().locations, loaded.locations) def test_mount_count_must_match_the_declared_roots(self): # 媒体目录按顺序对应数据库路径前缀,数目不一致时拒绝保存。 with TempTree() as tree: tree.write("[media.locations]\nlocal = ['D:\\Videos', 'E:\\Movies']\n" "[media.mounts]\nlocal = '/mnt/a'\n") with self.assertRaises(SettingsFileError) as caught: tree.load() self.assertIn("2 个数据库路径前缀", str(caught.exception)) with TempTree() as tree: tree.write("[media.locations]\nlocal = ['D:\\Videos', 'D:\\Videos']\n") with self.assertRaises(SettingsFileError) as caught: tree.load() self.assertIn("重复", str(caught.exception)) with TempTree() as tree: tree.write("[media.locations]\nlocal = []\n") with self.assertRaises(SettingsFileError) as caught: tree.load() self.assertIn("没有配置目录前缀", str(caught.exception)) def test_without_a_locations_table_no_media_roots_are_declared(self): with TempTree() as tree: tree.write("[server]\nport = 9100\n") loaded = tree.load() self.assertEqual(loaded.locations, {}) def test_empty_media_roots_survive_config_roundtrip(self): with TempTree() as tree: tree.write(settings_file.render(tree.load())) self.assertEqual(tree.load().locations, {}) def test_lenient_load_falls_back_to_builtin_defaults(self): """坏文件不能让进程连 Peach 数据目录都不知道——发现顺序不看文件内容。""" with TempTree() as tree: tree.write("nonsense = =\n") loaded = tree.load(strict=False) self.assertEqual(loaded.data_root, tree.data_root) self.assertEqual(loaded.server.port, 8900) self.assertFalse(loaded.present) def test_lenient_load_also_survives_a_merge_time_rejection(self): """语法没错、内容被拒的文件同样得退得下来。 `peach init --force` 是坏文件的唯一自救入口,它走的就是 `strict=False`。 合并期的拒绝漏在退路之外的话,`[media] R = ...` 这种写法会把自救入口 本身打崩——语法完全正确,抛错的是键空间校验。 """ for text in ("[media]\nR = '/mnt/res'\n", "[media.mounts]\nlocla = '/mnt/res'\n", "[directories]\ndatabses = 'db'\n"): with self.subTest(text=text): with TempTree() as tree: tree.write(text) with self.assertRaises(SettingsFileError): tree.load() loaded = tree.load(strict=False) self.assertEqual(loaded.data_root, tree.data_root) self.assertEqual(loaded.server.port, 8900) self.assertEqual(loaded.mounts, {}) self.assertFalse(loaded.present) class ConfiguredStateTests(unittest.TestCase): def test_existing_data_root_counts_as_configured_without_a_settings_file(self): # 现有部署还没生成过设置文件,但它们的账本在跑,不能被判成「未配置」。 with TempTree() as tree: loaded = tree.load() self.assertFalse(loaded.present) self.assertTrue(loaded.configured) def test_fresh_clone_is_unconfigured(self): with TempTree(with_data_root=False) as tree: loaded = tree.load() self.assertFalse(loaded.configured) self.assertEqual(loaded.path.name, settings_file.SETTINGS_FILENAME) class SerialisationTests(unittest.TestCase): def test_render_then_load_returns_the_same_values(self): with TempTree() as tree: tree.write("[media.locations]\nlocal = 'D:\\media'\n'115' = 'Q:\\Cloud'\n") captured = settings_file.capture_existing( tree.load(), overrides={ "mdns_name": "peach-two", "port": 9443, "review_writer_origin": "https://192.0.2.5", "smb_host": "peach-writer.local", "smb_user": "someone", }, # 反斜杠和 Windows 盘符必须原样活过一次往返:序列化器要是把 `\m` # 当转义处理,挂载点就会变成另一个目录。 mounts={"local": (r"D:\media",), "115": ("/mnt/115",)}, ) settings_file.write(captured, force=True) reloaded = tree.load() self.assertEqual(reloaded.server, captured.server) self.assertEqual(reloaded.replication, captured.replication) self.assertEqual(reloaded.mounts, {"local": (r"D:\media",), "115": ("/mnt/115",)}) self.assertEqual(reloaded.locations, captured.locations) self.assertEqual(reloaded.directories, captured.directories) self.assertTrue(reloaded.present) def test_values_with_quotes_survive_the_round_trip(self): with TempTree() as tree: captured = settings_file.capture_existing( tree.load(), overrides={"smb_user": "it's me"}) settings_file.write(captured) self.assertEqual(tree.load().replication.smb_user, "it's me") def test_write_refuses_to_overwrite_without_force(self): with TempTree() as tree: loaded = tree.load() settings_file.write(loaded) with self.assertRaises(SettingsFileError): settings_file.write(loaded) self.assertEqual(settings_file.write(loaded, force=True), loaded.path) def test_capture_keeps_existing_deployments_replicating(self): """第三阶段接上开关时,现有双机部署读到的必须还是 true。""" with TempTree() as tree: captured = settings_file.capture_existing(tree.load()) self.assertTrue(captured.replication.enabled) # 共享目录和共享名落成显式值,别让人去猜默认规则。 self.assertEqual(captured.replication.shared_root, str(tree.root / "peach-sync")) self.assertEqual(captured.replication.smb_share, "peach-sync") def test_fresh_capture_leaves_replication_off(self): with TempTree() as tree: captured = settings_file.capture_existing( tree.load(), replication_enabled=False) self.assertFalse(captured.replication.enabled) class ModuleConstantTests(unittest.TestCase): """`config` 的模块常量被大量 import;名字和相对关系都是契约。""" def test_paths_derive_from_the_active_data_root(self): self.assertEqual(config.DATABASE_PATH, config.DATA_ROOT / "database" / "ledger.db") self.assertEqual(config.DATABASE_PATH.parent, config.DATABASE_DIR) self.assertEqual(config.FFMPEG_DIR, config.TOOLS_DIR / "ffmpeg") self.assertEqual(config.COVER_DIR, config.GENERATED_DIR / "covers") # 迁移随代码走,不随数据走。 self.assertEqual(config.MIGRATIONS_DIR, config.PROJECT_ROOT / "migrations") def test_mdns_hostname_follows_the_configured_name(self): self.assertEqual(config.MDNS_HOSTNAME, f"{config.MDNS_NAME}.local") def test_share_name_defaults_to_the_shared_directory_name(self): # 挂载点是 `/Volumes/<共享名>`:两者分开写就会挂上一个没人去读的路径。 self.assertEqual(config.SHARED_SMB_SHARE, config.SHARED_DATA_ROOT.name) def test_location_declarations_stay_in_ledger_shape(self): # `asset.path` 与配置中的路径前缀均使用 Windows 格式,供目录访问权限检查匹配。 with TempTree() as tree: tree.write("[media.locations]\nlocal = ['P:\\Library']\npikpak = ['Q:\\Cloud']\n") try: with patch.dict(os.environ, {"PEACH_DATA_ROOT": str(tree.data_root)}): settings_file.reset_cache() reloaded = importlib.reload(config) self.assertEqual(reloaded.LOCATION_ROOT_DECLARATIONS, {"local": (r"P:\Library",), "pikpak": (r"Q:\Cloud",)}) self.assertEqual(reloaded.MEDIA_ROOT_DECLARATIONS, (r"P:\Library", r"Q:\Cloud")) finally: settings_file.reset_cache() importlib.reload(config) def test_settings_dataclass_reports_the_configured_state(self): from peach.config import PeachSettings self.assertEqual(PeachSettings().configured, config.CONFIGURED) def test_share_coordinates_can_be_overridden_without_editing_source(self): # 主机名换了、共享改名、账号换一个,都不该要求改源码再发一次版。 overrides = { "PEACH_SHARED_SMB_HOST": "198.51.100.9", "PEACH_SHARED_SMB_SHARE": "ledger-drop", "PEACH_SHARED_SMB_USER": "someone", } try: with patch.dict(os.environ, overrides): settings_file.reset_cache() reloaded = importlib.reload(config) self.assertEqual(reloaded.SHARED_SMB_HOST, "198.51.100.9") self.assertEqual(reloaded.SHARED_SMB_SHARE, "ledger-drop") self.assertEqual(reloaded.SHARED_SMB_USER, "someone") finally: settings_file.reset_cache() importlib.reload(config) self.assertNotEqual(config.SHARED_SMB_HOST, "198.51.100.9") if __name__ == "__main__": unittest.main()