# -*- 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 ( # noqa: E402 ConfigError, ConfigStore, default_config_path, load_config, resolve_secrets_path, ) 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 test_jsonc_and_secrets_merge_and_hot_reload(): with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) config_path = root / "config.jsonc" secrets_path = root / "secrets.json" config_path.write_text( """ // 业务配置 { "folders": [ {"year": 2026, "media_id": 1, "title": "A"}, ], "track_fans": [], "up_blacklist_seed": [], "telegram": { "enabled": true }, "web": { "host": "127.0.0.1", "port": 8080 }, } """, encoding="utf-8", ) secrets_path.write_text(json.dumps({ "bilibili": { "read": {"mid": 11, "csrf": "rc", "cookie": "rk"}, "write": {"mid": 22, "csrf": "wc", "cookie": "wk"}, }, "telegram": {"bot_token": "tok", "chat_id": 9}, "web": {"auth_token": "secret-token"}, }), encoding="utf-8") cfg = load_config(config_path, env={}) assert cfg.secrets_path == secrets_path assert cfg.bilibili.read.cookie == "rk" assert cfg.telegram.enabled is True assert cfg.telegram.bot_token == "tok" assert cfg.telegram.chat_id == 9 assert cfg.web.auth_token == "secret-token" assert resolve_secrets_path(config_path) == secrets_path # 主配置注释在热更新凭据后仍保留 original = config_path.read_text(encoding="utf-8") store = ConfigStore(config_path, env={}) secrets = json.loads(secrets_path.read_text(encoding="utf-8")) secrets["bilibili"]["read"]["cookie"] = "rk2" secrets_path.write_text(json.dumps(secrets), encoding="utf-8") bump_mtime(secrets_path) assert store.get().bilibili.read.cookie == "rk2" assert config_path.read_text(encoding="utf-8") == original assert "// 业务配置" in original def test_legacy_monolithic_json_still_loads(): with tempfile.TemporaryDirectory() as tmp: path = Path(tmp) / "config.json" write(path, base_config()) cfg = load_config(path, env={}) assert cfg.bilibili.read.mid == 11 assert cfg.secrets_path == Path(tmp) / "secrets.json" def test_default_config_path_prefers_jsonc(): with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) json_path = root / "config.json" jsonc_path = root / "config.jsonc" json_path.write_text("{}", encoding="utf-8") assert default_config_path(root) == json_path jsonc_path.write_text("{}", encoding="utf-8") assert default_config_path(root) == jsonc_path 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() test_jsonc_and_secrets_merge_and_hot_reload() test_legacy_monolithic_json_still_loads() test_default_config_path_prefers_jsonc() print("OK: 双账号配置加载、JSONC/secrets、拒绝 flat 格式、ENV 覆盖、热加载与 analytics/web.track 测试全部通过") if __name__ == "__main__": main()