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.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_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) 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"fallback", "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()