rgonsal/comunidadhll
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: persist and diagnose scoreboard correlation candidates

Browse Source
This commit is contained in:
devRaGonSa
2026-05-21 14:45:49 +02:00
parent 6383d08eba
commit f0820a1da7
13 changed files with 872 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files

154
backend/tests/test_scoreboard_match_links.py
View File

@@ -8,7 +8,9 @@ import sqlite3
import tempfile
import unittest
from pathlib import Path
from unittest.mock import patch
from app.scoreboard_candidate_backfill import run_backfill
from app.historical_storage import (
get_historical_match_detail,
initialize_historical_storage,
@@ -19,9 +21,80 @@ from app.rcon_historical_storage import initialize_rcon_historical_storage
from app.rcon_historical_storage import persist_rcon_historical_sample
from app.rcon_historical_storage import start_rcon_historical_capture_run
from app.rcon_historical_read_model import get_rcon_historical_match_detail
from app.rcon_admin_log_materialization import materialize_rcon_admin_log
from app.rcon_admin_log_storage import persist_rcon_admin_log_entries
from app.rcon_scoreboard_relink import relink_materialized_matches
from app.scoreboard_correlation_diagnostics import inspect_materialized_match_correlation
class PersistedScoreboardMatchLinkTests(unittest.TestCase):
def test_list_backfill_persists_foy_candidate_before_detail_fetch_failure(self) -> None:
stored: dict[tuple[str, str], dict[str, object]] = {}
class FoyListProvider:
def fetch_match_page(self, *, base_url: str, page: int, limit: int) -> dict[str, object]:
return {"maps": [_foy_list_match()]} if page == 1 else {"maps": []}
def fetch_match_details(
self,
*,
base_url: str,
match_ids: list[str],
max_workers: int,
) -> list[dict[str, object]]:
raise RuntimeError("detail endpoint unavailable")
def fake_upsert(*, server_slug: str, candidate: dict[str, object]) -> str:
key = (server_slug, str(candidate["external_match_id"]))
outcome = "updated" if key in stored else "inserted"
stored[key] = dict(candidate)
return outcome
server = {
"slug": "comunidad-hispana-02",
"scoreboard_base_url": "https://scoreboard.comunidadhll.es:5443",
"server_number": 2,
}
with (
patch("app.scoreboard_candidate_backfill.initialize_historical_storage"),
patch(
"app.scoreboard_candidate_backfill.PublicScoreboardHistoricalDataSource",
return_value=FoyListProvider(),
),
patch(
"app.scoreboard_candidate_backfill.upsert_scoreboard_candidate",
side_effect=fake_upsert,
),
):
first = run_backfill(
server=server,
start_at=_backfill_timestamp("2026-05-20T00:00:00Z"),
end_at=_backfill_timestamp("2026-05-21T23:59:59Z"),
max_pages=2,
page_size=100,
detail_workers=1,
)
second = run_backfill(
server=server,
start_at=_backfill_timestamp("2026-05-20T00:00:00Z"),
end_at=_backfill_timestamp("2026-05-21T23:59:59Z"),
max_pages=2,
page_size=100,
detail_workers=1,
)
candidate = stored[("comunidad-hispana-02", "1562115")]
self.assertEqual(
candidate["match_url"],
"https://scoreboard.comunidadhll.es:5443/games/1562115",
)
self.assertEqual(first["list_candidates_inserted"], 1)
self.assertEqual(first["list_candidates_updated"], 0)
self.assertEqual(first["errors"][0]["stage"], "fetch_match_details")
self.assertEqual(second["list_candidates_inserted"], 0)
self.assertEqual(second["list_candidates_updated"], 1)
self.assertEqual(len(stored), 1)
def test_recent_and_detail_payloads_expose_safe_persisted_match_url(self) -> None:
with tempfile.TemporaryDirectory() as tmpdir:
db_path = Path(tmpdir) / "historical.sqlite3"
@@ -201,6 +274,70 @@ class PersistedScoreboardMatchLinkTests(unittest.TestCase):
self.assertIsNone(detail["match_url"])
gc.collect()
def test_foy_relink_reports_existing_materialized_match_url(self) -> None:
with tempfile.TemporaryDirectory() as tmpdir:
db_path = Path(tmpdir) / "historical.sqlite3"
previous_storage_path = os.environ.get("HLL_BACKEND_STORAGE_PATH")
os.environ["HLL_BACKEND_STORAGE_PATH"] = str(db_path)
try:
_persist_match(
db_path,
server_slug="comunidad-hispana-02",
match_id="1562115",
map_name="Foy Warfare",
started_at="2026-05-20T20:54:11Z",
ended_at="2026-05-20T22:24:11Z",
)
persist_rcon_admin_log_entries(
target={
"target_key": "comunidad-hispana-02",
"external_server_id": "comunidad-hispana-02",
},
entries=[
{
"timestamp": "2026-05-20T20:54:11Z",
"message": "[1 min (1779310451)] MATCH START Foy Warfare",
},
{
"timestamp": "2026-05-20T22:24:11Z",
"message": "[91 min (1779315851)] MATCH ENDED `Foy Warfare` ALLIED (4 - 1) AXIS",
},
],
db_path=db_path,
)
materialize_rcon_admin_log(db_path=db_path)
report = relink_materialized_matches(
server_key="comunidad-hispana-02",
db_path=db_path,
)
detail = get_rcon_historical_match_detail(
server_key="comunidad-hispana-02",
match_id="comunidad-hispana-02:1779310451:1779315851:foywarfare",
)
diagnostics = inspect_materialized_match_correlation(
server_key="comunidad-hispana-02",
match_key="comunidad-hispana-02:1779310451:1779315851:foywarfare",
db_path=db_path,
)
finally:
if previous_storage_path is None:
os.environ.pop("HLL_BACKEND_STORAGE_PATH", None)
else:
os.environ["HLL_BACKEND_STORAGE_PATH"] = previous_storage_path
self.assertEqual(report["matches_scanned"], 1)
self.assertEqual(report["matches_linked"], 1)
self.assertGreaterEqual(report["candidates_scanned"], 1)
self.assertIsNotNone(detail)
self.assertEqual(
detail["match_url"],
"https://scoreboard.comunidadhll.es:5443/games/1562115",
)
self.assertEqual(diagnostics["final_reason"], "linked")
self.assertEqual(diagnostics["selected_candidate"]["external_match_id"], "1562115")
self.assertEqual(diagnostics["top_candidates"][0]["map"], "Foy Warfare")
gc.collect()
def _persist_match(
db_path: Path,
@@ -227,6 +364,23 @@ def _persist_match(
)
def _foy_list_match() -> dict[str, object]:
return {
"id": 1562115,
"server_number": 2,
"start": "2026-05-20T20:54:11+00:00",
"end": "2026-05-20T22:24:11+00:00",
"map": {"id": "foywarfare", "pretty_name": "Foy Warfare"},
"result": {"allied": 4, "axis": 1},
}
def _backfill_timestamp(raw_value: str):
from app.scoreboard_candidate_backfill import _parse_timestamp
return _parse_timestamp(raw_value, option_name="test")
def _persist_rcon_window(
db_path: Path,
*,