283 lines
10 KiB
Python
283 lines
10 KiB
Python
"""Regression checks for persisted public-scoreboard match links."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import gc
|
|
import os
|
|
import sqlite3
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
from app.historical_storage import (
|
|
get_historical_match_detail,
|
|
initialize_historical_storage,
|
|
list_recent_historical_matches,
|
|
upsert_historical_match,
|
|
)
|
|
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
|
|
|
|
|
|
class PersistedScoreboardMatchLinkTests(unittest.TestCase):
|
|
def test_recent_and_detail_payloads_expose_safe_persisted_match_url(self) -> None:
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
db_path = Path(tmpdir) / "historical.sqlite3"
|
|
match_url = "https://scoreboard.comunidadhll.es:5443/games/12345"
|
|
_persist_match(db_path, server_slug="comunidad-hispana-02", match_id="12345")
|
|
|
|
recent_items = list_recent_historical_matches(
|
|
server_slug="comunidad-hispana-02",
|
|
limit=5,
|
|
db_path=db_path,
|
|
)
|
|
detail = get_historical_match_detail(
|
|
server_slug="comunidad-hispana-02",
|
|
match_id="12345",
|
|
db_path=db_path,
|
|
)
|
|
|
|
self.assertEqual(recent_items[0]["match_url"], match_url)
|
|
self.assertIsNotNone(detail)
|
|
self.assertEqual(detail["match_url"], match_url)
|
|
gc.collect()
|
|
|
|
def test_untrusted_persisted_match_url_is_not_exposed(self) -> None:
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
db_path = Path(tmpdir) / "historical.sqlite3"
|
|
_persist_match(db_path, server_slug="comunidad-hispana-01", match_id="999")
|
|
_set_raw_payload_ref(
|
|
db_path,
|
|
match_id="999",
|
|
raw_payload_ref="https://scoreboard.comunidadhll.es:3443/games/999",
|
|
)
|
|
|
|
recent_items = list_recent_historical_matches(
|
|
server_slug="comunidad-hispana-01",
|
|
limit=5,
|
|
db_path=db_path,
|
|
)
|
|
detail = get_historical_match_detail(
|
|
server_slug="comunidad-hispana-01",
|
|
match_id="999",
|
|
db_path=db_path,
|
|
)
|
|
|
|
self.assertIsNone(recent_items[0]["match_url"])
|
|
self.assertIsNotNone(detail)
|
|
self.assertIsNone(detail["match_url"])
|
|
gc.collect()
|
|
|
|
def test_detail_player_links_use_trusted_scoreboard_steam_id(self) -> None:
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
db_path = Path(tmpdir) / "historical.sqlite3"
|
|
_persist_match(
|
|
db_path,
|
|
server_slug="comunidad-hispana-02",
|
|
match_id="steam-player-match",
|
|
player_stats=[
|
|
{
|
|
"player": "Steam Player",
|
|
"steaminfo": {"profile": {"steamid": "76561198000000009"}},
|
|
"team": {"side": "allies"},
|
|
"kills": 4,
|
|
"deaths": 2,
|
|
}
|
|
],
|
|
)
|
|
|
|
detail = get_historical_match_detail(
|
|
server_slug="comunidad-hispana-02",
|
|
match_id="steam-player-match",
|
|
db_path=db_path,
|
|
)
|
|
|
|
self.assertIsNotNone(detail)
|
|
player = detail["players"][0]
|
|
self.assertEqual(player["steam_id_64"], "76561198000000009")
|
|
self.assertEqual(player["platform"], "steam")
|
|
self.assertEqual(
|
|
player["external_profile_links"]["hll_records"],
|
|
"https://hllrecords.com/profiles/76561198000000009",
|
|
)
|
|
gc.collect()
|
|
|
|
def test_rcon_match_detail_does_not_fabricate_external_scoreboard_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:
|
|
initialize_rcon_historical_storage(db_path=db_path)
|
|
detail = get_rcon_historical_match_detail(
|
|
server_key="comunidad-hispana-01",
|
|
match_id="rcon:synthetic-window",
|
|
)
|
|
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.assertIsNone(detail)
|
|
gc.collect()
|
|
|
|
def test_rcon_match_detail_exposes_correlated_scoreboard_url_on_strong_evidence(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-01",
|
|
match_id="1561515",
|
|
map_name="St. Mere Eglise",
|
|
started_at="2026-04-12T16:20:00Z",
|
|
ended_at="2026-04-12T17:45:00Z",
|
|
)
|
|
session_key = _persist_rcon_window(
|
|
db_path,
|
|
map_name="St. Mere Eglise",
|
|
first_seen_at="2026-04-12T16:28:55.761810Z",
|
|
last_seen_at="2026-04-12T16:43:55.761810Z",
|
|
players=94,
|
|
max_players=98,
|
|
)
|
|
|
|
detail = get_rcon_historical_match_detail(
|
|
server_key="comunidad-hispana-01",
|
|
match_id=session_key,
|
|
)
|
|
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.assertIsNotNone(detail)
|
|
self.assertEqual(
|
|
detail["match_url"],
|
|
"https://scoreboard.comunidadhll.es/games/1561515",
|
|
)
|
|
gc.collect()
|
|
|
|
def test_rcon_match_detail_keeps_low_confidence_correlation_unlinked(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-01",
|
|
match_id="1561515",
|
|
map_name="Carentan",
|
|
started_at="2026-04-12T10:00:00Z",
|
|
ended_at="2026-04-12T11:30:00Z",
|
|
)
|
|
session_key = _persist_rcon_window(
|
|
db_path,
|
|
map_name="St. Mere Eglise",
|
|
first_seen_at="2026-04-12T16:28:55.761810Z",
|
|
last_seen_at="2026-04-12T16:43:55.761810Z",
|
|
players=94,
|
|
max_players=98,
|
|
)
|
|
|
|
detail = get_rcon_historical_match_detail(
|
|
server_key="comunidad-hispana-01",
|
|
match_id=session_key,
|
|
)
|
|
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.assertIsNotNone(detail)
|
|
self.assertIsNone(detail["match_url"])
|
|
gc.collect()
|
|
|
|
|
|
def _persist_match(
|
|
db_path: Path,
|
|
*,
|
|
server_slug: str,
|
|
match_id: str,
|
|
map_name: str = "carentan",
|
|
started_at: str = "2026-05-01T10:00:00Z",
|
|
ended_at: str = "2026-05-01T11:20:00Z",
|
|
player_stats: list[dict[str, object]] | None = None,
|
|
) -> None:
|
|
upsert_historical_match(
|
|
server_slug=server_slug,
|
|
match_payload={
|
|
"id": match_id,
|
|
"creation_time": started_at,
|
|
"start": started_at,
|
|
"end": ended_at,
|
|
"map": {"name": map_name},
|
|
"result": {"allied": 3, "axis": 2},
|
|
"player_stats": player_stats or [],
|
|
},
|
|
db_path=db_path,
|
|
)
|
|
|
|
|
|
def _persist_rcon_window(
|
|
db_path: Path,
|
|
*,
|
|
map_name: str,
|
|
first_seen_at: str,
|
|
last_seen_at: str,
|
|
players: int,
|
|
max_players: int,
|
|
) -> str:
|
|
initialize_rcon_historical_storage(db_path=db_path)
|
|
run_id = start_rcon_historical_capture_run(
|
|
mode="test",
|
|
target_scope="comunidad-hispana-01",
|
|
db_path=db_path,
|
|
)
|
|
target = {
|
|
"target_key": "comunidad-hispana-01",
|
|
"external_server_id": "comunidad-hispana-01",
|
|
"name": "Comunidad Hispana #01",
|
|
"host": "127.0.0.1",
|
|
"port": 7779,
|
|
}
|
|
for captured_at in (first_seen_at, last_seen_at):
|
|
persist_rcon_historical_sample(
|
|
run_id=run_id,
|
|
captured_at=captured_at,
|
|
target=target,
|
|
normalized_payload={
|
|
"status": "online",
|
|
"players": players,
|
|
"max_players": max_players,
|
|
"current_map": map_name,
|
|
},
|
|
raw_payload={},
|
|
db_path=db_path,
|
|
)
|
|
return f"1:{first_seen_at}"
|
|
|
|
|
|
def _set_raw_payload_ref(db_path: Path, *, match_id: str, raw_payload_ref: str) -> None:
|
|
with sqlite3.connect(db_path) as connection:
|
|
connection.execute(
|
|
"""
|
|
UPDATE historical_matches
|
|
SET raw_payload_ref = ?
|
|
WHERE external_match_id = ?
|
|
""",
|
|
(raw_payload_ref, match_id),
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|