252 lines
9.6 KiB
Python
252 lines
9.6 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""只读查询层测试（离线）: uv run python tests/test_web_queries.py"""
|
|
|
|
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.config import ConfigStore # noqa: E402
|
|
from reaction_board.database import Database # noqa: E402
|
|
from reaction_board.web.queries import DashboardQueries # noqa: E402
|
|
from tests.web_fixtures import DATES, build_db, write_config # noqa: E402
|
|
|
|
FIXED_NOW = 1_793_459_000 # 2026-10-31 前后
|
|
|
|
|
|
def setup(tmp: str):
|
|
build_db(tmp)
|
|
config = write_config(tmp)
|
|
db = Database(Path(tmp) / "app.db")
|
|
db.initialize()
|
|
store = ConfigStore(config)
|
|
queries = DashboardQueries(db, store, clock=lambda: FIXED_NOW)
|
|
return db, queries
|
|
|
|
|
|
def test_overview():
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
db, q = setup(tmp)
|
|
try:
|
|
overview = q.overview()
|
|
assert overview["data_date"] == DATES[-1]
|
|
assert overview["snapshot_days"] == 3
|
|
assert overview["is_first_day"] is False
|
|
assert overview["initial_import_count"] == 2
|
|
assert overview["counts"]["active"] == 2
|
|
assert overview["counts"]["unfaved"] == 1
|
|
assert overview["counts"]["removed"] == 1
|
|
assert overview["counts"]["active_ups"] == 2
|
|
assert overview["counts"]["candidates_pending"] == 1
|
|
assert overview["totals"]["view"] == 1200 + 5200 + 120
|
|
assert overview["interactions"] > 0
|
|
assert overview["view_delta"] == (100 + 100 + 0) # 101/102 的日增量
|
|
assert overview["snapshot_health"]["total"] == 3
|
|
assert overview["folders"][0]["media_count"] == 3
|
|
assert overview["folders"][0]["ratio"] < 1
|
|
assert overview["fans"]["follower"] == 1020
|
|
assert overview["fans"]["delta_1d"] == 10
|
|
assert overview["preview"] is False
|
|
assert "snapshot" in overview["tasks"]
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
def test_video_listing_filters_and_sorting():
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
db, q = setup(tmp)
|
|
try:
|
|
heat = q.list_videos(sort="heat")
|
|
aids = [item["aid"] for item in heat["items"]]
|
|
assert 104 not in aids, "removed 视频不应进入榜单"
|
|
assert aids[0] == 102, aids
|
|
assert heat["items"][0]["heat"] > heat["items"][-1]["heat"]
|
|
assert heat["items"][0]["heat_percentile"] == 100.0
|
|
|
|
views = q.list_videos(sort="view")
|
|
assert [item["aid"] for item in views["items"]] == [102, 101, 103]
|
|
|
|
interactions = q.list_videos(sort="interaction")
|
|
assert interactions["items"][0]["aid"] == 101
|
|
|
|
added = q.list_videos(sort="added", within_days=1)
|
|
assert [item["aid"] for item in added["items"]] == [103]
|
|
|
|
published = q.list_videos(sort="published", within_days=2)
|
|
assert {item["aid"] for item in published["items"]} == {101, 103}
|
|
|
|
searched = q.list_videos(sort="view", q="UP B")
|
|
assert [item["aid"] for item in searched["items"]] == [103]
|
|
searched = q.list_videos(sort="view", q="视频 102")
|
|
assert [item["aid"] for item in searched["items"]] == [102]
|
|
|
|
folder = q.list_videos(sort="view", folder_id=3700023351)
|
|
assert folder["total"] == 3
|
|
folder = q.list_videos(sort="view", folder_id=999)
|
|
assert folder["total"] == 0
|
|
|
|
page = q.list_videos(sort="view", page=2, page_size=1)
|
|
assert page["total"] == 3
|
|
assert page["pages"] == 3
|
|
assert [item["aid"] for item in page["items"]] == [101]
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
def test_growth_and_cold_start():
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
db, q = setup(tmp)
|
|
try:
|
|
growth = q.list_videos(sort="growth", growth_days=2)
|
|
assert growth["available"] is True
|
|
assert {item["aid"] for item in growth["items"]} == {101, 102}
|
|
first = next(item for item in growth["items"] if item["aid"] == 101)
|
|
assert first["growth"]["available"] is True
|
|
assert first["growth"]["delta"] == 200
|
|
assert first["growth"]["gap_days"] == 2
|
|
|
|
rate = q.list_videos(sort="growth_rate", growth_days=2)
|
|
assert {item["aid"] for item in rate["items"]} == {101, 102}
|
|
|
|
# 只有一个快照日时增长不可用，不能返回假 0
|
|
for day in DATES[:-1]:
|
|
db.write(lambda conn, day=day: conn.execute(
|
|
"DELETE FROM snapshot WHERE stat_date = ?", (day,)
|
|
))
|
|
cold = q.list_videos(sort="growth", growth_days=2)
|
|
assert cold["available"] is False
|
|
assert cold["items"] == []
|
|
assert "两天" in cold["reason"]
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
def test_video_detail_and_trend():
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
db, q = setup(tmp)
|
|
try:
|
|
detail = q.video_detail(101)
|
|
assert detail["aid"] == 101
|
|
assert detail["metrics"]["view"] == 1200
|
|
assert detail["growth"]["delta"] == 100
|
|
assert detail["snapshot_count"] == 3
|
|
assert detail["folder_title"] == "2026"
|
|
assert detail["url"].endswith("BV101")
|
|
assert q.video_detail(999999) is None
|
|
|
|
trend = q.video_trend(101, days=3)
|
|
assert trend["dates"] == list(DATES)
|
|
assert trend["series"]["view"] == [1000, 1100, 1200]
|
|
assert trend["available"] is True
|
|
assert len(trend["points"]) == 3
|
|
|
|
# 请求区间早于首次快照时，从首次快照开始，不制造空白历史
|
|
wide = q.video_trend(101, days=30)
|
|
assert wide["dates"] == list(DATES)
|
|
|
|
# 删除中间一天，折线在缺失日期应为 None，而不是插值
|
|
db.write(lambda conn: conn.execute(
|
|
"DELETE FROM snapshot WHERE aid=101 AND stat_date=?",
|
|
("2026-10-30",),
|
|
))
|
|
trend = q.video_trend(101, days=3)
|
|
assert trend["series"]["view"] == [1000, None, 1200]
|
|
|
|
# 失败快照也视为缺失
|
|
db.write(lambda conn: conn.execute(
|
|
"UPDATE snapshot SET ok=0, view=NULL WHERE aid=101 AND stat_date=?",
|
|
(DATES[-1],),
|
|
))
|
|
trend = q.video_trend(101, days=3)
|
|
assert trend["series"]["view"][-1] is None
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
def test_up_queries():
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
db, q = setup(tmp)
|
|
try:
|
|
ups = q.list_ups()
|
|
assert ups["total"] == 2
|
|
assert ups["items"][0]["up_mid"] == 1
|
|
assert ups["items"][0]["video_count"] == 2
|
|
assert ups["items"][0]["total_views"] == 6400
|
|
assert ups["items"][0]["avg_views"] == 3200.0
|
|
|
|
one = q.list_ups(min_videos=2)
|
|
assert one["total"] == 1
|
|
|
|
detail = q.up_detail(1)
|
|
assert detail["video_count"] == 2
|
|
assert detail["total_views"] == 6400
|
|
assert detail["blacklisted"] is False
|
|
assert detail["series"]["dates"] == list(DATES)
|
|
assert detail["series"]["views"] == [6000, 6200, 6400]
|
|
assert detail["space_url"].endswith("/1")
|
|
|
|
blacklisted = q.up_detail(2)
|
|
assert blacklisted["blacklisted"] is True
|
|
assert blacklisted["video_count"] == 1
|
|
assert q.up_detail(999999) is None
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
def test_fans_candidates_system_and_export():
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
db, q = setup(tmp)
|
|
try:
|
|
fans = q.fans(days=3)
|
|
assert fans["mid"] == 37754047
|
|
assert fans["name"] == "咻咻满"
|
|
assert fans["dates"] == list(DATES)
|
|
assert fans["series"] == [1000, 1010, 1020]
|
|
assert fans["delta_1d"] == 10
|
|
assert fans["delta_7d"] is None
|
|
assert fans["delta_cum"] == 20
|
|
assert fans["options"][0]["mid"] == 37754047
|
|
wide = q.fans(days=30)
|
|
assert wide["dates"] == list(DATES)
|
|
|
|
candidates = q.candidates()
|
|
assert candidates["total"] == 2
|
|
assert candidates["pending"] == 1
|
|
assert candidates["items"][0]["aid"] == 201
|
|
assert 1 <= candidates["items"][0]["days_left"] <= 7
|
|
assert candidates["items"][1]["decision"] == "rejected"
|
|
|
|
system = q.system()
|
|
assert system["data_date"] == DATES[-1]
|
|
assert system["snapshot_days"] == 3
|
|
assert system["blacklist"][0]["video_count"] == 1
|
|
assert system["tasks"]["snapshot"]["ok"] is True
|
|
assert system["errors"] == []
|
|
assert system["preview"] is False
|
|
assert system["next_runs"]["snapshot"] > 0
|
|
assert system["next_runs"]["folder_sync"] > FIXED_NOW
|
|
assert system["next_runs"]["discover"] > FIXED_NOW
|
|
|
|
exported = q.export_videos(sort="view")
|
|
assert exported["total"] == 3
|
|
assert all("metrics" in item for item in exported["items"])
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
def main_test():
|
|
test_overview()
|
|
test_video_listing_filters_and_sorting()
|
|
test_growth_and_cold_start()
|
|
test_video_detail_and_trend()
|
|
test_up_queries()
|
|
test_fans_candidates_system_and_export()
|
|
print("OK: Web 查询层总览/榜单/详情/UP/粉丝/候选/系统测试全部通过")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main_test()
|