NotSoToothless
280 lines
11 KiB
Python
280 lines
11 KiB
Python
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"
|
|
# past end + buffer → finished even with pending matches
|
|
assert tt.compute_status(mixed, 100, 100 + tt.RESCAN_END_BUFFER_SECONDS + 1) == "finished"
|
|
# absent from the active list → finished (old tournament, no date_end)
|
|
assert tt.compute_status(mixed, None, 1000, in_active=False) == "finished"
|
|
# unknown active-list state should not finish a tournament by absence alone
|
|
assert tt.compute_status(mixed, None, 1000, in_active=None) == "active"
|
|
|
|
|
|
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
|