refactor(sync): drop favorite-folder tracking and deal
This commit is contained in:
kairee committed
1 parent
62abbd7ea1
commit
6c4287b43a
21 files changed
+1390
-2766
No files matched your search
+352
-352
@@ -1,352 +1,352 @@
|
||||
# -*- 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": 4' 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_sync_requires_auth_without_network():
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
path = write_config(tmp, cookie="", csrf="")
|
||||
code, output = run_cli(["sync", "--config", str(path)])
|
||||
assert code == 2
|
||||
payload = json.loads(output.strip().splitlines()[-1])
|
||||
assert "cookie" in payload["error"]
|
||||
assert not (Path(tmp) / "app.db").exists()
|
||||
|
||||
|
||||
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": "folder_sync"' in output
|
||||
assert '"name": "snapshot"' 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 ("folder_sync", "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, args.allow_empty_folder))
|
||||
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, 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_sync_requires_auth_without_network()
|
||||
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()
|
||||
# -*- 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()
|
||||
Reference in new issue
Block a user