kairee
00572a3429
Keep small-base videos on the delta board only, and raise the smoothing prior to 1000 so videos that just clear the floor do not dominate the ranking.
395 lines
15 KiB
Python
395 lines
15 KiB
Python
# -*- 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.analytics.growth_rate_prior == 1000
|
|
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,
|
|
"growth_rate_prior": 2000,
|
|
}
|
|
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.analytics.growth_rate_prior == 2000
|
|
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"),
|
|
({"growth_rate_prior": -1}, "growth_rate_prior"),
|
|
):
|
|
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()
|