vosonandCursor
30218790ce
Ship Climperor Web 0.6.13: latest-patch summary panel, icon/innate fallbacks, and layout/scroll fixes; keep shared ability badges in git. Co-authored-by: Cursor <cursoragent@cursor.com>
519 lines
21 KiB
Python
519 lines
21 KiB
Python
from __future__ import annotations
|
||
|
||
import json
|
||
import os
|
||
import sys
|
||
import tempfile
|
||
import unittest
|
||
import urllib.error
|
||
from datetime import datetime, timedelta, timezone
|
||
from pathlib import Path
|
||
from unittest.mock import patch
|
||
|
||
ROOT = Path(__file__).resolve().parents[2]
|
||
sys.path.insert(0, str(ROOT))
|
||
sys.path.insert(0, str(ROOT / "web"))
|
||
|
||
from web import notify_site_traffic, refresh_web
|
||
from web.fetch_stratz_meta import preserve_failed_meta_brackets
|
||
from web.fetch_streamer_live import probe_streamers
|
||
from web.fetch_streamers import (
|
||
DOUYU_AUTHOR_HASH_RE,
|
||
extract_douyu_up_id_from_room_html,
|
||
extract_profile_from_douyu_data,
|
||
parse_douyu_dollar_data,
|
||
)
|
||
from web.refresh_cache import restore_patches, restore_streamers
|
||
from shared.http_utils import write_json_atomic
|
||
from shared.paths import HEROES_JSON, RELATIONS_JSON, WEB_FRONTEND
|
||
|
||
|
||
class RefreshDigestTests(unittest.TestCase):
|
||
def test_atomic_json_replaces_complete_document(self) -> None:
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
path = Path(tmp) / "data.json"
|
||
write_json_atomic(path, {"value": [1, 2, 3]})
|
||
self.assertEqual(json.loads(path.read_text(encoding="utf-8")), {"value": [1, 2, 3]})
|
||
self.assertEqual(list(path.parent.glob(f".{path.name}.*.tmp")), [])
|
||
|
||
def test_semantic_digest_ignores_refresh_timestamps(self) -> None:
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
path = Path(tmp) / "data.json"
|
||
path.write_text(
|
||
json.dumps({"fetched_at": "old", "by_hero": {"axe": {"pick": 1}}}),
|
||
encoding="utf-8",
|
||
)
|
||
before = refresh_web._semantic_file_digest(path)
|
||
path.write_text(
|
||
json.dumps({"fetched_at": "new", "by_hero": {"axe": {"pick": 1}}}),
|
||
encoding="utf-8",
|
||
)
|
||
self.assertEqual(before, refresh_web._semantic_file_digest(path))
|
||
path.write_text(
|
||
json.dumps({"fetched_at": "new", "by_hero": {"axe": {"pick": 2}}}),
|
||
encoding="utf-8",
|
||
)
|
||
self.assertNotEqual(before, refresh_web._semantic_file_digest(path))
|
||
|
||
def test_semantic_digest_ignores_streamer_is_live(self) -> None:
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
path = Path(tmp) / "streamers.json"
|
||
path.write_text(
|
||
json.dumps(
|
||
{
|
||
"fetched_at": "old",
|
||
"streamers": [
|
||
{
|
||
"id": "a",
|
||
"nickname": "A",
|
||
"is_live": False,
|
||
"live_probed_at": "t1",
|
||
}
|
||
],
|
||
}
|
||
),
|
||
encoding="utf-8",
|
||
)
|
||
before = refresh_web._semantic_file_digest(path)
|
||
path.write_text(
|
||
json.dumps(
|
||
{
|
||
"fetched_at": "new",
|
||
"streamers": [
|
||
{
|
||
"id": "a",
|
||
"nickname": "A",
|
||
"is_live": True,
|
||
"live_probed_at": "t2",
|
||
}
|
||
],
|
||
}
|
||
),
|
||
encoding="utf-8",
|
||
)
|
||
self.assertEqual(before, refresh_web._semantic_file_digest(path))
|
||
path.write_text(
|
||
json.dumps(
|
||
{
|
||
"fetched_at": "new",
|
||
"streamers": [
|
||
{
|
||
"id": "a",
|
||
"nickname": "B",
|
||
"is_live": True,
|
||
"live_probed_at": "t2",
|
||
}
|
||
],
|
||
}
|
||
),
|
||
encoding="utf-8",
|
||
)
|
||
self.assertNotEqual(before, refresh_web._semantic_file_digest(path))
|
||
|
||
def test_snapshot_watches_frontend_and_shared_inputs(self) -> None:
|
||
snap = refresh_web.snapshot()
|
||
self.assertIn(refresh_web._watch_key(RELATIONS_JSON), snap)
|
||
self.assertIn(refresh_web._watch_key(HEROES_JSON), snap)
|
||
self.assertIn(refresh_web._watch_key(WEB_FRONTEND / "app.js"), snap)
|
||
self.assertIn(
|
||
refresh_web._watch_key(WEB_FRONTEND / "functions") + "/",
|
||
snap,
|
||
)
|
||
self.assertIn(
|
||
refresh_web._watch_key(refresh_web.ROOT / "data" / "hero_grid_order.json"),
|
||
snap,
|
||
)
|
||
|
||
def test_frontend_change_is_data_not_asset(self) -> None:
|
||
before = {"web/frontend/app.js": "a", "web/assets/item_icons/": "x"}
|
||
after = {"web/frontend/app.js": "b", "web/assets/item_icons/": "x"}
|
||
data_changed, assets_changed = refresh_web.diff_snapshots(before, after)
|
||
self.assertTrue(data_changed)
|
||
self.assertFalse(assets_changed)
|
||
|
||
def test_assets_cannot_deploy_when_oss_is_skipped(self) -> None:
|
||
with self.assertRaises(RuntimeError):
|
||
refresh_web.validate_publish_options(
|
||
assets_changed=True,
|
||
skip_oss=True,
|
||
skip_deploy=False,
|
||
)
|
||
refresh_web.validate_publish_options(
|
||
assets_changed=True,
|
||
skip_oss=True,
|
||
skip_deploy=True,
|
||
)
|
||
|
||
def test_opendota_probe_failure_skips_opendota_daily_fetches(self) -> None:
|
||
calls: list[str] = []
|
||
|
||
def fake_run_script(script: str, *args: str, dry_run: bool = False, soft_fail: bool = False):
|
||
calls.append(script)
|
||
return True
|
||
|
||
with (
|
||
patch.object(refresh_web, "probe_opendota_available", return_value=False),
|
||
patch.object(refresh_web, "run_script", side_effect=fake_run_script),
|
||
patch.object(
|
||
refresh_web,
|
||
"patch_check",
|
||
return_value={"has_new": False, "new_versions": []},
|
||
),
|
||
):
|
||
self.assertFalse(refresh_web.run_daily(dry_run=False))
|
||
self.assertNotIn("fetch_hero_stats.py", calls)
|
||
self.assertNotIn("fetch_hero_matches.py", calls)
|
||
self.assertNotIn("fetch_pro_matches.py", calls)
|
||
self.assertIn("fetch_leaderboards.py", calls)
|
||
self.assertIn("fetch_streamers.py", calls)
|
||
|
||
def test_critical_empty_by_hero_refuses_deploy(self) -> None:
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
data = Path(tmp)
|
||
paths = [
|
||
data / "hero_stats.json",
|
||
data / "stratz_hero_meta.json",
|
||
data / "stratz_matchup_tops.json",
|
||
]
|
||
write_json_atomic(paths[0], {"by_hero": {"axe": {"pick": 1}}})
|
||
write_json_atomic(paths[1], {"by_hero": {}})
|
||
write_json_atomic(paths[2], {"by_hero": {"axe": {}}})
|
||
with patch.object(refresh_web, "CRITICAL_BY_HERO_FILES", tuple(paths)):
|
||
with self.assertRaises(RuntimeError) as ctx:
|
||
refresh_web.assert_critical_data_ready()
|
||
self.assertIn("stratz_hero_meta.json", str(ctx.exception))
|
||
write_json_atomic(paths[1], {"by_hero": {"axe": {}}})
|
||
with patch.object(refresh_web, "CRITICAL_BY_HERO_FILES", tuple(paths)):
|
||
refresh_web.assert_critical_data_ready()
|
||
|
||
def test_dry_run_skips_cache_restore_and_summary_write(self) -> None:
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
summary_path = Path(tmp) / "summary.json"
|
||
calls: list[tuple] = []
|
||
|
||
def fake_run_script(script: str, *args: str, dry_run: bool = False, soft_fail: bool = False):
|
||
calls.append((script, args, dry_run))
|
||
return True
|
||
|
||
env = {"REFRESH_CACHE_ENABLED": "1"}
|
||
with (
|
||
patch.object(refresh_web, "SUMMARY_PATH", summary_path),
|
||
patch.object(refresh_web, "acquire_refresh_lock", return_value=object()),
|
||
patch.object(refresh_web, "release_refresh_lock"),
|
||
patch.object(refresh_web, "run_script", side_effect=fake_run_script),
|
||
patch.object(refresh_web, "run_weekly"),
|
||
patch.object(refresh_web, "run_daily", return_value=False),
|
||
patch.object(refresh_web, "run_patch_tier", return_value=False),
|
||
patch.dict(os.environ, env, clear=False),
|
||
patch.object(sys, "argv", ["refresh_web.py", "--tier", "weekly", "--dry-run"]),
|
||
):
|
||
refresh_web.main()
|
||
self.assertFalse(summary_path.is_file())
|
||
self.assertFalse(any(script == "refresh_cache.py" for script, *_ in calls))
|
||
|
||
def test_streamer_cache_preserves_manual_rows_and_restores_runtime_fields(self) -> None:
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
cached = Path(tmp) / "cached.json"
|
||
checkout = Path(tmp) / "checkout.json"
|
||
write_json_atomic(
|
||
cached,
|
||
{
|
||
"fetched_at": "2026-07-29T10:00:00+00:00",
|
||
"streamers": [
|
||
{
|
||
"id": "known",
|
||
"nickname": "cached-name",
|
||
"follower_count": 99,
|
||
"live_url": "must-not-restore",
|
||
},
|
||
{"id": "deleted", "nickname": "deleted"},
|
||
],
|
||
},
|
||
)
|
||
write_json_atomic(
|
||
checkout,
|
||
{
|
||
"fetched_at": "2026-07-29T09:00:00+00:00",
|
||
"streamers": [
|
||
{
|
||
"id": "known",
|
||
"nickname": "git-name",
|
||
"live_url": "new-manual-url",
|
||
},
|
||
{"id": "new", "nickname": "new-manual-row"},
|
||
],
|
||
},
|
||
)
|
||
restore_streamers(cached, checkout)
|
||
payload = json.loads(checkout.read_text(encoding="utf-8"))
|
||
rows = {row["id"]: row for row in payload["streamers"]}
|
||
self.assertEqual(rows["known"]["nickname"], "cached-name")
|
||
self.assertEqual(rows["known"]["follower_count"], 99)
|
||
self.assertEqual(rows["known"]["live_url"], "new-manual-url")
|
||
self.assertIn("new", rows)
|
||
self.assertNotIn("deleted", rows)
|
||
|
||
def test_patch_cache_merges_runtime_and_committed_versions(self) -> None:
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
cached = Path(tmp) / "cached.json"
|
||
checkout = Path(tmp) / "checkout.json"
|
||
write_json_atomic(
|
||
cached,
|
||
{
|
||
"meta": {"fetched_at": "cached"},
|
||
"patches": [{"version": "7.40", "timestamp": 200}],
|
||
"lookup": {"heroes": {"1": {"name": "cached"}}},
|
||
"details": {"7.40": {"general_notes": ["cached"]}},
|
||
},
|
||
)
|
||
write_json_atomic(
|
||
checkout,
|
||
{
|
||
"meta": {"source": "git"},
|
||
"patches": [{"version": "7.41", "timestamp": 300}],
|
||
"lookup": {"heroes": {"2": {"name": "committed"}}},
|
||
"details": {"7.41": {"general_notes": ["committed"]}},
|
||
},
|
||
)
|
||
restore_patches(cached, checkout)
|
||
payload = json.loads(checkout.read_text(encoding="utf-8"))
|
||
self.assertEqual(
|
||
[row["version"] for row in payload["patches"]],
|
||
["7.41", "7.40"],
|
||
)
|
||
self.assertEqual(set(payload["details"]), {"7.40", "7.41"})
|
||
self.assertEqual(set(payload["lookup"]["heroes"]), {"1", "2"})
|
||
|
||
|
||
class LiveProbeFallbackTests(unittest.TestCase):
|
||
def test_probe_failure_clears_stale_live_badge(self) -> None:
|
||
payload = {
|
||
"streamers": [
|
||
{
|
||
"id": "a",
|
||
"live_url": "https://live.bilibili.com/123",
|
||
"is_live": True,
|
||
"live_probed_at": "2026-07-29T00:00:00+00:00",
|
||
}
|
||
]
|
||
}
|
||
|
||
def boom(_room_id: str) -> bool:
|
||
raise urllib.error.URLError("blocked")
|
||
|
||
with patch("web.fetch_streamer_live.probe_bilibili", side_effect=boom):
|
||
live, offline, fail = probe_streamers(payload)
|
||
self.assertEqual((live, offline, fail), (0, 0, 1))
|
||
row = payload["streamers"][0]
|
||
self.assertFalse(row["is_live"])
|
||
self.assertNotIn("live_probed_at", row)
|
||
|
||
def test_douyu_probe_path_writes_live_flag(self) -> None:
|
||
payload = {
|
||
"streamers": [
|
||
{
|
||
"id": "gouhuang",
|
||
"platform": "douyu",
|
||
"live_url": "https://www.douyu.com/957090",
|
||
}
|
||
]
|
||
}
|
||
with patch("web.fetch_streamer_live.probe_douyu", return_value=True):
|
||
live, offline, fail = probe_streamers(payload)
|
||
self.assertEqual((live, offline, fail), (1, 0, 0))
|
||
self.assertTrue(payload["streamers"][0]["is_live"])
|
||
self.assertIn("live_probed_at", payload["streamers"][0])
|
||
|
||
|
||
class DouyuProfileParseTests(unittest.TestCase):
|
||
def test_author_and_author_video_hashes(self) -> None:
|
||
self.assertEqual(
|
||
DOUYU_AUTHOR_HASH_RE.search(
|
||
"https://v.douyu.com/author/JPw9YOLKlw5X"
|
||
).group(1),
|
||
"JPw9YOLKlw5X",
|
||
)
|
||
self.assertEqual(
|
||
DOUYU_AUTHOR_HASH_RE.search(
|
||
"https://v.douyu.com/author-video/JPw9YOLKlw5X"
|
||
).group(1),
|
||
"JPw9YOLKlw5X",
|
||
)
|
||
|
||
def test_dollar_data_maps_fan_fields(self) -> None:
|
||
blob = (
|
||
'{type:"0",uid:"13506421",upId:"JPw9YOLKlw5X",upFollowNum:"16",'
|
||
'subscribeNum:"465781",playCount:"26074077",'
|
||
'avatar:"https://example.com/a.jpg",'
|
||
'name: "18yearsold\\u5929\\u6b8b\\u5c11\\u5e74K9",'
|
||
'contents: "",roomId:"235520",roomName: "k9 room"}'
|
||
)
|
||
profile = extract_profile_from_douyu_data(parse_douyu_dollar_data(blob))
|
||
self.assertEqual(profile["nickname"], "18yearsold天残少年K9")
|
||
self.assertEqual(profile["unique_id"], "235520")
|
||
self.assertEqual(profile["following_count"], 16)
|
||
self.assertEqual(profile["follower_count"], 465781)
|
||
self.assertEqual(profile["total_favorited"], 26074077)
|
||
self.assertEqual(profile["profile_url"], "https://v.douyu.com/author/JPw9YOLKlw5X")
|
||
self.assertEqual(profile["live_url"], "https://www.douyu.com/235520")
|
||
self.assertNotIn("signature", profile)
|
||
|
||
def test_room_html_up_id_extraction(self) -> None:
|
||
plain = '{"rid":957090,"up_id":"EqAvg1lQD75L","ban_display":0}'
|
||
escaped = (
|
||
r'{\"rid\":957090,\"up_id\":\"EqAvg1lQD75L\",\"ban_display\":0}'
|
||
)
|
||
self.assertEqual(extract_douyu_up_id_from_room_html(plain), "EqAvg1lQD75L")
|
||
self.assertEqual(extract_douyu_up_id_from_room_html(escaped), "EqAvg1lQD75L")
|
||
self.assertIsNone(extract_douyu_up_id_from_room_html("<html></html>"))
|
||
|
||
|
||
class NotifySummaryTests(unittest.TestCase):
|
||
def test_skipped_summary_is_labeled(self) -> None:
|
||
self.assertEqual(
|
||
notify_site_traffic._fmt_refresh_summary_result(
|
||
{
|
||
"ok": True,
|
||
"skipped": True,
|
||
"skip_reason": "full refresh already running",
|
||
}
|
||
),
|
||
"已跳过（full refresh already running）",
|
||
)
|
||
self.assertFalse(
|
||
notify_site_traffic._summary_proves_refresh(
|
||
{"ok": True, "skipped": True, "steps": [{"step": "fetch_hero_stats.py", "ok": True}]},
|
||
"英雄统计",
|
||
"fetch_hero_stats.py",
|
||
)
|
||
)
|
||
|
||
|
||
class StratzFallbackTests(unittest.TestCase):
|
||
def test_failed_bracket_keeps_previous_data_and_marks_stale(self) -> None:
|
||
fresh = {
|
||
"totals": {"immortal": {"pick": 0}},
|
||
"meta_board": {"immortal": []},
|
||
"by_hero": {
|
||
"axe": {
|
||
"weeks": {"immortal": []},
|
||
"latest": {},
|
||
"positions": {"immortal": {}},
|
||
}
|
||
},
|
||
}
|
||
previous = {
|
||
"totals": {"immortal": {"pick": 100}},
|
||
"meta_board": {"immortal": [{"key": "axe"}]},
|
||
"by_hero": {
|
||
"axe": {
|
||
"weeks": {"immortal": [{"week": 1, "pick": 100, "win": 50}]},
|
||
"latest": {"immortal": {"pick": 100, "win": 50}},
|
||
"positions": {"immortal": {"POSITION_3": {"pick": 40, "win": 20}}},
|
||
}
|
||
},
|
||
}
|
||
result = preserve_failed_meta_brackets(
|
||
fresh,
|
||
previous,
|
||
failed_weeks={"immortal"},
|
||
failed_positions={"immortal"},
|
||
)
|
||
self.assertTrue(result["stale"])
|
||
self.assertEqual(result["totals"]["immortal"]["pick"], 100)
|
||
self.assertEqual(
|
||
result["by_hero"]["axe"]["positions"]["immortal"]["POSITION_3"]["pick"],
|
||
40,
|
||
)
|
||
|
||
|
||
class ProductionHealthTests(unittest.TestCase):
|
||
def test_stale_payload_and_invalid_live_api_are_reported(self) -> None:
|
||
now = datetime(2026, 7, 29, 9, tzinfo=timezone.utc)
|
||
old = (now - timedelta(hours=48)).isoformat()
|
||
payload = {
|
||
"heroes": [{"key": "axe"}],
|
||
"meta": {"refresh_run_id": "123"},
|
||
"hero_stats": {"fetched_at": old},
|
||
"leaderboards": {"fetched_at": old},
|
||
"hero_matches": {"meta": {"fetched_at": old}},
|
||
"pro_matches": {"meta": {"fetched_at": old}},
|
||
"streamers": {"fetched_at": old},
|
||
"stratz_hero_meta": {"fetched_at": old},
|
||
"stratz_matchup_tops": {"fetched_at": old},
|
||
"hero_items": {"meta": {"fetched_at": old}},
|
||
}
|
||
responses = [
|
||
(200, b'var SITE_VERSION = "1.2.3";', "application/javascript", {}),
|
||
(200, json.dumps(payload).encode(), "application/json", {}),
|
||
(200, b"<html>fallback</html>", "text/html", {}),
|
||
]
|
||
with patch.object(
|
||
notify_site_traffic,
|
||
"http_response_soft",
|
||
side_effect=responses,
|
||
):
|
||
health = notify_site_traffic.fetch_production_health(now=now)
|
||
self.assertEqual(health.site_version, "1.2.3")
|
||
self.assertEqual(health.refresh_run_id, "123")
|
||
self.assertIn("英雄统计", health.stale)
|
||
self.assertIn("直播状态 API", health.stale)
|
||
self.assertFalse(health.live_api_ok)
|
||
|
||
def test_successful_refresh_summary_prevents_false_freshness_alarm(self) -> None:
|
||
now = datetime(2026, 7, 29, 9, tzinfo=timezone.utc)
|
||
old_daily = (now - timedelta(hours=48)).isoformat()
|
||
recent_weekly = (now - timedelta(days=2)).isoformat()
|
||
payload = {
|
||
"heroes": [{"key": "axe"}],
|
||
"meta": {"refresh_run_id": "456"},
|
||
"hero_stats": {"fetched_at": old_daily},
|
||
"leaderboards": {"fetched_at": old_daily},
|
||
"hero_matches": {"meta": {"fetched_at": old_daily}},
|
||
"pro_matches": {"meta": {"fetched_at": old_daily}},
|
||
"streamers": {"fetched_at": old_daily},
|
||
"stratz_hero_meta": {"fetched_at": recent_weekly},
|
||
"stratz_matchup_tops": {"fetched_at": recent_weekly},
|
||
"hero_items": {"meta": {"fetched_at": recent_weekly}},
|
||
}
|
||
responses = [
|
||
(200, b'var SITE_VERSION = "1.2.3";', "application/javascript", {}),
|
||
(200, json.dumps(payload).encode(), "application/json", {}),
|
||
(
|
||
200,
|
||
b'{"ok":true,"streamers":{}}',
|
||
"application/json",
|
||
{"x-live-cache": "hit"},
|
||
),
|
||
]
|
||
with patch.object(
|
||
notify_site_traffic,
|
||
"http_response_soft",
|
||
side_effect=responses,
|
||
):
|
||
health = notify_site_traffic.fetch_production_health(
|
||
now=now,
|
||
refresh_summaries={
|
||
"web-daily.yml": {
|
||
"ok": True,
|
||
"steps": [
|
||
{"step": "fetch_hero_stats.py", "ok": True},
|
||
{"step": "fetch_leaderboards.py", "ok": True},
|
||
{"step": "fetch_hero_matches.py", "ok": True},
|
||
{"step": "fetch_pro_matches.py", "ok": True},
|
||
{"step": "fetch_streamers.py", "ok": True},
|
||
],
|
||
"health": {
|
||
"streamer_profile_missing": 0,
|
||
"streamer_live_probe_missing": 0,
|
||
},
|
||
}
|
||
},
|
||
)
|
||
self.assertEqual(health.stale, [])
|
||
self.assertTrue(health.live_api_ok)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|