353 lines
12 KiB
Python
353 lines
12 KiB
Python
# -*- 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": 5' 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 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()
|
|
print("OK: CLI init/sync/snapshot/auth-check/run/status/backfill/错误路径测试全部通过")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main_test()
|