# -*- coding: utf-8 -*- """双账号配置加载 / 校验 / 热加载测试（不联网）: uv run python tests/test_config.py""" import json import os import sys import tempfile import time 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.config import ConfigError, ConfigStore, load_config # noqa: E402 def write(path: Path, data) -> None: path.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8") def bump_mtime(path: Path) -> None: target = max(path.stat().st_mtime_ns, time.time_ns()) + 1_000_000 os.utime(path, ns=(target, target)) def base_config() -> dict: return { "bilibili": { "read": {"mid": 11, "csrf": "read-csrf", "cookie": "read-cookie"}, "write": {"mid": 22, "csrf": "write-csrf", "cookie": "write-cookie"}, }, "overrides": {"1": {"mid": 2, "name": "x"}}, "sleep": {"min": 1}, "folders": [ {"year": 2025, "media_id": 4451053851, "title": "2025"}, {"year": 2026, "media_id": 3700023351, "title": "2026"}, ], "track_fans": [37754047], "up_blacklist_seed": [1, 2, 2], } def test_dual_accounts_and_env_override(): with tempfile.TemporaryDirectory() as tmp: path = Path(tmp) / "config.json" write(path, base_config()) env = { "BILI_READ_MID": "101", "BILI_READ_CSRF": "env-read", "BILI_READ_COOKIE": "env-read-cookie", "BILI_WRITE_MID": "202", "BILI_WRITE_CSRF": "env-write", "BILI_WRITE_COOKIE": "env-write-cookie", "DATABASE_PATH": "custom.db", } cfg = load_config(path, env=env) assert (cfg.bilibili.read.mid, cfg.bilibili.read.csrf, cfg.bilibili.read.cookie) == (101, "env-read", "env-read-cookie") assert (cfg.bilibili.write.mid, cfg.bilibili.write.csrf, cfg.bilibili.write.cookie) == (202, "env-write", "env-write-cookie") assert cfg.storage.database_path == Path(tmp) / "custom.db" assert cfg.up_blacklist_seed == (1, 2) assert cfg.discover.keywords == ("咻咻满",) assert cfg.discover.pages == 3 assert cfg.discover.candidate_min_score == 4 assert cfg.classify["self_mid"] == 37754047 assert cfg.classify["reaction"] == ["re", "reaction", "反应"] assert cfg.classify["date_negative"] is True def test_flat_and_missing_account_rejected(): with tempfile.TemporaryDirectory() as tmp: path = Path(tmp) / "config.json" write(path, {"mid": 1, "csrf": "x", "cookie": "y", "folders": []}) try: load_config(path, env={}) except ConfigError as exc: assert "bilibili.read" in str(exc) else: raise AssertionError("flat 格式必须拒绝") write(path, {"bilibili": {"read": {}}, "folders": []}) try: load_config(path, env={}) except ConfigError as exc: assert "bilibili.write" in str(exc) else: raise AssertionError("缺少 write 必须拒绝") def test_validation_require_read_auth_and_hot_reload(): with tempfile.TemporaryDirectory() as tmp: path = Path(tmp) / "config.json" data = base_config() data["bilibili"]["read"].update(cookie="", csrf="") write(path, data) load_config(path, env={}, require_auth=False) try: load_config(path, env={}, require_auth=True) except ConfigError as exc: assert "bilibili.read.cookie" in str(exc) else: raise AssertionError("read 空凭据必须拒绝") write(path, base_config()) store = ConfigStore(path, env={}) data = base_config() data["folders"][0]["title"] = "2025-new" write(path, data) bump_mtime(path) assert store.get().folders[0].title == "2025-new" path.write_text("{bad", encoding="utf-8") bump_mtime(path) assert store.get().folders[0].title == "2025-new" assert store.last_error def test_invalid_values(): with tempfile.TemporaryDirectory() as tmp: path = Path(tmp) / "config.json" data = base_config() data["bilibili"]["write"]["mid"] = "bad" write(path, data) try: load_config(path, env={}) except ConfigError as exc: assert "bilibili.write.mid" in str(exc) else: raise AssertionError("非法 mid 必须拒绝") write(path, base_config()) try: load_config(path, env={"BILI_READ_MID": "bad"}) except ConfigError as exc: assert "BILI_READ_MID" in str(exc) else: raise AssertionError("非法环境变量必须拒绝") def test_analytics_defaults_and_validation(): with tempfile.TemporaryDirectory() as tmp: path = Path(tmp) / "config.json" write(path, base_config()) cfg = load_config(path, env={}) assert cfg.analytics.min_duration_seconds == 600 assert cfg.analytics.maturity_days == 2 assert cfg.analytics.min_view_for_ranking == 2000 assert cfg.analytics.video_shrink_minutes == 60 assert cfg.analytics.shrink_minutes is None assert cfg.analytics.home_growth_days == "auto" assert cfg.schedule.recent_snapshot_minutes == 60 assert cfg.schedule.recent_video_days == 2 assert cfg.schedule.snapshot_at == "23:15" assert cfg.schedule.seed_at == "00:00" data = base_config() data["analytics"] = { "min_duration_seconds": 300, "maturity_days": 7, "min_view_for_ranking": 100, "video_shrink_minutes": 30, "shrink_minutes": 90, "home_growth_days": 1, } data["schedule"] = { "discover_minutes": 30, "snapshot_at": "23:15", "seed_at": "00:00", "candidate_ttl_days": 7, "recent_snapshot_minutes": 15, "recent_video_days": 3, } write(path, data) cfg = load_config(path, env={}) assert cfg.analytics.min_duration_seconds == 300 assert cfg.analytics.maturity_days == 7 assert cfg.analytics.min_view_for_ranking == 100 assert cfg.analytics.video_shrink_minutes == 30 assert cfg.analytics.shrink_minutes == 90 assert cfg.analytics.home_growth_days == 1 assert cfg.schedule.recent_snapshot_minutes == 15 assert cfg.schedule.recent_video_days == 3 data = base_config() data["analytics"] = {"home_growth_days": 7} write(path, data) assert load_config(path, env={}).analytics.home_growth_days == 7 for bad, hint in ( ({"min_duration_seconds": -1}, "min_duration_seconds"), ({"maturity_days": -1}, "maturity_days"), ({"min_view_for_ranking": -1}, "min_view_for_ranking"), ({"video_shrink_minutes": -1}, "video_shrink_minutes"), ({"shrink_minutes": 0}, "shrink_minutes"), ({"shrink_minutes": "auto"}, "shrink_minutes"), ({"home_growth_days": 3}, "home_growth_days"), ({"home_growth_days": "7"}, "home_growth_days"), ({"home_growth_days": 0}, "home_growth_days"), ): data = base_config() data["analytics"] = bad write(path, data) try: load_config(path, env={}) except ConfigError as exc: assert hint in str(exc), (bad, str(exc)) else: raise AssertionError(f"非法 analytics 必须拒绝: {bad}") def test_web_track_defaults_and_validation(): with tempfile.TemporaryDirectory() as tmp: path = Path(tmp) / "config.json" write(path, base_config()) cfg = load_config(path, env={}) track = cfg.web.track assert track.enabled is True assert track.header_order == ("X-Real-IP", "X-Forwarded-For") assert "172.16.0.0/12" in track.trusted_proxies assert track.count_bots is False assert track.flush_seconds == 60 assert track.keep_days == 180 assert track.max_hits_per_ip_path_per_day == 0 data = base_config() data["web"] = { "host": "0.0.0.0", "port": 8080, "auth_token": "t", "public_base_url": "", "track": { "enabled": False, "header_order": ["X-Forwarded-For", "X-Real-IP", "X-Real-IP"], "trusted_proxies": ["10.0.0.0/8", "192.168.1.5"], "count_bots": True, "flush_seconds": 0, "keep_days": 30, "max_hits_per_ip_path_per_day": 500, }, } write(path, data) cfg = load_config(path, env={}) assert cfg.web.track.enabled is False # 去重且保持顺序 assert cfg.web.track.header_order == ("X-Forwarded-For", "X-Real-IP") assert cfg.web.track.trusted_proxies == ("10.0.0.0/8", "192.168.1.5/32") assert cfg.web.track.count_bots is True assert cfg.web.track.flush_seconds == 0 assert cfg.web.track.max_hits_per_ip_path_per_day == 500 for bad, hint in ( ({"header_order": []}, "header_order"), ({"header_order": ["X-Real-IP", 5]}, "header_order"), ({"trusted_proxies": ["not-a-cidr"]}, "trusted_proxies"), ({"flush_seconds": -1}, "flush_seconds"), ({"keep_days": 0}, "keep_days"), ({"max_hits_per_ip_path_per_day": -1}, "max_hits_per_ip_path_per_day"), ): data = base_config() data["web"] = {"track": bad} write(path, data) try: load_config(path, env={}) except ConfigError as exc: assert hint in str(exc), (bad, str(exc)) else: raise AssertionError(f"非法 web.track 必须拒绝: {bad}") # web.track 不是对象 data = base_config() data["web"] = {"track": "nope"} write(path, data) try: load_config(path, env={}) except ConfigError as exc: assert "web.track" in str(exc) else: raise AssertionError("web.track 必须是对象") def test_schedule_recent_validation(): with tempfile.TemporaryDirectory() as tmp: path = Path(tmp) / "config.json" for bad, hint in ( ({"recent_snapshot_minutes": 0}, "recent_snapshot_minutes"), ({"recent_video_days": 0}, "recent_video_days"), ({"recent_snapshot_minutes": -5}, "recent_snapshot_minutes"), ): data = base_config() data["schedule"] = { "discover_minutes": 60, "snapshot_at": "23:15", "seed_at": "00:00", "candidate_ttl_days": 7, **bad, } write(path, data) try: load_config(path, env={}) except ConfigError as exc: assert hint in str(exc), (bad, str(exc)) else: raise AssertionError(f"非法 schedule 必须拒绝: {bad}") def main(): test_dual_accounts_and_env_override() test_flat_and_missing_account_rejected() test_validation_require_read_auth_and_hot_reload() test_invalid_values() test_analytics_defaults_and_validation() test_schedule_recent_validation() test_web_track_defaults_and_validation() print("OK: 双账号配置加载、拒绝 flat 格式、ENV 覆盖、热加载与 analytics/web.track 测试全部通过") if __name__ == "__main__": main()