import asyncio import os import pathlib import sys import tempfile sys.path.insert(0, str(pathlib.Path(__file__).resolve().parents[1])) sys.path.insert(0, str(pathlib.Path(__file__).resolve().parents[2] / "SHARED")) # STORAGE_DIR is resolved at import time from STORAGE_VOL_PATH; point it at a temp # dir so importing the module (and creating tss_tournaments.db) is self-contained. _TMP = tempfile.mkdtemp(prefix="tss_tournaments_test_") os.environ.setdefault("STORAGE_VOL_PATH", _TMP) from BOT import tss_tournaments as tt # noqa: E402 # --- captured sample shapes (see TSSBOT/docs/tss_tournament_api_reference.md) --- SCHEDULE = { "all_teams": [ {"teamName": "uuid-a", "realName": "NUGOB", "id_team": "1"}, {"teamName": "uuid-b", "realName": "GRIDAC", "id_team": "2"}, {"teamName": "", "realName": "IVOXY", "id_team": "3"}, # empty uuid row ], "all_matches": [ { # played "id": "686556", "typeBracket": "Looser", "round": "1", "matchNumber": "0", "teamA": "uuid-a", "teamB": "uuid-b", "winner": "uuid-a", "scoreA": "2", "scoreB": "1", "finished": "1", "timeStart": "1720286700", }, { # TBD future match (no teams yet) "id": "686600", "typeBracket": "Final", "round": "5", "matchNumber": "0", "teamA": "", "teamB": "", "winner": "", "scoreA": "0", "scoreB": "0", "finished": "0", "timeStart": "0", }, ], } SWISS = { "GroupMatch": [ { "id": "203269", "typeGroup": "Swiss", "teamA": "uuid-x", "teamB": "uuid-y", "winner": "uuid-x", "realNameA": "avrcls", "realNameB": "111111", "scoreA": "1", "scoreB": "0", "statsStatus": 1, } ], "GroupStage": [[ {"teamName": "uuid-x", "realName": "avrcls", "points": 4, "won": 2, "draw": 0, "defeats": 1, "buchholz_points": "36"}, {"teamName": "uuid-y", "realName": "111111", "points": 2, "won": 1, "draw": 0, "defeats": 2, "buchholz_points": "30"}, ]], } BATTLES_PLAYED = [ {"url": "224584316650954636", "position": 0, "statusReplay": "view replay", "winner": "NUGOB"}, {"url": "224584400000000000", "position": 1, "statusReplay": "view replay", "winner": "NUGOB"}, ] BATTLES_TECHNICAL = [{"url": "", "position": 0, "statusReplay": "technical victory", "winner": "GRIDAC"}] BATTLES_WAIT = [{"url": "224584500000000000", "position": 0, "statusReplay": "wait", "winner": ""}] def test_to_hex(): assert tt.to_hex("224584316650954636") == "31de23f001a9f8c" assert tt.to_hex("") is None assert tt.to_hex(None) is None assert tt.to_hex("not-a-number") is None def test_normalize_side(): assert tt.normalize_side("Winner") == "winner" assert tt.normalize_side("Looser") == "loser" assert tt.normalize_side("LooserFinal") == "loser" # 'looser' wins over 'final' assert tt.normalize_side("Final") == "final" assert tt.normalize_side("Semifinal") == "final" assert tt.normalize_side("Swiss") == "swiss" assert tt.normalize_side("Group") == "group" def test_apply_tournament_side_context_moves_double_elim_semifinal_to_loser(): matches = [ {"type_bracket": "Winner", "side": "winner"}, {"type_bracket": "Looser", "side": "loser"}, {"type_bracket": "Semifinal", "side": "final"}, {"type_bracket": "Final", "side": "final"}, ] tt.apply_tournament_side_context(matches) assert matches[2]["side"] == "loser" assert matches[3]["side"] == "final" single_elim = [{"type_bracket": "Semifinal", "side": "final"}] tt.apply_tournament_side_context(single_elim) assert single_elim[0]["side"] == "final" def test_derive_format(): assert tt.derive_format([], "double-elumination") == "double-elim" assert tt.derive_format([], "single-elumination") == "single-elim" assert tt.derive_format({"Winner", "Final"}) == "single-elim" assert tt.derive_format({"Winner", "Looser", "Final"}) == "double-elim" assert tt.derive_format({"Swiss"}) == "swiss" assert tt.derive_format({"Group"}) == "group" def test_team_name_map_skips_empty_uuid(): names = tt.team_name_map(SCHEDULE["all_teams"]) assert names == {"uuid-a": "NUGOB", "uuid-b": "GRIDAC"} assert "" not in names def test_parse_schedule_match_played_and_tbd(): names = tt.team_name_map(SCHEDULE["all_teams"]) played = tt.parse_schedule_match(SCHEDULE["all_matches"][0], names) assert played["team_a_name"] == "NUGOB" assert played["team_b_name"] == "GRIDAC" assert played["winner_name"] == "NUGOB" assert played["side"] == "loser" assert (played["round"], played["position"]) == (1, 0) assert (played["score_a"], played["score_b"]) == (2, 1) assert played["status"] == "played" tbd = tt.parse_schedule_match(SCHEDULE["all_matches"][1], names) assert tbd["team_a_name"] is None and tbd["team_b_name"] is None assert tbd["status"] == "pending" def test_parse_bracket_fallback_uses_inline_names(): row = { "id": "686548", "typeBracket": "Looser", "round": "0", "teamA": "", "teamB": "uuid-b", "realNameA": "", "realNameB": "GRIDAC", "winner": "uuid-b", "scoreA": "0", "scoreB": "2", "matchResult": "done", } match = tt.parse_schedule_match(row, {}) assert match["team_b_name"] == "GRIDAC" assert match["winner_name"] == "GRIDAC" assert match["status"] == "bye" def test_parse_group_match_inline_names_and_winner(): m = tt.parse_group_match(SWISS["GroupMatch"][0]) assert m["team_a_name"] == "avrcls" assert m["team_b_name"] == "111111" assert m["winner_name"] == "avrcls" assert m["side"] == "swiss" assert m["round"] is None assert m["status"] == "played" def test_parse_standings(): rows = tt.parse_standings(SWISS["GroupStage"]) assert len(rows) == 2 assert rows[0]["team_name"] == "avrcls" assert rows[0]["wins"] == 2 and rows[0]["losses"] == 1 assert rows[0]["buchholz"] == 36.0 assert rows[0]["rank"] == 1 and rows[1]["rank"] == 2 def test_parse_battles_filters_and_flags(): battles, technical = tt.parse_battles(BATTLES_PLAYED, 20042, "686556", "Looser") assert len(battles) == 2 assert battles[0]["session_hex"] == "31de23f001a9f8c" assert technical is False none_battles, tech = tt.parse_battles(BATTLES_TECHNICAL, 20042, "686548", "Looser") assert none_battles == [] assert tech is True waiting, wait_technical = tt.parse_battles(BATTLES_WAIT, 20042, "686700", "Winner") assert waiting == [] assert wait_technical is False def test_fill_names_from_battles_by_uuid(): match = { "team_a_uuid": "uuid-a", "team_a_name": None, "team_b_uuid": "uuid-b", "team_b_name": None, "winner_name": None, "score_a": 2, "score_b": 1, } rows = [{ "teamA": {"teamName": "uuid-b", "realName": "GRIDAC"}, "teamB": {"teamName": "uuid-a", "realName": "NUGOB"}, }] tt.fill_names_from_battles(match, rows) assert match["team_a_name"] == "NUGOB" assert match["team_b_name"] == "GRIDAC" assert match["winner_name"] == "NUGOB" def test_fetch_battles_for_matches_concurrent(monkeypatch): calls = [] def fake_request(method, action, **params): calls.append((method, action, params["idMatch"])) return [{ "url": "224584316650954636", "position": 0, "statusReplay": "view replay", "winner": "NUGOB", "teamA": {"teamName": "uuid-a", "realName": "NUGOB"}, "teamB": {"teamName": "uuid-b", "realName": "GRIDAC"}, }] monkeypatch.setattr(tt, "_request", fake_request) matches = [{ "match_id": "m1", "type_bracket": "Winner", "status": "played", "team_a_uuid": "uuid-a", "team_a_name": None, "team_b_uuid": "uuid-b", "team_b_name": None, "winner_name": None, "score_a": 1, "score_b": 0, }] battles = tt.fetch_battles_for_matches(123, matches, workers=2) assert len(calls) == 1 assert battles[0]["session_hex"] == "31de23f001a9f8c" assert matches[0]["team_a_name"] == "NUGOB" assert matches[0]["team_b_name"] == "GRIDAC" def test_compute_status(): played = [{"status": "played"}, {"status": "bye"}] mixed = [{"status": "played"}, {"status": "pending"}] assert tt.compute_status([], None, 1000) == "pending" assert tt.compute_status(played, None, 1000) == "finished" assert tt.compute_status(mixed, None, 1000) == "active" # A tournament still listed as active is NOT finished just because it overran # its *scheduled* dateEndTournament — finals routinely get played late. assert tt.compute_status(mixed, 100, 100 + tt.RESCAN_END_BUFFER_SECONDS + 1) == "active" assert tt.compute_status(mixed, 100, 100 + tt.RESCAN_END_BUFFER_SECONDS + 1, in_active=True) == "active" # absent from the active list → finished (old tournament, no date_end) assert tt.compute_status(mixed, None, 1000, in_active=False) == "finished" # active-list state unknown → fall back to the scheduled end + buffer assert tt.compute_status(mixed, 100, 100 + tt.RESCAN_END_BUFFER_SECONDS + 1, in_active=None) == "finished" # unknown active-list state without a past end should not finish by absence alone assert tt.compute_status(mixed, None, 1000, in_active=None) == "active" def test_needs_scan_overrun_keeps_scanning(): # A live tournament past its *scheduled* end must stay eligible for re-scan so # a late final still lands; only status=='finished' permanently stops re-scans. async def run(): await tt.init_tss_tournaments_db() now = 1_000_000 match = { "match_id": "m1", "type_bracket": "Winner", "side": "winner", "round": 0, "position": 0, "team_a_uuid": None, "team_a_name": "A", "team_b_uuid": None, "team_b_name": "B", "winner_name": None, "score_a": 0, "score_b": 0, "status": "pending", "time_start": None, } scan = { "tournament_id": 77777, "name": "Overrun Cup", "format": "single-elim", "game_mode": "RB", "cluster": "EU", "date_start": 1, "date_end": now - 10 * tt.RESCAN_END_BUFFER_SECONDS, # long past scheduled end "team_count": 2, "match_count": 1, "status": "active", "scanned_unix": now - tt.RESCAN_INTERVAL_SECONDS - 1, # interval elapsed "matches": [match], "battles": [], "standings": [], } await tt.store_scan(scan) assert await tt.needs_scan(77777, now=now) is True # Once truly finished, never re-scan again. scan["status"] = "finished" await tt.store_scan(scan) assert await tt.needs_scan(77777, now=now) is False asyncio.run(run()) def test_store_scan_roundtrip(): names = tt.team_name_map(SCHEDULE["all_teams"]) matches = [tt.parse_schedule_match(r, names) for r in SCHEDULE["all_matches"]] battles, _ = tt.parse_battles(BATTLES_PLAYED, 99999, "686556", "Looser") scan = { "tournament_id": 99999, "name": "Test Cup", "format": "double-elim", "game_mode": "RB", "cluster": "EU", "date_start": 1, "date_end": 2, "team_count": tt.team_count(matches, []), "match_count": len(matches), "status": "finished", "scanned_unix": 1000, "matches": matches, "battles": battles, "standings": [], } async def run(): await tt.init_tss_tournaments_db() await tt.store_scan(scan) import sqlite3 with sqlite3.connect(tt.TSS_TOURNAMENTS_DB_PATH) as conn: trow = conn.execute( "SELECT name, format, match_count FROM tournaments WHERE tournament_id=99999" ).fetchone() (mcount,) = conn.execute( "SELECT COUNT(*) FROM tournament_matches WHERE tournament_id=99999" ).fetchone() hexes = [ r[0] for r in conn.execute( "SELECT session_hex FROM tournament_battles WHERE tournament_id=99999 ORDER BY position" ).fetchall() ] return trow, mcount, hexes trow, mcount, hexes = asyncio.run(run()) assert trow == ("Test Cup", "double-elim", 2) assert mcount == 2 assert hexes == ["31de23f001a9f8c", "31de25268182000"] # Idempotent re-store replaces children, doesn't duplicate. asyncio.run(_restore_and_count(scan)) async def _restore_and_count(scan): import sqlite3 await tt.store_scan(scan) with sqlite3.connect(tt.TSS_TOURNAMENTS_DB_PATH) as conn: (mcount,) = conn.execute( "SELECT COUNT(*) FROM tournament_matches WHERE tournament_id=99999" ).fetchone() assert mcount == 2 def test_fetch_schedule_sync_has_no_battles_and_makes_no_battle_calls(monkeypatch): calls = [] def fake_request(method, action, **params): calls.append(action) if action == "get_shedule_matches": return {"data": SCHEDULE} return {"data": {}} monkeypatch.setattr(tt, "_request", fake_request) monkeypatch.setattr(tt, "_active_meta", lambda tid, now: ({}, True)) scan = tt.fetch_schedule_sync(12345, now=1000) assert scan["battles"] == [] assert scan["match_count"] == len(SCHEDULE["all_matches"]) assert "getListAllBattles" not in calls # the whole point: no per-match battle loop def test_store_schedule_preserves_battles(): # A schedule refresh updates matches/standings/status but must NOT wipe the # battle links the fast path inserted. names = tt.team_name_map(SCHEDULE["all_teams"]) matches = [tt.parse_schedule_match(r, names) for r in SCHEDULE["all_matches"]] battles, _ = tt.parse_battles(BATTLES_PLAYED, 88888, "686556", "Looser") full = { "tournament_id": 88888, "name": "Cup", "format": "double-elim", "game_mode": "RB", "cluster": "EU", "date_start": 1, "date_end": 2, "team_count": 2, "match_count": len(matches), "status": "active", "scanned_unix": 1000, "matches": matches, "battles": battles, "standings": [], } async def run(): await tt.init_tss_tournaments_db() await tt.store_scan(full) # seeds matches + battles # Now a schedule-only refresh with NO battles in the payload. sched = dict(full, status="finished", scanned_unix=2000, battles=[]) await tt.store_schedule(sched) import sqlite3 with sqlite3.connect(tt.TSS_TOURNAMENTS_DB_PATH) as conn: n_battles = conn.execute( "SELECT COUNT(*) FROM tournament_battles WHERE tournament_id = 88888" ).fetchone()[0] status = conn.execute( "SELECT status FROM tournaments WHERE tournament_id = 88888" ).fetchone()[0] assert n_battles == len(battles) # battles preserved assert status == "finished" # meta updated asyncio.run(run()) def test_request_schedule_refresh_coalesces_burst(monkeypatch): fetched = [] def fake_fetch(tid, **kwargs): fetched.append(tid) return { "tournament_id": tid, "name": "T", "format": "single-elim", "game_mode": "RB", "cluster": "EU", "date_start": 1, "date_end": 2, "team_count": 0, "match_count": 0, "status": "active", "scanned_unix": 1000, "matches": [], "standings": [], "battles": [], } monkeypatch.setattr(tt, "fetch_schedule_sync", fake_fetch) monkeypatch.setattr(tt, "_REFRESH_DEBOUNCE_SECONDS", 0.05) async def run(): await tt.init_tss_tournaments_db() for _ in range(8): # burst of 8 games for the same tournament tt.request_schedule_refresh(60002) await asyncio.sleep(0.2) # let the single debounced refresh fire asyncio.run(run()) assert fetched == [60002] # 8 requests collapsed into one fetch def test_link_battle_sync_links_known_match(): async def run(): await tt.init_tss_tournaments_db() match = { "match_id": "330669", "type_bracket": "Swiss", "side": "swiss", "round": None, "position": None, "team_a_uuid": None, "team_a_name": "NEXOOOS", "team_b_uuid": None, "team_b_name": "GFRIS2", "winner_name": None, "score_a": 0, "score_b": 0, "status": "pending", "time_start": None, } scan = { "tournament_id": 24836, "name": "ACL", "format": "swiss", "game_mode": "RB", "cluster": "EU", "date_start": 1, "date_end": 2, "team_count": 2, "match_count": 1, "status": "active", "scanned_unix": 1000, "matches": [match], "battles": [], "standings": [], } await tt.store_scan(scan) linked = tt.link_battle_sync(24836, "330669", "1abcd", winner_name="NEXOOOS") assert linked is True import sqlite3 with sqlite3.connect(tt.TSS_TOURNAMENTS_DB_PATH) as conn: row = conn.execute( "SELECT type_bracket FROM tournament_battles WHERE session_hex = '1abcd'" ).fetchone() assert row == ("Swiss",) # type_bracket pulled from the stored match asyncio.run(run()) def test_link_battle_sync_returns_false_for_unknown_match(): async def run(): await tt.init_tss_tournaments_db() assert tt.link_battle_sync(999999, "nope", "deadbeef") is False asyncio.run(run()) def test_maybe_link_battle_known_match_makes_no_network_call(monkeypatch): def boom(*a, **k): raise AssertionError("fast path must not call the TSS API") monkeypatch.setattr(tt, "_request", boom) refreshed = [] monkeypatch.setattr(tt, "request_schedule_refresh", lambda tid, **k: refreshed.append(tid)) async def run(): await tt.init_tss_tournaments_db() match = { "match_id": "330669", "type_bracket": "Swiss", "side": "swiss", "round": None, "position": None, "team_a_uuid": None, "team_a_name": "NEXOOOS", "team_b_uuid": None, "team_b_name": "GFRIS2", "winner_name": None, "score_a": 0, "score_b": 0, "status": "pending", "time_start": None, } scan = { "tournament_id": 24836, "name": "ACL", "format": "swiss", "game_mode": "RB", "cluster": "EU", "date_start": 1, "date_end": 2, "team_count": 2, "match_count": 1, "status": "active", "scanned_unix": 1000, "matches": [match], "battles": [], "standings": [], } await tt.store_scan(scan) game = { "_id": "1abcd", "winner": "1", "tss": {"tournament_id": 24836, "match_id": "330669", "bracket": "swiss", "1": {"name": "NEXOOOS"}, "2": {"name": "GFRIS2"}}, } await tt.maybe_link_battle(game) asyncio.run(run()) assert refreshed == [24836] # linked + scheduled a refresh, no network def test_reconcile_targets_only_played_without_battles(): async def run(): await tt.init_tss_tournaments_db() m_played = { "match_id": "p1", "type_bracket": "Winner", "side": "winner", "round": 0, "position": 0, "team_a_uuid": None, "team_a_name": "A", "team_b_uuid": None, "team_b_name": "B", "winner_name": "A", "score_a": 1, "score_b": 0, "status": "played", "time_start": None, } m_pending = dict(m_played, match_id="p2", status="pending", winner_name=None) m_linked = dict(m_played, match_id="p3") battle = { "session_hex": "abc", "session_decimal": "2748", "tournament_id": 70001, "match_id": "p3", "type_bracket": "Winner", "position": 0, "winner_name": "A", "status_replay": "view replay", } scan = { "tournament_id": 70001, "name": "T", "format": "single-elim", "game_mode": "RB", "cluster": "EU", "date_start": 1, "date_end": 2, "team_count": 2, "match_count": 3, "status": "active", "scanned_unix": 1000, "matches": [m_played, m_pending, m_linked], "battles": [battle], "standings": [], } await tt.store_scan(scan) targets = tt.reconcile_targets_sync(70001) ids = {t["match_id"] for t in targets} assert ids == {"p1"} # played + no battle; p2 pending excluded, p3 already linked asyncio.run(run())