TSSBOT/tests/test_wl.py

import pathlib
import sqlite3
import sys

sys.path.insert(0, str(pathlib.Path(__file__).resolve().parents[1]))
sys.path.insert(0, str(pathlib.Path(__file__).resolve().parents[2] / "SHARED"))

from BOT import wl


def _make_db(path: str, matches, players):
    """Seed the two real tables with the columns get_tss_standings reads.

    matches: list of (session_id, tournament_id)
    players: list of (session_id, team_id, vehicle_internal, victor_bool)
    """
    conn = sqlite3.connect(path)
    conn.execute(
        "CREATE TABLE match_summary (session_id TEXT PRIMARY KEY, tournament_id INTEGER)"
    )
    conn.execute(
        "CREATE TABLE player_games_hist ("
        " UID TEXT, session_id TEXT, team_id INTEGER,"
        " vehicle_internal TEXT, victor_bool TEXT)"
    )
    conn.executemany("INSERT INTO match_summary VALUES (?, ?)", matches)
    conn.executemany(
        "INSERT INTO player_games_hist (UID, session_id, team_id, vehicle_internal, victor_bool)"
        " VALUES ('u', ?, ?, ?, ?)",
        players,
    )
    conn.commit()
    conn.close()


def _point_at(monkeypatch, path):
    monkeypatch.setattr(wl, "_db_path", lambda: str(path))


def test_basic_wins_and_losses(tmp_path, monkeypatch):
    db = tmp_path / "battles.db"
    _make_db(
        str(db),
        matches=[("s1", 100), ("s2", 100), ("s3", 100)],
        players=[
            ("s1", 7, "t34", "Win"),
            ("s2", 7, "t34", "Win"),
            ("s3", 7, "t34", "Loss"),
        ],
    )
    _point_at(monkeypatch, db)
    assert wl.get_tss_standings(7, 100) == {"wins": 2, "losses": 1}


def test_per_tournament_isolation(tmp_path, monkeypatch):
    db = tmp_path / "battles.db"
    # Same team_id appears in two tournaments; records must not bleed across.
    _make_db(
        str(db),
        matches=[("a", 100), ("b", 100), ("c", 200)],
        players=[
            ("a", 7, "t34", "Win"),
            ("b", 7, "t34", "Loss"),
            ("c", 7, "t34", "Win"),
        ],
    )
    _point_at(monkeypatch, db)
    assert wl.get_tss_standings(7, 100) == {"wins": 1, "losses": 1}
    assert wl.get_tss_standings(7, 200) == {"wins": 1, "losses": 0}


def test_draw_is_loss_for_both_teams(tmp_path, monkeypatch):
    db = tmp_path / "battles.db"
    # A draw is persisted as victor_bool='Loss' for both participating teams.
    _make_db(
        str(db),
        matches=[("s1", 100)],
        players=[
            ("s1", 7, "t34", "Loss"),
            ("s1", 9, "pz", "Loss"),
        ],
    )
    _point_at(monkeypatch, db)
    assert wl.get_tss_standings(7, 100) == {"wins": 0, "losses": 1}
    assert wl.get_tss_standings(9, 100) == {"wins": 0, "losses": 1}


def test_vehicle_duplicate_rows_count_once(tmp_path, monkeypatch):
    db = tmp_path / "battles.db"
    # One player on the team flew two vehicles -> two rows, same session/outcome.
    _make_db(
        str(db),
        matches=[("s1", 100)],
        players=[
            ("s1", 7, "t34", "Win"),
            ("s1", 7, "is2", "Win"),
        ],
    )
    _point_at(monkeypatch, db)
    assert wl.get_tss_standings(7, 100) == {"wins": 1, "losses": 0}


def test_string_team_id_is_coerced(tmp_path, monkeypatch):
    db = tmp_path / "battles.db"
    _make_db(str(db), matches=[("s1", 100)], players=[("s1", 7, "t34", "Win")])
    _point_at(monkeypatch, db)
    # The render model carries team_id as a string.
    assert wl.get_tss_standings("7", 100) == {"wins": 1, "losses": 0}


def test_missing_ids_return_none(tmp_path, monkeypatch):
    db = tmp_path / "battles.db"
    _make_db(str(db), matches=[("s1", 100)], players=[("s1", 7, "t34", "Win")])
    _point_at(monkeypatch, db)
    assert wl.get_tss_standings(None, 100) is None
    assert wl.get_tss_standings(7, None) is None
    assert wl.get_tss_standings("not-a-number", 100) is None


def test_unknown_team_returns_none(tmp_path, monkeypatch):
    db = tmp_path / "battles.db"
    _make_db(str(db), matches=[("s1", 100)], players=[("s1", 7, "t34", "Win")])
    _point_at(monkeypatch, db)
    assert wl.get_tss_standings(999, 100) is None


def test_missing_db_returns_none(tmp_path, monkeypatch):
    _point_at(monkeypatch, tmp_path / "does_not_exist.db")
    assert wl.get_tss_standings(7, 100) is None


def test_reads_wal_mode_db_with_open_writer(tmp_path, monkeypatch):
    # The production DB is WAL. Reproduce a live-ingest state: WAL journal, no
    # autocheckpoint (so a non-empty -wal/-shm lingers), and a writer connection
    # still open while we read — get_tss_standings must still return the record.
    db = tmp_path / "battles.db"
    _make_db(
        str(db),
        matches=[("s1", 100), ("s2", 100)],
        players=[("s1", 7, "t34", "Win"), ("s2", 7, "t34", "Loss")],
    )
    writer = sqlite3.connect(str(db))
    writer.execute("PRAGMA journal_mode=WAL")
    writer.execute("PRAGMA wal_autocheckpoint=0")
    writer.execute(
        "INSERT INTO player_games_hist (UID, session_id, team_id, vehicle_internal, victor_bool)"
        " VALUES ('u', 's3', 7, 't34', 'Win')"
    )
    writer.execute("INSERT INTO match_summary VALUES ('s3', 100)")
    writer.commit()  # committed to the WAL, not checkpointed; writer stays open
    try:
        _point_at(monkeypatch, db)
        assert wl.get_tss_standings(7, 100) == {"wins": 2, "losses": 1}
    finally:
        writer.close()