# -*- coding: utf-8 -*- """收藏/忽略/拉黑动作测试（假写客户端，不联网）: uv run python tests/test_favorites.py""" import json import os import sys import tempfile from dataclasses import dataclass 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.client import BilibiliAPIError # noqa: E402 from reaction_board.bilibili.models import FavResource # noqa: E402 from reaction_board.config import ConfigStore # noqa: E402 from reaction_board.database import Database # noqa: E402 from reaction_board.repository import Repository # noqa: E402 from reaction_board.services.favorites import FavoriteActionService # noqa: E402 NOW = 1_793_444_393 FOLDER_ID = 3700023351 @dataclass class FakeFavId: aid: int type: int = 2 class FakeWriteClient: def __init__(self, order, *, fail_deal=False): self.order = list(order) self.fail_deal = fail_deal self.dealt = [] self.sorts = [] def fav_ids(self, media_id): return [FakeFavId(aid) for aid in self.order] def deal(self, aid, folder_id): if self.fail_deal: raise BilibiliAPIError(-1, "收藏夹已满") self.dealt.append((aid, folder_id)) self.order.append(aid) def sort(self, media_id, spec): predecessor, _, moved, _ = (int(part) for part in spec.split(":")) self.sorts.append(spec) self.order.remove(moved) index = self.order.index(predecessor) self.order.insert(index + 1, moved) class FakeProvider: def __init__(self, client): self.client = client def get(self): return self.client def write_config(tmp): data = { "bilibili": { "read": {"mid": 1, "csrf": "r", "cookie": "read-cookie"}, "write": {"mid": 2, "csrf": "w", "cookie": "write-cookie"}, }, "storage": { "database_path": str(Path(tmp) / "app.db"), "backup_dir": str(Path(tmp) / "backup"), }, "folders": [{"year": 2026, "media_id": FOLDER_ID, "title": "2026"}], "discover": {"keywords": ["x"], "pages": 1, "window_days": 14}, } path = Path(tmp) / "config.json" path.write_text(json.dumps(data, ensure_ascii=False), encoding="utf-8") return path def add_video(repo, aid, up_mid, pubdate): repo.upsert_from_fav( FavResource( aid=aid, bvid=f"BV{aid}", title=f"视频{aid}", up_mid=up_mid, up_name=f"UP{up_mid}", pubdate=pubdate, duration=600, cover=None, attr=0, fav_time=None, ), FOLDER_ID, was_added=True, ) def add_candidate(repo, aid, up_mid=100, pubdate=NOW - 3600): return repo.upsert_candidate( aid=aid, bvid=f"BV{aid}", title=f"咻咻满 reaction {aid}", up_mid=up_mid, up_name=f"UP{up_mid}", pubdate=pubdate, cover="https://example/cover.jpg", score=3, channel="search:test#p1", now=NOW, ) def setup(tmp, order, *, fail_deal=False): path = write_config(tmp) store = ConfigStore(path, env={}) db = Database(Path(tmp) / "app.db", backup_dir=Path(tmp) / "backup") db.initialize() repo = Repository(db, clock=lambda: NOW) add_video(repo, 1, 100, 1_792_000_000) add_video(repo, 2, 200, 1_791_000_000) add_candidate(repo, 9, up_mid=100, pubdate=1_793_000_000) client = FakeWriteClient(order, fail_deal=fail_deal) service = FavoriteActionService( store, repo, FakeProvider(client), clock=lambda: NOW ) return db, repo, client, service def test_favorite_inserts_and_positions(): with tempfile.TemporaryDirectory() as tmp: db, repo, client, service = setup(tmp, [1, 2]) try: result = service.favorite(9) assert result.ok is True assert result.position == 1 assert result.already_in_folder is False assert client.dealt == [(9, FOLDER_ID)] assert client.sorts == ["1:2:9:2", "9:2:1:2"] assert client.order == [9, 1, 2] assert repo.get_candidate(9) is None video = repo.get_video(9) assert video["status"] == "active" assert video["folder_id"] == FOLDER_ID action = repo.get_fav_action("fav:9") assert action["state"] == "done" assert action["position"] == 1 finally: db.close() def test_favorite_already_in_folder_skips_writes(): with tempfile.TemporaryDirectory() as tmp: db, repo, client, service = setup(tmp, [1, 9, 2]) try: result = service.favorite(9) assert result.ok is True assert result.already_in_folder is True assert result.position == 2 assert client.dealt == [] assert client.sorts == [] assert repo.get_candidate(9) is None finally: db.close() def test_favorite_failure_keeps_candidate_and_allows_retry(): with tempfile.TemporaryDirectory() as tmp: db, repo, client, service = setup(tmp, [1, 2], fail_deal=True) try: result = service.favorite(9) assert result.ok is False assert "收藏夹已满" in result.message assert repo.get_candidate(9) is not None action = repo.get_fav_action("fav:9") assert action["state"] == "failed" assert "收藏夹已满" in action["err"] client.fail_deal = False retry = service.favorite(9) assert retry.ok is True assert retry.position == 1 assert repo.get_candidate(9) is None finally: db.close() def test_ignore_and_blacklist(): with tempfile.TemporaryDirectory() as tmp: db, repo, client, service = setup(tmp, [1, 2]) try: add_candidate(repo, 11, up_mid=300) add_candidate(repo, 12, up_mid=300) ignored = service.ignore(11) assert ignored.ok is True assert repo.get_candidate(11)["decision"] == "rejected" result = service.blacklist(12) assert result.ok is True assert repo.is_blacklisted(300) assert repo.get_candidate(12)["decision"] == "rejected" # 已有视频不受拉黑影响 assert repo.get_video(1)["status"] == "active" finally: db.close() def main(): test_favorite_inserts_and_positions() test_favorite_already_in_folder_skips_writes() test_favorite_failure_keeps_candidate_and_allows_retry() test_ignore_and_blacklist() print("OK: 收藏、忽略、拉黑与失败重试测试全部通过") if __name__ == "__main__": main()