# -*- coding: utf-8 -*- """CLI 入口测试（离线）: uv run python tests/test_cli.py""" import contextlib import io import json import os import sys import tempfile from pathlib import Path sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) sys.stdout.reconfigure(encoding="utf-8") from reaction_board.bilibili.models import LoginState # noqa: E402 from reaction_board.cli import main # noqa: E402 from reaction_board.database import Database # noqa: E402 from reaction_board.repository import Repository # noqa: E402 import reaction_board.cli as cli # noqa: E402 def write_config(tmp, *, cookie="SESSDATA=x; buvid3=y", csrf="csrf"): path = Path(tmp) / "config.json" path.write_text(json.dumps({ "bilibili": { "read": {"mid": 1, "csrf": csrf, "cookie": cookie}, "write": {"mid": 2, "csrf": "write-csrf", "cookie": "write-cookie"}, }, "storage": { "database_path": str(Path(tmp) / "app.db"), "backup_dir": str(Path(tmp) / "backup"), }, "folders": [ {"year": 2025, "media_id": 11, "title": "2025"}, {"year": 2026, "media_id": 22, "title": "2026"}, ], "up_blacklist_seed": [101, 202], }, ensure_ascii=False), encoding="utf-8") return path def run_cli(argv): out = io.StringIO() with contextlib.redirect_stdout(out): code = main(argv) return code, out.getvalue() def test_init_and_status(): with tempfile.TemporaryDirectory() as tmp: path = write_config(tmp) code, output = run_cli(["init", "--config", str(path)]) assert code == 0 assert '"schema_version": 8' in output db = Database(Path(tmp) / "app.db") db.initialize() try: repo = Repository(db) assert len(repo.list_folders()) == 2 assert [r["mid"] for r in repo.blacklist()] == [101, 202] assert repo.counts()["video_total"] == 0 finally: db.close() code, output = run_cli(["status", "--config", str(path)]) assert code == 0 assert '"configured_folders"' in output assert '"media_count": null' in output assert '"video_total": 0' in output def test_missing_config(): with tempfile.TemporaryDirectory() as tmp: code, output = run_cli(["init", "--config", str(Path(tmp) / "nope.json")]) assert code == 2 payload = json.loads(output.strip().splitlines()[-1]) assert payload["level"] == "error" assert "缺少配置文件" in payload["error"] def test_ingest_without_refs_is_offline(): with tempfile.TemporaryDirectory() as tmp: path = write_config(tmp, cookie="", csrf="") code, output = run_cli(["ingest", "--config", str(path), "没有视频"]) assert code == 1 data = json.loads(output[output.rindex("{\n"):]) assert data["refs"] == 0 assert data["results"] == [] def test_snapshot_offline_empty_db(): with tempfile.TemporaryDirectory() as tmp: path = write_config(tmp, cookie="", csrf="") data = json.loads(path.read_text(encoding="utf-8")) data["folders"] = [] data["track_fans"] = [] path.write_text(json.dumps(data, ensure_ascii=False), encoding="utf-8") code, output = run_cli(["snapshot", "--config", str(path)]) assert code == 0 assert '"completed": true' in output assert '"total": 0' in output assert len(list((Path(tmp) / "backup").glob("app-*.db"))) == 1 def test_auth_check_without_cookie(): with tempfile.TemporaryDirectory() as tmp: path = write_config(tmp, cookie="", csrf="") code, output = run_cli(["auth-check", "--config", str(path)]) assert code == 1 payload = json.loads(output) assert payload["is_login"] is False assert "未配置 cookie" in payload["message"] def test_auth_check_all_and_auth_dispatch(): with tempfile.TemporaryDirectory() as tmp: path = write_config(tmp, cookie="", csrf="") code, output = run_cli(["auth-check", "--account", "all", "--config", str(path)]) assert code == 1 payload = json.loads(output) assert payload["ok"] is False assert [item["account"] for item in payload["accounts"]] == ["read", "write"] original = cli.AuthService calls = [] class FakeAuthService: def __init__(self, path, *, account, logger): calls.append((str(path), account)) def run(self, *, timeout, no_browser): assert timeout == 9 and no_browser is True return 7 cli.AuthService = FakeAuthService try: code, _ = run_cli([ "auth", "--account", "write", "--timeout", "9", "--no-browser", "--config", str(path), ]) finally: cli.AuthService = original assert code == 7 assert calls == [(str(path), "write")] def test_run_once_offline(): class FakeClient: def check_login(self): return LoginState(True, 1, "tester", 0, "OK") class FakeDiscovery: def __init__(self, *args, **kwargs): pass def run(self): from reaction_board.services.discovery import DiscoverySummary return DiscoverySummary(started_at=1, finished_at=1) with tempfile.TemporaryDirectory() as tmp: path = write_config(tmp) data = json.loads(path.read_text(encoding="utf-8")) data["folders"] = [] data["track_fans"] = [] path.write_text(json.dumps(data, ensure_ascii=False), encoding="utf-8") original = cli._client original_discovery = cli.DiscoveryService selected = [] def fake_client(credentials, *args, **kwargs): selected.append(credentials) return FakeClient() cli._client = fake_client cli.DiscoveryService = FakeDiscovery try: code, output = run_cli(["run", "--once", "--config", str(path)]) finally: cli._client = original cli.DiscoveryService = original_discovery assert code == 0 assert '"name": "snapshot"' in output assert '"name": "folder_sync"' not in output assert selected and all(credentials.mid == 1 for credentials in selected) db = Database(Path(tmp) / "app.db") db.initialize() try: repo = Repository(db) for name in ("snapshot", "discovery", "candidate_cleanup"): assert repo.kv_get(f"task:last:{name}"), name finally: db.close() def test_serve_preview_guard(): with tempfile.TemporaryDirectory() as tmp: path = write_config(tmp) assert run_cli(["init", "--config", str(path)])[0] == 0 db_path = Path(tmp) / "app.db" try: cli._assert_preview_database(db_path) except cli.ConfigError as exc: assert "preview:seeded" in str(exc) else: raise AssertionError("普通数据库不应通过 --preview 校验") db = Database(db_path) db.initialize() try: Repository(db).kv_set("preview:seeded", "1") finally: db.close() cli._assert_preview_database(db_path) try: cli._assert_preview_database(Path(tmp) / "missing.db") except cli.ConfigError as exc: assert "不存在" in str(exc) else: raise AssertionError("缺失数据库应报错") def test_default_command_starts_web_and_scheduler(): with tempfile.TemporaryDirectory() as tmp: path = write_config(tmp) calls = [] original = cli.cmd_run def fake_run(args, logger, *, serve_web=False): calls.append((args.command, serve_web, args.once)) return 0 cli.cmd_run = fake_run try: code = main(["--config", str(path)]) finally: cli.cmd_run = original assert code == 0 assert calls == [("run", True, False)] def test_snapshot_backup_failure_returns_error_and_retries(): import reaction_board.services.snapshot as snapshot_module with tempfile.TemporaryDirectory() as tmp: path = write_config(tmp, cookie="", csrf="") data = json.loads(path.read_text(encoding="utf-8")) data["folders"] = [] data["track_fans"] = [] path.write_text(json.dumps(data, ensure_ascii=False), encoding="utf-8") original = snapshot_module.backup_daily def fail_backup(*args, **kwargs): raise OSError("disk full") snapshot_module.backup_daily = fail_backup try: code, output = run_cli(["snapshot", "--config", str(path)]) finally: snapshot_module.backup_daily = original assert code == 1 marker = output.find("{\n") assert marker >= 0, output payload = json.loads(output[marker:]) assert payload["completed"] is True assert "disk full" in payload["backup_error"] code, output = run_cli(["snapshot", "--config", str(path)]) assert code == 0 marker = output.find("{\n") payload = json.loads(output[marker:]) assert payload["backup_error"] is None assert Path(payload["backup_path"]).exists() def test_candidates_and_discover_commands(): from reaction_board.services.discovery import DiscoverySummary with tempfile.TemporaryDirectory() as tmp: path = write_config(tmp) assert run_cli(["init", "--config", str(path)])[0] == 0 db = Database(Path(tmp) / "app.db") db.initialize() try: repo = Repository(db) repo.upsert_candidate( aid=123, bvid="BV123", title="咻咻满 reaction", up_mid=100, up_name="UP", pubdate=1, cover=None, score=3, channel="search:test#p1", ) finally: db.close() code, output = run_cli(["candidates", "--config", str(path)]) assert code == 0 assert '"count": 1' in output assert '"aid": 123' in output class FakeDiscovery: def __init__(self, *args, **kwargs): pass def run(self): return DiscoverySummary( started_at=1, finished_at=2, candidates=3, notified=1 ) original = cli.DiscoveryService cli.DiscoveryService = FakeDiscovery try: code, output = run_cli(["discover", "--config", str(path)]) finally: cli.DiscoveryService = original assert code == 0 assert '"candidates": 3' in output assert '"notified": 1' in output def test_backfill_followers_dry_run(): with tempfile.TemporaryDirectory() as tmp: path = write_config(tmp, cookie="", csrf="") code, output = run_cli([ "backfill-followers", "--dry-run", "--config", str(path), ]) assert code == 0 assert '"requested": 0' in output assert '"failed": 0' in output assert '"dry_run": true' in output assert '"total": 0' in output def test_help_lists_snapshot_seed(): from reaction_board.cli import _build_parser parser = _build_parser() help_text = parser.format_help() assert "snapshot-seed" in help_text args = parser.parse_args(["snapshot-seed", "--date", "2026-10-31"]) assert args.command == "snapshot-seed" assert args.date == "2026-10-31" def test_snapshot_seed_cli_clones_offline(): from reaction_board.bilibili.models import FavResource with tempfile.TemporaryDirectory() as tmp: path = write_config(tmp, cookie="", csrf="") code, _ = run_cli(["init", "--config", str(path)]) assert code == 0 db = Database(Path(tmp) / "app.db") db.initialize() try: repo = Repository(db) media_id = 11 repo.upsert_from_fav( FavResource( aid=1, bvid="BV1xx411c7mD", title="t", up_mid=9, up_name="u", pubdate=1, duration=10, cover=None, attr=0, fav_time=None, ), media_id, was_added=True, ) yday, today = "2026-10-30", "2026-10-31" repo.upsert_snapshot( yday, 1, metrics={"view": 10, "like": 1, "coin": 0, "favorite": 0, "share": 0, "reply": 0, "danmaku": 0}, ok=True, err=None, captured_at=1_800_000_000, ) repo.kv_set("snapshot:last_completed", yday) repo.kv_set("snapshot:completed:" + yday, "1") finally: db.close() code, output = run_cli([ "snapshot-seed", "--config", str(path), "--date", today, ]) assert code == 0, output assert '"reason": "seeded"' in output db = Database(Path(tmp) / "app.db") db.initialize() try: repo = Repository(db) assert repo.kv_get("snapshot:seeded:" + today) == "1" row = repo.get_snapshot(today, 1) assert row is not None and int(row["view"]) == 10 assert row["src"] == "seed" finally: db.close() def test_snapshot_seed_defers_when_snapshot_running(): with tempfile.TemporaryDirectory() as tmp: path = write_config(tmp, cookie="", csrf="") code, _ = run_cli(["init", "--config", str(path)]) assert code == 0 db = Database(Path(tmp) / "app.db") db.initialize() try: repo = Repository(db) repo.kv_set( "task:running:snapshot", json.dumps({"name": "snapshot", "started_at": 1}), ) repo.kv_set("snapshot:last_completed", "2026-10-30") finally: db.close() code, output = run_cli([ "snapshot-seed", "--config", str(path), "--date", "2026-10-31", ]) assert code == 1, output assert "snapshot_running" in output db = Database(Path(tmp) / "app.db") db.initialize() try: repo = Repository(db) last = json.loads(repo.kv_get("task:last:snapshot_seed")) assert last["ok"] is False assert last["skipped"] is True assert last["error"] == "snapshot_running" assert repo.kv_get("task:running:snapshot_seed") is None assert repo.kv_get("task:running:snapshot") is not None finally: db.close() def test_snapshot_seed_no_source_is_non_success(): with tempfile.TemporaryDirectory() as tmp: path = write_config(tmp, cookie="", csrf="") code, _ = run_cli(["init", "--config", str(path)]) assert code == 0 code, output = run_cli([ "snapshot-seed", "--config", str(path), "--date", "2026-10-31", ]) assert code == 1, output assert "no_source" in output db = Database(Path(tmp) / "app.db") db.initialize() try: repo = Repository(db) last = json.loads(repo.kv_get("task:last:snapshot_seed")) assert last["ok"] is False and last["skipped"] is True assert last["error"] == "no_source" finally: db.close() def test_defer_snapshot_seed_schedules_ten_minutes(): """调度辅助：skip 记非成功，next_run = finished + 600。""" from reaction_board.logging_json import JsonLogger with tempfile.TemporaryDirectory() as tmp: path = write_config(tmp, cookie="", csrf="") code, _ = run_cli(["init", "--config", str(path)]) assert code == 0 db = Database(Path(tmp) / "app.db") db.initialize() try: repo = Repository(db) logger = JsonLogger() started = 1_800_000_000 finished = started + 5 ctrl = cli._defer_snapshot_seed( repo, logger, started_at=started, reason="snapshot_running", finished_at=finished, ) assert ctrl.ok is False assert ctrl.error == "snapshot_running" assert abs(ctrl.next_run - (finished + cli._SEED_RETRY_SECONDS)) < 0.01 last = json.loads(repo.kv_get("task:last:snapshot_seed")) assert last["skipped"] is True and last["ok"] is False finally: db.close() def test_snapshot_seed_same_day_completed_succeeds(): """当日已完成时手动 snapshot-seed 应成功 already_completed，不走 no_source。""" with tempfile.TemporaryDirectory() as tmp: path = write_config(tmp, cookie="", csrf="") code, _ = run_cli(["init", "--config", str(path)]) assert code == 0 today = "2026-10-31" db = Database(Path(tmp) / "app.db") db.initialize() try: repo = Repository(db) repo.kv_set("snapshot:last_completed", today) repo.kv_set("snapshot:completed:" + today, "1") finally: db.close() code, output = run_cli([ "snapshot-seed", "--config", str(path), "--date", today, ]) assert code == 0, output assert '"reason": "already_completed"' in output assert "no_source" not in output db = Database(Path(tmp) / "app.db") db.initialize() try: repo = Repository(db) last = json.loads(repo.kv_get("task:last:snapshot_seed")) assert last["ok"] is True finally: db.close() def test_candidate_cleanup_failure_retries_soon(): """candidate_cleanup 异常应记失败任务并约 5 分钟补跑，而非落到 86400 间隔。""" class FakeClient: def check_login(self): return LoginState(True, 1, "tester", 0, "OK") class FakeDiscovery: def __init__(self, *args, **kwargs): pass def run(self): from reaction_board.services.discovery import DiscoverySummary return DiscoverySummary(started_at=1, finished_at=1) with tempfile.TemporaryDirectory() as tmp: path = write_config(tmp) data = json.loads(path.read_text(encoding="utf-8")) data["folders"] = [] data["track_fans"] = [] path.write_text(json.dumps(data, ensure_ascii=False), encoding="utf-8") original_client = cli._client original_discovery = cli.DiscoveryService original_delete = Repository.delete_expired_candidates captured = {} def boom(self, *args, **kwargs): raise RuntimeError("cleanup db boom") def fake_client(credentials, *args, **kwargs): return FakeClient() # 只跑 cleanup：把其它任务 next_run 推远；通过 Scheduler 包装拦截。 original_scheduler = cli.Scheduler class CaptureScheduler(original_scheduler): def __init__(self, tasks, **kwargs): far = 1_000_000_000_000.0 for task in tasks: if task.name != "candidate_cleanup": task.next_run = far else: task.next_run = 0.0 super().__init__(tasks, **kwargs) captured["scheduler"] = self cli._client = fake_client cli.DiscoveryService = FakeDiscovery cli.Scheduler = CaptureScheduler Repository.delete_expired_candidates = boom try: code, output = run_cli(["run", "--once", "--config", str(path)]) finally: cli._client = original_client cli.DiscoveryService = original_discovery cli.Scheduler = original_scheduler Repository.delete_expired_candidates = original_delete assert code == 0, output db = Database(Path(tmp) / "app.db") db.initialize() try: repo = Repository(db) last = json.loads(repo.kv_get("task:last:candidate_cleanup")) assert last["ok"] is False assert "cleanup db boom" in last["error"] finally: db.close() scheduler = captured["scheduler"] cleanup_task = next(t for t in scheduler.tasks if t.name == "candidate_cleanup") # next_run ≈ finished + 300；允许几秒时钟误差 assert cleanup_task.next_run > 0 delta = cleanup_task.next_run - __import__("time").time() assert 250 <= delta <= 360, delta def test_initial_daily_next_run_uses_wall_clock_not_zero(): """常驻启动：无上次记录时应对齐下一档墙钟，不得为 0。""" from reaction_board.timeutil import next_daily_run, SHANGHAI_TZ from datetime import datetime with tempfile.TemporaryDirectory() as tmp: path = write_config(tmp, cookie="", csrf="") assert run_cli(["init", "--config", str(path)])[0] == 0 db = Database(Path(tmp) / "app.db") db.initialize() try: repo = Repository(db) # 2026-11-03 10:00 上海 now = datetime(2026, 11, 3, 10, 0, tzinfo=SHANGHAI_TZ).timestamp() nxt = cli._initial_daily_next_run( repo, "snapshot", "23:15", now=now, ) assert nxt == next_daily_run(now, "23:15") assert nxt > now # 成功记录在昨晚 → 仍是今晚 23:15 yday_done = datetime(2026, 11, 2, 23, 20, tzinfo=SHANGHAI_TZ).timestamp() repo.kv_set( "task:last:snapshot", json.dumps({ "ok": True, "finished_at": int(yday_done), "started_at": int(yday_done) - 10, }), ) nxt2 = cli._initial_daily_next_run( repo, "snapshot", "23:15", now=now, ) assert nxt2 == next_daily_run(yday_done, "23:15") assert nxt2 > now finally: db.close() def test_reached_daily_slot_gates_force_seed(): from reaction_board.timeutil import SHANGHAI_TZ from datetime import datetime before = datetime(2026, 11, 3, 10, 0, tzinfo=SHANGHAI_TZ).timestamp() at = datetime(2026, 11, 3, 23, 15, tzinfo=SHANGHAI_TZ).timestamp() after = datetime(2026, 11, 3, 23, 16, tzinfo=SHANGHAI_TZ).timestamp() assert cli._reached_daily_slot("23:15", before) is False assert cli._reached_daily_slot("23:15", at) is True assert cli._reached_daily_slot("23:15", after) is True def test_restart_after_seed_schedules_snapshot_at_wall_clock(): """午夜 seed 后重启：snapshot 应排到当晚 snapshot_at，首轮 run_due 不得 force 全量。""" from reaction_board.scheduler import ScheduledTask, Scheduler from reaction_board.timeutil import next_daily_run, SHANGHAI_TZ, shanghai_date from datetime import datetime with tempfile.TemporaryDirectory() as tmp: path = write_config(tmp, cookie="", csrf="") assert run_cli(["init", "--config", str(path)])[0] == 0 db = Database(Path(tmp) / "app.db") db.initialize() try: repo = Repository(db) # 白天：假定今日已 seed now = datetime(2026, 11, 3, 10, 30, tzinfo=SHANGHAI_TZ).timestamp() today = shanghai_date(now) repo.kv_set(f"snapshot:seeded:{today}", "1") repo.kv_set( "task:last:snapshot_seed", json.dumps({ "ok": True, "finished_at": int( datetime(2026, 11, 3, 0, 1, tzinfo=SHANGHAI_TZ).timestamp() ), "started_at": int( datetime(2026, 11, 3, 0, 0, tzinfo=SHANGHAI_TZ).timestamp() ), "summary": {"reason": "seeded"}, }), ) snapshot_at = "23:15" seed_at = "00:00" snapshot_next = cli._initial_daily_next_run( repo, "snapshot", snapshot_at, now=now, ) seed_next = cli._initial_daily_next_run( repo, "snapshot_seed", seed_at, now=now, ) fans_next = cli._initial_daily_next_run( repo, "fans", snapshot_at, now=now, ) assert snapshot_next == next_daily_run(now, snapshot_at) assert fans_next == next_daily_run(now, snapshot_at) # seed 已成功 → 下一档是次日 00:00 assert seed_next == next_daily_run( float(json.loads(repo.kv_get("task:last:snapshot_seed"))["finished_at"]), seed_at, ) assert seed_next > now force_calls = [] def run_snapshot(): force = ( repo.kv_get(f"snapshot:seeded:{today}") == "1" and cli._reached_daily_slot(snapshot_at, now) ) force_calls.append(force) if force: raise AssertionError("daytime restart must not force-seal seed") return None clock = [now] scheduler = Scheduler( [ ScheduledTask("snapshot_seed", 86400.0, lambda: None, next_run=seed_next), ScheduledTask("snapshot", 86400.0, run_snapshot, next_run=snapshot_next), ScheduledTask("fans", 86400.0, lambda: None, next_run=fans_next), ], clock=lambda: clock[0], ) assert scheduler.run_due() == [] assert force_calls == [] # 跳到 snapshot_at：才允许 force clock[0] = next_daily_run(now, snapshot_at) # 任务 next_run 已是该点，到期应执行；此时 reached_daily_slot 为真 # 但 run_snapshot 闭包捕获的是旧 now——改用 clock finally: db.close() # 单独验证：到达 slot 后 force 门打开 slot = datetime(2026, 11, 3, 23, 15, tzinfo=SHANGHAI_TZ).timestamp() assert cli._reached_daily_slot("23:15", slot) is True def test_daytime_seed_marker_does_not_force_without_slot(): """即使 seeded=1，未到 snapshot_at 也不得 force（防守：即便误触发 run_snapshot）。""" from reaction_board.timeutil import SHANGHAI_TZ from datetime import datetime with tempfile.TemporaryDirectory() as tmp: path = write_config(tmp, cookie="", csrf="") assert run_cli(["init", "--config", str(path)])[0] == 0 db = Database(Path(tmp) / "app.db") db.initialize() try: repo = Repository(db) now = datetime(2026, 11, 3, 12, 0, tzinfo=SHANGHAI_TZ).timestamp() today = "2026-11-03" repo.kv_set(f"snapshot:seeded:{today}", "1") force_seed = ( repo.kv_get(f"snapshot:seeded:{today}") == "1" and cli._reached_daily_slot("23:15", now) ) assert force_seed is False finally: db.close() def main_test(): test_init_and_status() test_missing_config() test_ingest_without_refs_is_offline() test_snapshot_offline_empty_db() test_auth_check_without_cookie() test_auth_check_all_and_auth_dispatch() test_run_once_offline() test_serve_preview_guard() test_default_command_starts_web_and_scheduler() test_snapshot_backup_failure_returns_error_and_retries() test_candidates_and_discover_commands() test_backfill_followers_dry_run() test_help_lists_snapshot_seed() test_snapshot_seed_cli_clones_offline() test_snapshot_seed_defers_when_snapshot_running() test_snapshot_seed_no_source_is_non_success() test_snapshot_seed_same_day_completed_succeeds() test_defer_snapshot_seed_schedules_ten_minutes() test_candidate_cleanup_failure_retries_soon() test_initial_daily_next_run_uses_wall_clock_not_zero() test_reached_daily_slot_gates_force_seed() test_restart_after_seed_schedules_snapshot_at_wall_clock() test_daytime_seed_marker_does_not_force_without_slot() print("OK: CLI init/sync/snapshot/snapshot-seed/auth-check/run/status/backfill/错误路径测试全部通过") if __name__ == "__main__": main_test()