from __future__ import annotations import json import sys import tempfile import unittest 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.refresh_cache import restore_patches, restore_streamers from shared.http_utils import write_json_atomic 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_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_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 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()