tests/test_cli.py
Rama master · réplica de lectura del repositorio original.
"""`peach` 顶层 CLI 的可达性与只读保证。
这里管的是 `serve`/`migrate`/`status`/`ledger-sync`
这一层，以及打包后的 EXE 到底能不能调到它们——判据见 `build_app_entry.py`。
"""
from __future__ import annotations
import argparse
import importlib.util
import io
import os
import sqlite3
import tempfile
import unittest
from contextlib import redirect_stdout
from dataclasses import replace
from pathlib import Path, PureWindowsPath
from unittest import mock
from peach import cli, onboarding, settings_file
from peach.migrations import plan
from support.ledger import fresh_ledger
ROOT = Path(__file__).resolve().parents[1]
MIGRATIONS = ROOT / "migrations"
HAS_HTTP_DEPS = all(importlib.util.find_spec(name) for name in ("fastapi", "httpx"))
NATIVE_WINDOWS = os.name == "nt"
def load_script(name: str):
path = ROOT / "scripts" / f"{name}.py"
spec = importlib.util.spec_from_file_location(f"test_{name}", path)
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
return module
class StatusCommandTests(unittest.TestCase):
def test_configured_reader_can_inspect_pending_schema_but_cannot_migrate(self):
config = settings_file.load_config(environ={'PEACH_DATA_ROOT': str(self.root)})
database = config.directory('database') / 'ledger.db'
database.parent.mkdir()
sqlite3.connect(database).close()
before = database.read_bytes()
arguments = argparse.Namespace(db=database, action='status', yes=True)
with mock.patch.object(cli, '_require_readable_settings'), mock.patch.object(
settings_file, 'active', return_value=config), mock.patch(
'peach.runtime_prepare.configured_ledger_writer', return_value=(False, '读取端')), \
redirect_stdout(io.StringIO()):
self.assertEqual(cli._migrate(arguments), 0)
arguments.action = 'upgrade'
with self.assertRaisesRegex(SystemExit, '写入端迁移并同步'):
cli._migrate(arguments)
self.assertEqual(database.read_bytes(), before)
self.assertEqual(list(database.parent.iterdir()), [database])
def setUp(self):
self.tmp = tempfile.TemporaryDirectory()
self.addCleanup(self.tmp.cleanup)
self.root = Path(self.tmp.name).resolve()
self.db = fresh_ledger(self.root)
self.shared = self.root / "shared" / "ledger.db"
self.state = self.root / "state"
connection = sqlite3.connect(self.db)
connection.executemany(
"INSERT INTO asset(location,path,name,medium,size,duration,snapshot_path,"
"creator,ctx_length) VALUES(?,?,?,?,?,?,?,?,?)",
[
("local", r"R:\media\a.mp4", "a.mp4", "video", 2 << 30, 1800.0,
r"R:\snap\a.jpg", "Someone", "中"),
("local", r"R:\media\b.mp4", "b.mp4", "video", 1 << 30, None, None,
None, None),
("online", "https://example.test/x", "x", "account", None, None, None,
"Handle", None),
])
connection.commit()
connection.close()
def _run(self, argv):
output = io.StringIO()
with mock.patch.object(cli, "STATE_DIR", self.state), redirect_stdout(output):
code = cli.main(argv)
return code, output.getvalue()
def test_status_reports_ledger_migration_and_sync_state(self):
code, output = self._run(["status", "--db", str(self.db),
"--shared-db", str(self.shared)])
self.assertEqual(code, 0)
self.assertIn("合计", output)
self.assertIn("加工进度（本机视频 2 条）", output)
# 在线资产不进「加工进度」的分母，否则百分比永远到不了 100。
self.assertIn("有时长 (ffprobe)", output)
self.assertIn("1 / 2", output)
self.assertIn("待应用 0 个", output)
self.assertIn("同步：", output)
def test_status_never_writes_the_ledger_or_invents_a_device_id(self):
before = self.db.stat().st_mtime_ns, self.db.stat().st_size
code, _ = self._run(["status", "--db", str(self.db),
"--shared-db", str(self.shared)])
self.assertEqual(code, 0)
self.assertEqual((self.db.stat().st_mtime_ns, self.db.stat().st_size), before)
self.assertFalse((self.state / "device-id").exists(),
"状态命令不该顺手生成一个从没写过库的写入端标识")
def test_missing_ledger_is_reported_rather_than_created(self):
missing = self.root / "nope.db"
code, output = self._run(["status", "--db", str(missing),
"--shared-db", str(self.shared)])
self.assertEqual(code, 4)
self.assertIn("数据库不存在", output)
self.assertFalse(missing.exists())
class InitCommandTests(unittest.TestCase):
"""`peach init`：一台全新机器从零到能跑的唯一入口。"""
def setUp(self):
self.tmp = tempfile.TemporaryDirectory()
self.addCleanup(self.tmp.cleanup)
self.root = Path(self.tmp.name).resolve() / "peach-data"
# 证书生成要调 openssl；那是 `test_certs.py` 的事，这里只验初始化流程。
patcher = mock.patch.object(cli.onboarding.certs, "bootstrap_certificates")
self.certs = patcher.start()
self.addCleanup(patcher.stop)
def _run(self, argv):
output = io.StringIO()
with redirect_stdout(output):
code = cli.main(argv)
return code, output.getvalue()
def _init(self, *extra):
return self._run(["init", "--data-root", str(self.root), *extra])
def test_fresh_init_builds_a_runnable_data_root(self):
code, output = self._init()
self.assertEqual(code, 0)
for key in settings_file.DIRECTORY_KEYS:
self.assertTrue((self.root / key).is_dir(), key)
database = self.root / "database" / "ledger.db"
self.assertTrue(database.is_file())
# 建完就必须是最新 schema，否则第一次 serve 就撞上待应用迁移。
self.assertEqual(plan(database, MIGRATIONS)[1], [])
self.assertIn("下一步", output)
self.certs.assert_called_once()
def test_declared_mount_output_is_readable_by_the_settings_layer(self):
self.root.mkdir()
(self.root / "config.toml").write_text(
"[media.locations]\nlocal = ['P:\\Library']\n", encoding="utf-8")
self._init("--force", "--mdns-name", "peach-two", "--port", "9443",
"--mount", "local=/mnt/media")
loaded = settings_file.load_config(environ={"PEACH_DATA_ROOT": str(self.root)})
self.assertTrue(loaded.present)
self.assertTrue(loaded.configured)
self.assertEqual(loaded.server.mdns_name, "peach-two")
self.assertEqual(loaded.server.port, 9443)
self.assertEqual(loaded.mounts, {"local": ("/mnt/media",)})
# 全新机器不开复制：单机用户不该凭空多出一条同步路径。
self.assertFalse(loaded.replication.enabled)
def test_mount_for_an_undeclared_location_is_refused(self):
"""打错来源 ID 静默收下，那个来源就会安静地进脱盘模式。"""
with self.assertRaises(SystemExit) as caught:
self._init("--mount", "locla=/mnt/media")
self.assertIn("locla", str(caught.exception))
self.assertIn("尚未声明媒体来源", str(caught.exception))
def test_mount_without_an_equals_sign_is_refused(self):
with self.assertRaises(SystemExit) as caught:
self._init("--mount", "/mnt/media")
self.assertIn("来源ID=路径", str(caught.exception))
def test_mount_ids_keep_their_case(self):
"""来源 ID 不是盘符，`local` 不能被改写成 `LOCAL`。"""
self.assertEqual(
cli._parse_mounts(["local=/mnt/res"], {"local": (r"R:\media",)}),
{"local": ("/mnt/res",)})
# 同一来源的多个媒体目录，按顺序对应数据库路径前缀。
self.assertEqual(
cli._parse_mounts(["local=/mnt/a", "local=/mnt/b"], {"local": (r"R:\media", r"R:\media2")}),
{"local": ("/mnt/a", "/mnt/b")})
def test_the_new_ledger_is_readable_by_status(self):
self._init()
code, output = self._run(
["status", "--db", str(self.root / "database" / "ledger.db"),
"--shared-db", str(self.root / "nope.db")])
self.assertEqual(code, 0)
self.assertIn("待应用 0 个", output)
def test_existing_settings_file_is_not_overwritten_without_force(self):
self._init()
stamp = (self.root / "config.toml").read_text(encoding="utf-8")
code, output = self._init("--mdns-name", "changed")
self.assertEqual(code, 3)
self.assertIn("已存在", output)
self.assertEqual((self.root / "config.toml").read_text(encoding="utf-8"), stamp)
self.assertEqual(self._init("--mdns-name", "changed", "--force")[0], 0)
self.assertNotEqual(
(self.root / "config.toml").read_text(encoding="utf-8"), stamp)
def test_from_existing_only_writes_the_settings_file(self):
code, _ = self._init("--from-existing", "--smb-host", "peach-writer.local",
"--smb-user", "someone")
self.assertEqual(code, 0)
self.assertFalse((self.root / "database").exists())
self.assertFalse((self.root / "generated").exists())
self.certs.assert_not_called()
loaded = settings_file.load_config(environ={"PEACH_DATA_ROOT": str(self.root)})
# 现有部署本来就在复制；写一份设置文件不该把它悄悄关掉。
self.assertTrue(loaded.replication.enabled)
self.assertEqual(loaded.replication.smb_host, "peach-writer.local")
self.assertEqual(loaded.replication.smb_user, "someone")
def test_from_existing_says_so_when_the_old_file_is_unreadable(self):
"""升级路径上最贵的一次静默：第一阶段的 `[media] R = ...` 现在是硬错误。
`--from-existing` 声称落盘的是「当前生效的配置」，而旧文件读不出来时那份配置
其实是内建默认。不说这一句，重写出来的文件会安静地丢掉 mDNS 名和 SMB 坐标。
"""
self.root.mkdir(parents=True)
(self.root / "config.toml").write_text(
"[media]\nR = '/Volumes/Media'\n", encoding="utf-8")
code, output = self._init("--from-existing", "--force")
self.assertEqual(code, 0)
self.assertIn("media.mounts", output) # 原始错误里的升级提示
self.assertIn("没有**继承", output)
loaded = settings_file.load_config(environ={"PEACH_DATA_ROOT": str(self.root)})
self.assertEqual(loaded.mounts, {})
def test_blank_coordinates_are_listed_instead_of_guessed(self):
_, output = self._init("--from-existing")
self.assertIn("replication.smb_host", output)
self.assertIn("--smb-host", output)
def test_an_existing_ledger_is_never_rebuilt_by_a_fresh_init(self):
database = self.root / "database" / "ledger.db"
database.parent.mkdir(parents=True)
database.write_bytes(b"not really a database")
code, output = self._init()
self.assertEqual(code, 0)
self.assertEqual(database.read_bytes(), b"not really a database")
self.assertIn("--from-existing", output)
class InteractiveInitTests(unittest.TestCase):
"""`peach init` 不带参数、stdin 是终端时的问答流。
答案一律脚本化。两种形态各有一条专属用例，其余与形态无关的用例按 `NATIVE_WINDOWS`
走本机形态：媒体目录题只收真实存在的目录，而目录是不是盘符路径由测试机决定，
在 POSIX 上硬填 `windows=True` 会被媒体目录校验直接拒掉。
"""
def setUp(self):
self.tmp = tempfile.TemporaryDirectory()
self.addCleanup(self.tmp.cleanup)
self.root = Path(self.tmp.name).resolve()
self.data_root = self.root / "peach-data"
self.home = self.root / "home"
self.home.mkdir()
self.media = self.root / "media"
(self.media / "sub").mkdir(parents=True)
(self.media / "sub" / "a.mp4").write_bytes(b"0" * 10)
(self.media / "b.jpg").write_bytes(b"1")
patcher = mock.patch.object(cli.onboarding.certs, "bootstrap_certificates")
self.certs = patcher.start()
self.addCleanup(patcher.stop)
# 第一题的默认值要落在临时目录里，回车接受默认才不会碰真实 Peach 数据目录。
env = mock.patch.dict(os.environ, {"PEACH_DATA_ROOT": str(self.data_root)})
env.start()
self.addCleanup(env.stop)
def _run(self, answers, *, windows):
queue = list(answers)
prompts: list[str] = []
def ask(prompt: str, default: str) -> str:
prompts.append(prompt)
return queue.pop(0)
output = io.StringIO()
with redirect_stdout(output):
code = cli._init_interactive(ask, windows=windows, home=self.home)
return code, output.getvalue(), prompts
def _ledger_paths(self):
connection = sqlite3.connect(self.data_root / "database" / "ledger.db")
try:
return sorted(row[0] for row in connection.execute("SELECT path FROM asset"))
finally:
connection.close()
def _loaded(self):
return settings_file.load_config(environ={"PEACH_DATA_ROOT": str(self.data_root)})
@unittest.skipUnless(NATIVE_WINDOWS, "本用例需要真实存在的 Windows 盘符目录")
def test_windows_flow_declares_the_directory_and_scans_it(self):
second = self.root / "more"
second.mkdir()
(second / "c.mp4").write_bytes(b"2")
code, output, prompts = self._run(
["", str(self.media), str(second), "", "", "", "", ""], windows=True)
self.assertEqual(code, 0)
self.assertEqual(len(prompts), 8)
self.assertEqual(prompts[2], prompts[3], "追加目录问到回车为止")
self.assertTrue(prompts[-1].startswith("现在扫描 这 2 个文件夹"))
loaded = self._loaded()
self.assertTrue(loaded.present)
self.assertEqual(loaded.locations, {"local": (str(self.media), str(second))})
self.assertEqual(loaded.mounts, {})
self.assertEqual((loaded.server.host, loaded.server.port, loaded.server.mdns_name),
("0.0.0.0", 8900, "peach"))
self.assertFalse(loaded.replication.enabled)
for key in settings_file.DIRECTORY_KEYS:
self.assertTrue((self.data_root / key).is_dir(), key)
self.certs.assert_called_once()
# 账本里的路径就是本机路径，每一行都指向真实文件。
self.assertEqual(self._ledger_paths(),
sorted([str(self.media / "b.jpg"), str(self.media / "sub" / "a.mp4"),
str(second / "c.mp4")]))
self.assertIn("✓ local: 2 文件", output)
self.assertIn("✓ local: 1 文件", output)
self.assertIn("下一步", output)
self.assertEqual(output.count("扫描结果："), 2, "每个目录各报一行")
self.assertNotIn("115", loaded.locations)
def test_posix_flow_keeps_the_ledger_shape_and_mounts_the_directory(self):
code, output, _ = self._run(["", str(self.media), "", "2", "9443", "peach-two", ""],
windows=False)
self.assertEqual(code, 0)
loaded = self._loaded()
self.assertEqual(loaded.locations, {"local": (r"R:\media",)})
self.assertEqual(loaded.mounts, {"local": (str(self.media),)})
self.assertEqual((loaded.server.host, loaded.server.port, loaded.server.mdns_name),
("0.0.0.0", 9443, "peach-two"))
self.assertIn("媒体目录", output)
paths = self._ledger_paths()
self.assertEqual(paths, [r"R:\media\b.jpg", r"R:\media\sub\a.mp4"])
# 读取文件时，将数据库路径前缀之后的层级拼到本机媒体目录。
for path in paths:
tail = PureWindowsPath(path).relative_to(PureWindowsPath(r"R:\media")).parts
self.assertTrue(Path(loaded.mounts["local"][0]).joinpath(*tail).is_file(), path)
def test_an_invalid_media_directory_is_asked_again(self):
missing = self.root / "nope"
code, output, prompts = self._run(
["", str(missing), str(self.media), "", "", "", "", "n"], windows=NATIVE_WINDOWS)
self.assertEqual(code, 0)
self.assertEqual(prompts.count("媒体目录（必须已经存在，可以在外置硬盘上）"), 2)
self.assertIn("目录不存在", output)
self.assertFalse(missing.exists(), "问答不替人建媒体目录")
def test_three_bad_answers_abort_before_anything_is_written(self):
with self.assertRaises(SystemExit) as caught:
self._run(["", "/nope1", "/nope2", "/nope3"], windows=NATIVE_WINDOWS)
self.assertIn("3 次", str(caught.exception))
self.assertIn("没有写任何文件", str(caught.exception))
self.assertFalse(self.data_root.exists())
def test_the_scan_can_be_declined(self):
code, output, _ = self._run(["", str(self.media), "", "", "", "", "n"], windows=NATIVE_WINDOWS)
self.assertEqual(code, 0)
self.assertEqual(self._ledger_paths(), [])
self.assertNotIn("扫描结果", output)
self.assertIn("下一步", output)
def test_an_existing_settings_file_is_not_overwritten(self):
self.data_root.mkdir()
(self.data_root / "config.toml").write_text("[server]\nport = 9100\n", encoding="utf-8")
code, output, _ = self._run(["", str(self.media), "", "", ""], windows=NATIVE_WINDOWS)
self.assertEqual(code, 3)
self.assertIn("已存在", output)
self.assertEqual((self.data_root / "config.toml").read_text(encoding="utf-8"),
"[server]\nport = 9100\n")
@unittest.skipUnless(HAS_HTTP_DEPS, "需要 fastapi 与 httpx")
def test_a_config_declaring_only_local_serves_health_items_and_sources(self):
"""只声明 `local` 的设置文件要能直接起服务：没有 115/pikpak 不是错误。"""
import asyncio
import httpx
from peach import web_resource_sync
from peach.api import create_app
from peach.config import PeachSettings
from peach.platform import translate_roots
self._run(["", str(self.media), "", "", "", "", ""], windows=NATIVE_WINDOWS)
loaded = self._loaded()
# POSIX 系统也用 `R:\media` 作为数据库路径前缀，按 [media.mounts] 转换为本机目录；
# 模块级 `active()` 指的是这台机器自己的设置，临时目录不在里面。
active = mock.patch.object(settings_file, "active", lambda: loaded)
active.start()
self.addCleanup(active.stop)
settings = PeachSettings(
db_path=loaded.directory("database") / "ledger.db", configured=True, token="secret",
cover_root=loaded.directory("generated") / "covers",
allowed_media_roots=translate_roots(tuple(root for roots in loaded.locations.values() for root in roots)),
)
async def probe():
app = create_app(settings)
async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app),
base_url="http://test") as client:
health = await client.get("/healthz")
items = await client.get("/api/items?limit=10", headers={"Cookie": "tok=secret"})
sources = await client.get("/api/sources", headers={"Cookie": "tok=secret"})
return health, items, sources
with mock.patch("peach.config.LOCATION_ROOT_DECLARATIONS", dict(loaded.locations)), \
mock.patch.object(web_resource_sync, "LOCATION_ROOT_DECLARATIONS",
dict(loaded.locations)):
health, items, sources = asyncio.run(probe())
# 两个闸门断言必须留在 patch 里：出了 with 就换回模块默认声明，
# 那份声明带着 115/pikpak，在真挂上它们的机器上 source_is_online 会返回 True。
self.assertTrue(web_resource_sync.source_is_online("local"))
self.assertFalse(web_resource_sync.source_is_online("115"))
self.assertEqual(health.status_code, 200)
self.assertEqual(items.status_code, 200)
self.assertGreaterEqual(items.json()["total"], 1, "扫进去的本地资产要能列出来")
self.assertEqual(sources.status_code, 200)
listed = [row["location"] for row in sources.json()["sources"]]
self.assertIn("local", listed)
self.assertNotIn("115", listed)
self.assertNotIn("pikpak", listed)
class InitDispatchTests(unittest.TestCase):
"""问答只在「不带任何参数且 stdin 是终端」时进入；其余一律走非交互路径。"""
def setUp(self):
self.tmp = tempfile.TemporaryDirectory()
self.addCleanup(self.tmp.cleanup)
self.root = Path(self.tmp.name).resolve() / "peach-data"
patcher = mock.patch.object(cli, "_init_interactive", return_value=0)
self.interactive = patcher.start()
self.addCleanup(patcher.stop)
certs = mock.patch.object(cli.onboarding.certs, "bootstrap_certificates")
certs.start()
self.addCleanup(certs.stop)
def _run(self, argv, *, tty):
with mock.patch.object(onboarding, "is_interactive", return_value=tty), \
redirect_stdout(io.StringIO()):
return cli.main(argv)
def test_bare_init_on_a_terminal_enters_the_interview(self):
self.assertEqual(self._run(["init"], tty=True), 0)
self.interactive.assert_called_once_with()
def test_bare_init_without_a_terminal_stays_non_interactive(self):
with mock.patch.dict(os.environ, {"PEACH_DATA_ROOT": str(self.root)}):
self.assertEqual(self._run(["init"], tty=False), 0)
self.interactive.assert_not_called()
self.assertTrue((self.root / "config.toml").is_file())
def test_any_argument_or_no_input_stays_non_interactive(self):
self.assertEqual(self._run(["init", "--data-root", str(self.root)], tty=True), 0)
self.assertEqual(
self._run(["init", "--data-root", str(self.root), "--no-input", "--force"], tty=True), 0)
self.interactive.assert_not_called()
loaded = settings_file.load_config(environ={"PEACH_DATA_ROOT": str(self.root)})
self.assertEqual(loaded.locations, {})
class ScanCommandTests(unittest.TestCase):
"""`peach scan <来源ID> [根目录]`：目录省略时取配置中的数据库路径前缀。
`cli._scan` 不带 `windows=`，走的就是测试机的形态，所以设置也按本机形态搭：Windows 上
数据库路径前缀就是真实目录、挂载表为空；POSIX 系统使用 `R:\\media`，挂载表指向真实目录。
"""
def setUp(self):
self.tmp = tempfile.TemporaryDirectory()
self.addCleanup(self.tmp.cleanup)
self.root = Path(self.tmp.name).resolve()
self.db = fresh_ledger(self.root)
self.media = self.root / "media"
(self.media / "sub").mkdir(parents=True)
(self.media / "sub" / "a.mp4").write_bytes(b"0")
(self.media / "b.jpg").write_bytes(b"1")
config = settings_file.load_config(environ={}, strict=False)
declared = str(self.media) if NATIVE_WINDOWS else r"R:\media"
mounts = {} if NATIVE_WINDOWS else {"local": (Path(self.media),)}
fixed = replace(config, locations={"local": (declared,)},
mounts={key: tuple(str(item) for item in value)
for key, value in mounts.items()})
for target in (mock.patch.object(settings_file, "active", lambda: fixed),
mock.patch.object(cli, "location_mounts", lambda: mounts),
mock.patch.object(cli, "SETTINGS_ERROR", None)):
target.start()
self.addCleanup(target.stop)
def _run(self, argv):
output = io.StringIO()
with redirect_stdout(output):
code = cli.main(argv)
return code, output.getvalue()
def _count(self):
connection = sqlite3.connect(self.db)
try:
return connection.execute("SELECT COUNT(*) FROM asset").fetchone()[0]
finally:
connection.close()
def test_root_defaults_to_the_declared_root(self):
code, output = self._run(["scan", "local", "--db", str(self.db)])
self.assertEqual(code, 0)
self.assertIn("✓ local: 2 文件", output)
self.assertEqual(self._count(), 2)
def test_configured_scan_uses_each_online_source_id(self):
fixed = replace(settings_file.active(), locations={'115': ('B:/',), 'pikpak': ('A:/',)})
with mock.patch.object(settings_file, 'active', return_value=fixed), \
mock.patch('peach.platform.root_online', side_effect=[False, True]), \
mock.patch.object(cli.scan, 'scan_location') as scanner:
code, output = self._run(['scan', 'configured', '--db', str(self.db)])
self.assertEqual(code, 0)
self.assertIn('115', output)
self.assertEqual(scanner.call_count, 1)
self.assertEqual(scanner.call_args.args[1:3], ('pikpak', 'A:/'))
def test_an_explicit_subdirectory_is_scanned_alone(self):
code, _ = self._run(["scan", "local", str(self.media / "sub"), "--db", str(self.db)])
self.assertEqual(code, 0)
self.assertEqual(self._count(), 1)
def test_an_undeclared_source_is_refused_and_lists_the_known_ones(self):
with self.assertRaises(SystemExit) as caught:
self._run(["scan", "nas", "--db", str(self.db)])
self.assertIn("nas", str(caught.exception))
self.assertIn("local", str(caught.exception))
self.assertEqual(self._count(), 0)
def test_a_missing_ledger_points_at_init(self):
code, output = self._run(["scan", "local", "--db", str(self.root / "nope.db")])
self.assertEqual(code, 4)
self.assertIn("peach init", output)
class ProcessCommandTests(unittest.TestCase):
"""`peach process` 和网页那一轮同一个处理函数，结算也要一样：声明的后继进任务中心。"""
def setUp(self):
self.tmp = tempfile.TemporaryDirectory()
self.addCleanup(self.tmp.cleanup)
self.db = fresh_ledger(self.tmp.name)
config = settings_file.load_config(environ={}, strict=False)
for target in (mock.patch.object(settings_file, "active", lambda: config),
mock.patch.object(cli, "SETTINGS_ERROR", None)):
target.start()
self.addCleanup(target.stop)
def _run(self, result):
output = io.StringIO()
with mock.patch("peach.library_processing.process_library", return_value=result), \
redirect_stdout(output):
code = cli.main(["process", "--db", str(self.db)])
connection = sqlite3.connect(self.db)
try:
rows = connection.execute(
"SELECT id,task_key,trigger,status,parent_run_id,followup_key FROM task_run "
"ORDER BY id").fetchall()
finally:
connection.close()
return code, output.getvalue(), rows
def test_the_followups_a_run_declares_are_queued_under_it(self):
declared = [{"key": "performer-alias:9378", "task_key": "performer-alias",
"label": "补女优别名：白川麻衣"}]
code, output, rows = self._run({"status": "complete", "followups": declared})
self.assertEqual(code, 0)
parent, child = rows
self.assertEqual(parent[1:4], ("library-processing", "cli", "succeeded"))
self.assertEqual(child[1:], ("performer-alias", "cli", "pending", parent[0],
"performer-alias:9378"))
self.assertIn("入队 1 条", output)
def test_an_unfinished_run_is_recorded_as_failed_and_still_queues_its_followups(self):
declared = [{"key": "performer-alias:1", "task_key": "performer-alias", "label": ""}]
code, _, rows = self._run({"status": "failed", "error": "网盘离线",
"followups": declared})
self.assertEqual(code, 1)
self.assertEqual(rows[0][3], "failed")
self.assertEqual(rows[1][3], "pending")
def test_user_cancel_returns_a_cancelled_receipt_without_dispatching_followups(self):
from peach.jobs import JobCancelled
from contextlib import closing
with mock.patch('peach.library_processing.process_library',side_effect=JobCancelled('用户停止任务')), \
redirect_stdout(io.StringIO()) as output:
code=cli.main(['process','--db',str(self.db)])
with closing(sqlite3.connect(self.db)) as connection:
rows=connection.execute('SELECT status,parent_run_id FROM task_run').fetchall()
self.assertEqual(code,130)
self.assertEqual(rows,[('cancelled',None)])
self.assertIn('已保存结果保留',output.getvalue())
def test_progress_lines_show_files_found_while_scanning_and_skip_repeats(self):
def process(*_args, report, **_kwargs):
scanning = {"stage": "扫描文件", "checked": 0, "total": 0, "current_action": "scanning_files"}
report({**scanning, "scanned": 0})
report({**scanning, "scanned": 8096})
report({**scanning, "scanned": 8096})
report({"stage": "读取时长与分辨率", "checked": 12, "total": 40, "scanned": 8096})
report({"stage": "整理结果", "checked": 0, "total": 0, "scanned": 8096})
return {"status": "complete", "followups": []}
with mock.patch("peach.library_processing.process_library", side_effect=process), \
redirect_stdout(io.StringIO()) as output:
code = cli.main(["process", "--db", str(self.db)])
self.assertEqual(code, 0)
self.assertEqual(output.getvalue().splitlines(), [
"扫描文件：已发现 0 个文件", "扫描文件：已发现 8,096 个文件",
"读取时长与分辨率：12/40", "整理结果"])
class ReplicationAssemblyTests(unittest.TestCase):
"""ADR-0023 第 3 阶段：整条复制链路按 `replication.enabled` 装配或完全不装配。"""
def _build(self, enabled):
args = argparse.Namespace(shared_db=Path("/nonexistent/shared.db"))
settings = mock.Mock(db_path=Path("/nonexistent/local.db"))
output = io.StringIO()
with mock.patch.object(cli, "REPLICATION_ENABLED", enabled), \
mock.patch.object(cli, "LedgerSync") as ledger_sync, \
redirect_stdout(output):
sync = cli._build_sync(args, settings)
return sync, ledger_sync, output.getvalue()
def test_replication_off_builds_no_observer_and_probes_nothing(self):
"""没有第二台机器就没有「读者」：服务按独立写者跑，写接口照常开。"""
sync, ledger_sync, output = self._build(False)
self.assertIsNone(sync)
ledger_sync.assert_not_called()
self.assertIn("disabled", output)
def test_replication_on_keeps_todays_behaviour(self):
sync, ledger_sync, _ = self._build(True)
ledger_sync.assert_called_once()
ledger_sync.return_value.observe.assert_called_once()
self.assertIs(sync, ledger_sync.return_value)
class BrokenSettingsTests(unittest.TestCase):
"""设置文件读不出来时，命令必须响亮地失败，而不是拿默认值跑出个假状态。"""
def test_commands_refuse_to_run_and_name_the_file(self):
broken = settings_file.SettingsFileError(
"设置文件读不出来：/tmp/peach-data/config.toml：TOML 语法错误")
with mock.patch.object(cli, "SETTINGS_ERROR", broken):
with self.assertRaises(SystemExit) as caught:
cli.main(["status", "--db", "/tmp/nope.db"])
message = str(caught.exception)
self.assertIn("config.toml", message)
self.assertIn("peach init --force", message)
class PackagedEntryTests(unittest.TestCase):
"""打包后的 EXE 必须能调到 `peach.cli` 的每一个子命令。
判据硬编码成 `{"serve", "migrate"}` 的话，`ledger-sync` 和后来加的
`status` 在 EXE 里全部不可达——参数被当成托盘参数吞掉，既不报错也不执行。
"""
@classmethod
def setUpClass(cls):
cls.entry = load_script("build_app_entry")
def setUp(self):
self.enterContext(mock.patch("peach.error_reporting.install"))
self.enterContext(mock.patch("peach.error_reporting.event"))
self.enterContext(mock.patch("peach.error_reporting.console_ready"))
def test_every_cli_subcommand_is_reachable_from_the_packaged_entry(self):
subcommands = set()
for action in cli.build_parser()._subparsers._group_actions:
subcommands.update(action.choices)
self.assertIn("status", subcommands)
self.assertEqual(subcommands, self.entry.cli_commands())
def test_windows_redirected_console_accepts_chinese_startup_messages(self):
output = io.BytesIO()
stream = io.TextIOWrapper(output, encoding="cp1252", newline="\n")
with mock.patch.object(self.entry.os, "name", "nt"), mock.patch.object(
self.entry.sys, "stdout", stream
), mock.patch.object(self.entry.sys, "stderr", stream):
self.entry._prepare_console()
print("服务已启动", flush=True)
self.assertEqual(output.getvalue(), "服务已启动\n".encode("utf-8"))
def test_tray_still_starts_when_the_first_argument_is_not_a_subcommand(self):
self.assertFalse(self.entry.wants_cli(["peach.exe"]))
self.assertFalse(self.entry.wants_cli(["peach.exe", "--tray-only"]))
self.assertTrue(self.entry.wants_cli(["peach.exe", "status"]))
self.assertTrue(self.entry.wants_cli(["peach.exe", "ledger-sync"]))
def test_installer_hooks_reach_the_installer_handoff_not_the_tray(self):
with mock.patch("peach.desktop_installer.main", return_value=0) as handoff, \
mock.patch.object(self.entry, "tray_main") as tray:
self.assertEqual(self.entry.main(["peach.exe", "--installer-uninstall"]), 0)
handoff.assert_called_once_with("--installer-uninstall")
tray.assert_not_called()
def test_packaged_command_failure_returns_without_a_native_error_dialog(self):
output = io.StringIO()
with mock.patch.object(self.entry, "cli_main", side_effect=RuntimeError("migration failed")), mock.patch.object(
self.entry, "_prepare_console"
), mock.patch.object(self.entry.sys, "stderr", output):
self.assertEqual(self.entry.main(["peach.exe", "migrate", "upgrade", "--yes"]), 1)
self.assertIn("migration failed", output.getvalue())
if __name__ == "__main__":
unittest.main()
class SourceChecksCliTests(unittest.TestCase):
def test_source_check_selection_json_and_failed_exit_status(self):
result = {'ok': False, 'sources': [], 'elapsed_s': 0}
with mock.patch('peach.source_preflight.check_sources', return_value=result) as check, \
mock.patch('peach.cli.settings_file.active') as active, redirect_stdout(io.StringIO()) as output:
self.assertEqual(cli.main(['check-sources', '--source', 'dmm', '--source', 'javdb', '--json']), 1)
check.assert_called_once_with(active.return_value.directory.return_value, ['dmm', 'javdb'])
self.assertEqual(__import__('json').loads(output.getvalue()), result)