557 lines
23 KiB
Python
557 lines
23 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.repository import Repository # noqa: E402
|
||
from reaction_board.timeutil import day_bounds # noqa: E402
|
||
from reaction_board.web.cache import DASHBOARD_CACHE # 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 overview["years"] == [2026]
|
||
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 视频不应进入榜单"
|
||
# 综合热度榜应用最小播放门槛 2000：101（1200）与 103（120）被排除
|
||
assert aids == [102], aids
|
||
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,)
|
||
))
|
||
DASHBOARD_CACHE.clear()
|
||
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["sort"] == "avg_views"
|
||
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
|
||
assert ups["items"][0]["up_face"] == "https://i0.hdslb.com/bfs/face/a.jpg"
|
||
assert ups["items"][0]["follower_latest"] == 12345
|
||
|
||
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["up_face"] == "https://i0.hdslb.com/bfs/face/a.jpg"
|
||
assert detail["series"]["dates"] == list(DATES)
|
||
assert detail["series"]["views"] == [6000, 6200, 6400]
|
||
assert detail["space_url"].endswith("/1")
|
||
assert "folders" not in detail
|
||
assert detail["first_pubdate"] > 0
|
||
|
||
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_query_cache_invalidation():
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
db, q = setup(tmp)
|
||
try:
|
||
DASHBOARD_CACHE.clear()
|
||
first = q.overview()
|
||
db.write(lambda conn: conn.execute(
|
||
"UPDATE snapshot SET \"view\" = 999999 WHERE aid=101 AND stat_date=?",
|
||
(DATES[-1],),
|
||
))
|
||
cached = q.overview()
|
||
assert cached["totals"]["view"] == first["totals"]["view"]
|
||
|
||
repo = Repository(db)
|
||
repo.bump_cache_version("videos")
|
||
refreshed = q.overview()
|
||
assert refreshed["totals"]["view"] > first["totals"]["view"]
|
||
|
||
fans_first = q.fans(days=3)
|
||
db.write(lambda conn: conn.execute(
|
||
"UPDATE up_stat SET follower=999999 WHERE mid=37754047 AND stat_date=?",
|
||
(DATES[-1],),
|
||
))
|
||
fans_cached = q.fans(days=3)
|
||
assert fans_cached["follower"] == fans_first["follower"]
|
||
|
||
repo.bump_cache_version("fans")
|
||
fans_refreshed = q.fans(days=3)
|
||
assert fans_refreshed["follower"] == 999999
|
||
finally:
|
||
db.close()
|
||
|
||
|
||
def test_fans_custom_range_modes():
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
db, q = setup(tmp)
|
||
try:
|
||
daily = q.fans(start="2026-10-29", end="2026-10-31")
|
||
assert daily["mode"] == "daily"
|
||
assert daily["range_start"] == "2026-10-29"
|
||
assert daily["range_end"] == "2026-10-31"
|
||
assert daily["dates"] == list(DATES)
|
||
assert daily["series"] == [1000, 1010, 1020]
|
||
|
||
# 超过一年触发按月峰值模式
|
||
db.write(lambda conn: conn.execute(
|
||
"INSERT OR REPLACE INTO up_stat"
|
||
"(stat_date, mid, follower, following, captured_at, ok) "
|
||
"VALUES ('2025-01-01', 37754047, 500, NULL, 1, 1)"
|
||
))
|
||
DASHBOARD_CACHE.clear()
|
||
monthly = q.fans(start="2025-01-01", end="2026-10-31")
|
||
assert monthly["mode"] == "monthly"
|
||
assert monthly["span_days"] == 669
|
||
assert monthly["dates"][0] == "2025-01-01"
|
||
assert monthly["series"][0] == 500
|
||
assert monthly["dates"][-1] == "2026-10-31"
|
||
assert monthly["series"][-1] == 1020
|
||
|
||
# 同月多条时取峰值日期，并优先较晚日期
|
||
db.write(lambda conn: conn.execute(
|
||
"INSERT OR REPLACE INTO up_stat"
|
||
"(stat_date, mid, follower, following, captured_at, ok) "
|
||
"VALUES ('2026-10-15', 37754047, 5000, NULL, 1, 1), "
|
||
"('2026-10-16', 37754047, 5000, NULL, 1, 1)"
|
||
))
|
||
DASHBOARD_CACHE.clear()
|
||
monthly = q.fans(start="2025-01-01", end="2026-10-31")
|
||
assert monthly["dates"][-1] == "2026-10-16"
|
||
assert monthly["series"][-1] == 5000
|
||
|
||
try:
|
||
q.fans(start="2026-11-01", end="2026-10-01")
|
||
except ValueError:
|
||
pass
|
||
else:
|
||
raise AssertionError("start 晚于 end 应报错")
|
||
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)
|
||
|
||
# “最近记录”只返回最近 14 天，早于窗口的记录不进入表格
|
||
db.write(lambda conn: conn.execute(
|
||
"INSERT OR REPLACE INTO up_stat"
|
||
"(stat_date, mid, follower, following, captured_at, ok) "
|
||
"VALUES (?, ?, ?, NULL, 1, 1)",
|
||
("2026-10-01", 37754047, 900),
|
||
))
|
||
DASHBOARD_CACHE.clear()
|
||
recent = q.fans(days=90)
|
||
assert recent["recent_days"] == 14
|
||
assert [point["stat_date"] for point in recent["recent_points"]] == list(DATES)
|
||
deltas = {point["stat_date"]: point["delta"] for point in recent["recent_points"]}
|
||
assert deltas["2026-10-29"] == 100 # 对比 2026-10-01 的 900
|
||
assert deltas["2026-10-30"] == 10
|
||
assert deltas["2026-10-31"] == 10
|
||
one_day = q.fans(days=90, recent_days=1)
|
||
assert [point["stat_date"] for point in one_day["recent_points"]] == [DATES[-1]]
|
||
|
||
# 缺失前一天时不能把跨多天的变化标成“日增”
|
||
db.write(lambda conn: conn.execute(
|
||
"DELETE FROM up_stat WHERE mid=37754047 AND stat_date=?",
|
||
("2026-10-30",),
|
||
))
|
||
DASHBOARD_CACHE.clear()
|
||
gapped = q.fans(days=3)
|
||
assert gapped["delta_1d"] is None
|
||
assert gapped["delta_7d"] is None
|
||
|
||
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"]["discover"] > FIXED_NOW
|
||
finally:
|
||
db.close()
|
||
|
||
|
||
def test_duration_efficiency_queries():
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
db, q = setup(tmp)
|
||
try:
|
||
db.write(lambda conn: conn.executemany(
|
||
"UPDATE video SET duration = ?, pubdate = ? WHERE aid = ?",
|
||
[
|
||
(9000, 1_756_684_800, 101), # 2025-09-01 前后，2.5 小时
|
||
(7200, 1_756_771_200, 102), # 2 小时
|
||
(300, 1_756_857_600, 103), # 5 分钟，应被排除
|
||
],
|
||
))
|
||
DASHBOARD_CACHE.clear()
|
||
|
||
efficiency = q.list_videos(sort="efficiency")
|
||
assert [item["aid"] for item in efficiency["items"]] == [102]
|
||
first = efficiency["items"][0]
|
||
assert first["duration"] == 7200
|
||
assert first["content_hours"] == 2.0
|
||
assert first["efficiency"] == 2600.0 # 原始播放/小时，详情页口径
|
||
assert abs(first["efficiency_score"] - 2207.5) < 0.1
|
||
assert efficiency["duration_filter"] == {
|
||
"min_duration_seconds": 600, "maturity_days": 2,
|
||
"min_view": 2000, "shrink_minutes": 60,
|
||
}
|
||
|
||
# 播放不足 2000 的视频不进入效率榜（101 只有 1200）
|
||
assert all(item["aid"] != 101 for item in efficiency["items"])
|
||
|
||
interaction = q.list_videos(sort="interaction_efficiency")
|
||
assert [item["aid"] for item in interaction["items"]] == [102]
|
||
assert abs(interaction["items"][0]["interaction_efficiency_score"] - 159.36) < 0.1
|
||
|
||
# 综合热度榜也应用最小播放门槛
|
||
heat = q.list_videos(sort="heat")
|
||
assert [item["aid"] for item in heat["items"]] == [102]
|
||
|
||
# 时长缺失不计入时长效率榜
|
||
ups = q.list_ups(sort="efficiency")
|
||
assert ups["sort"] == "efficiency"
|
||
up_a = next(item for item in ups["items"] if item["up_mid"] == 1)
|
||
assert up_a["total_duration"] == int(9000 + 7200)
|
||
assert up_a["content_hours"] == 4.5
|
||
assert up_a["views_per_hour"] == round(6400 / 4.5, 1)
|
||
assert up_a["efficiency"] is not None
|
||
assert up_a["interaction_efficiency"] is not None
|
||
|
||
by_duration = q.list_ups(sort="duration")
|
||
assert [item["up_mid"] for item in by_duration["items"]][0] == 1
|
||
|
||
db.write(lambda conn: conn.execute(
|
||
"UPDATE video SET duration = NULL WHERE aid = 101"
|
||
))
|
||
DASHBOARD_CACHE.clear()
|
||
missing = q.list_videos(sort="efficiency")
|
||
assert [item["aid"] for item in missing["items"]] == [102]
|
||
|
||
# 发布未满 2 天的视频不进入效率榜
|
||
db.write(lambda conn: conn.execute(
|
||
"UPDATE video SET duration = 7200, pubdate = ? WHERE aid = 102",
|
||
(int(day_bounds("2026-10-30")[0]),),
|
||
))
|
||
DASHBOARD_CACHE.clear()
|
||
fresh = q.list_videos(sort="efficiency")
|
||
assert fresh["items"] == []
|
||
assert fresh["available"] is False
|
||
assert "发布满 2 天" in fresh["reason"]
|
||
assert "播放 ≥ 2000" in fresh["reason"]
|
||
finally:
|
||
db.close()
|
||
|
||
|
||
def test_year_filter_and_published_sort():
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
db, q = setup(tmp)
|
||
try:
|
||
by_year = q.list_videos(sort="view", year=2026)
|
||
assert [item["aid"] for item in by_year["items"]] == [102, 101, 103]
|
||
assert by_year["year"] == 2026
|
||
|
||
none = q.list_videos(sort="view", year=2025)
|
||
assert none["items"] == []
|
||
assert none["total"] == 0
|
||
|
||
published = q.list_videos(sort="published")
|
||
assert [item["aid"] for item in published["items"]] == [103, 101, 102]
|
||
|
||
added = q.list_videos(sort="added")
|
||
assert added["total"] == 3
|
||
|
||
ups = q.list_ups(sort="efficiency", year=2026)
|
||
assert ups["year"] == 2026 and ups["total"] == 2
|
||
ups_none = q.list_ups(sort="views", year=2025)
|
||
assert ups_none["total"] == 0
|
||
finally:
|
||
db.close()
|
||
|
||
|
||
def test_about_payload():
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
db, q = setup(tmp)
|
||
try:
|
||
about = q.about()
|
||
assert about["data_date"] == DATES[-1]
|
||
assert about["snapshot_days"] == 3
|
||
assert about["snapshot_first_date"] == DATES[0]
|
||
assert about["fans_start_date"] == DATES[0]
|
||
assert about["schema_version"] == 7
|
||
assert about["analytics"] == {
|
||
"min_duration_seconds": 600,
|
||
"maturity_days": 2,
|
||
"min_view_for_ranking": 2000,
|
||
"video_shrink_minutes": 60,
|
||
"shrink_minutes": None,
|
||
}
|
||
assert about["heat_weights"]["view"] == 0.30
|
||
assert about["followers"]["videos_total"] == 3
|
||
assert about["followers"]["videos_covered"] == 3
|
||
assert about["followers"]["videos_missing"] == 0
|
||
assert about["folders"][0]["media_id"] == 3700023351
|
||
assert about["preview"] is False
|
||
finally:
|
||
db.close()
|
||
|
||
|
||
def test_stats_query():
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
db, q = setup(tmp)
|
||
try:
|
||
empty = q.stats(days=7)
|
||
assert empty["today"] == "2026-10-31"
|
||
assert empty["today_pv"] == 0 and empty["today_uv"] == 0
|
||
assert len(empty["series"]["dates"]) == 7
|
||
assert empty["series"]["dates"][-1] == "2026-10-31"
|
||
assert empty["total_pv"] == 0 and empty["total_uv"] == 0
|
||
assert empty["top_pages"] == []
|
||
assert empty["enabled"] is True
|
||
assert empty["pending"] is False
|
||
|
||
def insert(conn, day, ip, path, hits, first, last):
|
||
conn.execute(
|
||
"INSERT INTO visit(stat_date, ip, path, hits, first_at, last_at) "
|
||
"VALUES (?, ?, ?, ?, ?, ?)",
|
||
(day, ip, path, hits, first, last),
|
||
)
|
||
|
||
db.write(lambda conn: [
|
||
insert(conn, "2026-10-31", "1.1.1.1", "/home", 3, 1, 2),
|
||
insert(conn, "2026-10-31", "2.2.2.2", "/home", 1, 1, 1),
|
||
insert(conn, "2026-10-31", "1.1.1.1", "/videos", 2, 1, 1),
|
||
insert(conn, "2026-10-30", "1.1.1.1", "/home", 5, 1, 1),
|
||
insert(conn, "2026-09-01", "9.9.9.9", "/home", 100, 1, 1),
|
||
])
|
||
stats = q.stats(days=7)
|
||
assert stats["today_pv"] == 6
|
||
assert stats["today_uv"] == 2
|
||
assert stats["total_pv"] == 11 # 今天 6 + 昨天 5，区间外不计
|
||
assert stats["total_uv"] == 2
|
||
assert stats["series"]["pv"][-1] == 6
|
||
assert stats["series"]["pv"][-2] == 5
|
||
assert stats["series"]["uv"][-1] == 2
|
||
pages = {page["path"]: page for page in stats["top_pages"]}
|
||
assert pages["/home"]["pv"] == 9 and pages["/home"]["uv"] == 2
|
||
assert pages["/videos"]["pv"] == 2 and pages["/videos"]["uv"] == 1
|
||
assert stats["top_pages"][0]["path"] == "/home"
|
||
|
||
# 内存缓冲合并：不同键相加，重叠键去重合并
|
||
pending = [
|
||
{"stat_date": "2026-10-31", "ip": "3.3.3.3",
|
||
"path": "/about", "hits": 4, "first_at": 1, "last_at": 1},
|
||
{"stat_date": "2026-10-31", "ip": "1.1.1.1",
|
||
"path": "/home", "hits": 1, "first_at": 3, "last_at": 3},
|
||
]
|
||
merged = q.stats(days=7, pending=pending)
|
||
assert merged["today_pv"] == 11
|
||
assert merged["today_uv"] == 3
|
||
assert merged["pending"] is True
|
||
pages = {page["path"]: page for page in merged["top_pages"]}
|
||
assert pages["/about"]["pv"] == 4
|
||
assert pages["/home"]["pv"] == 10
|
||
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_query_cache_invalidation()
|
||
test_duration_efficiency_queries()
|
||
test_year_filter_and_published_sort()
|
||
test_about_payload()
|
||
test_stats_query()
|
||
test_fans_custom_range_modes()
|
||
test_fans_candidates_system_and_export()
|
||
print("OK: Web 查询层总览/榜单/详情/UP/时长效率/粉丝/候选/系统/访问统计测试全部通过")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main_test()
|