Files
bridzik/tests/test_history.py
T
timandClaude Sonnet 5.5 04846aee26 pridanie rebricka (leaderboard)
Nova polozka menu Rebricek a stranka /leaderboard: top 10 podla najlepsej hry (hry len s ludmi / s aspon jednym botom) a podla hodin v hre (hry do 3 h). Backend api/leaderboard.py + socket udalost get_leaderboard, testy v tests/test_history.py.

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_019YzEAACE9VgnLtXPuWGjwy
2026-10-26 22:06:50 +01:00

479 lines
20 KiB
Python

"""Testy persistentnej vrstvy (db/ + api/history.py + api/auth.py).
Bezia na docasnom SQLite subore. Engine sa nepouziva priamo -- pre zapisovu
logiku staci lahky stub, ktory zrkadli rozhranie bridzik.Bridzik (series ->
rounds -> guesses/get_points_summary). Cisty engine ma vlastne testy v tests.py.
"""
import asyncio
import os
import tempfile
import unittest
import uuid
from datetime import datetime, timedelta
from types import SimpleNamespace
# Nastav DB/ENCRYPTION_KEY PRED importom db/api modulov -- engine sa vytvara pri importe.
_DB_FILE = os.path.join(tempfile.gettempdir(), f"bridzik_test_{uuid.uuid4().hex}.db")
os.environ["DATABASE_URL"] = "sqlite+aiosqlite:///" + _DB_FILE.replace("\\", "/")
from cryptography.fernet import Fernet # noqa: E402
os.environ.setdefault("ENCRYPTION_KEY", Fernet.generate_key().decode())
import pyotp # noqa: E402
from sqlalchemy import select # noqa: E402
from api import auth, history, leaderboard # noqa: E402
from api import bots as bots_module # noqa: E402
from bridzik import ROUNDS_PER_SERIES # noqa: E402
from db import crypto # noqa: E402
from db.db import async_session, init_db # noqa: E402
from db.models import Game, Guess, Player # noqa: E402
def run(coro):
return asyncio.run(coro)
def make_core(completed=True):
"""Stub jednej hry s jednym dohratym kolom (seria 0, kolo 0)."""
rnd = SimpleNamespace(
round_number=0,
guesses={0: 2, 1: 1, 2: 0, 3: 1},
is_completed=lambda: True,
get_points_summary=lambda: [12, 0, 10, 11],
)
series = SimpleNamespace(
series_number=0, rounds=[rnd], get_last_round=lambda: rnd
)
return SimpleNamespace(series=[series], is_completed=lambda: completed)
def make_players(n=4):
"""Zalozi `n` potvrdenych uctov, vrati ich player_id."""
ids = []
for _ in range(n):
username = "u_" + uuid.uuid4().hex[:8]
data = run(auth.register_account(username))
ident = run(auth.confirm_account(username, pyotp.TOTP(data["secret"]).now()))
ids.append(ident["player_id"])
return ids
class HistoryCase(unittest.TestCase):
@classmethod
def setUpClass(cls):
run(init_db())
def _make_players(self, n=4):
return make_players(n)
@staticmethod
def _next_step_code(secret):
"""Kod pre NASLEDUJUCI casovy krok -- confirm uz spotreboval aktualny."""
return pyotp.TOTP(secret).at((auth._current_step() + 1) * auth.TOTP_PERIOD)
def test_register_login_token(self):
username = "alice_" + uuid.uuid4().hex[:6]
data = run(auth.register_account(username))
self.assertIn("otpauth_uri", data)
# Nedokoncena registracia: opakovany register vyda NOVY secret
data = run(auth.register_account(username))
self.assertIn("otpauth_uri", data)
# Login pred potvrdenim -> RegistrationIncomplete (novy QR namiesto chyby)
with self.assertRaises(auth.RegistrationIncomplete):
run(auth.login(username, "000000"))
code = pyotp.TOTP(data["secret"]).now()
ident = run(auth.confirm_account(username, code))
self.assertEqual(ident["username"], username)
self.assertTrue(ident["token"])
# Po potvrdeni je uz meno obsadene
with self.assertRaises(auth.AuthError):
run(auth.register_account(username))
# Token sa da spatne rozlustit na identitu
resolved = run(auth.player_by_token(ident["token"]))
self.assertEqual(resolved["player_id"], ident["player_id"])
# Zly kod neprejde
with self.assertRaises(auth.AuthError):
run(auth.login(username, "000000"))
def test_reissued_secret_invalidates_old_qr(self):
username = "fred_" + uuid.uuid4().hex[:6]
old = run(auth.register_account(username))
new = run(auth.register_account(username))
self.assertNotEqual(old["secret"], new["secret"])
# Kod zo stareho QR uz neplati, z noveho ano
with self.assertRaises(auth.AuthError):
run(auth.confirm_account(username, pyotp.TOTP(old["secret"]).now()))
ident = run(auth.confirm_account(username, pyotp.TOTP(new["secret"]).now()))
self.assertEqual(ident["username"], username)
def test_login_lockout_after_repeated_failures(self):
username = "bob_" + uuid.uuid4().hex[:6]
data = run(auth.register_account(username))
run(auth.confirm_account(username, pyotp.TOTP(data["secret"]).now()))
for _ in range(auth._LOGIN_ATTEMPT_LIMIT):
with self.assertRaises(auth.AuthError):
run(auth.login(username, "000000"))
# Lockout odmietne aj spravny kod, kym neubehne okno
with self.assertRaises(auth.AuthError):
run(auth.login(username, self._next_step_code(data["secret"])))
def test_successful_login_clears_failed_attempts(self):
username = "carol_" + uuid.uuid4().hex[:6]
data = run(auth.register_account(username))
run(auth.confirm_account(username, pyotp.TOTP(data["secret"]).now()))
for _ in range(auth._LOGIN_ATTEMPT_LIMIT - 1):
with self.assertRaises(auth.AuthError):
run(auth.login(username, "000000"))
ident = run(auth.login(username, self._next_step_code(data["secret"])))
self.assertEqual(ident["username"], username)
self.assertNotIn(username, auth._failed_attempts)
def test_record_rounds_and_idempotency(self):
ids = self._make_players()
gid = str(uuid.uuid4())
core = make_core()
run(history.record_game_started(gid, "Test", ids))
run(history.record_completed_rounds(gid, core))
# Opakovany zapis nesmie zalozit duplikaty.
run(history.record_completed_rounds(gid, core))
detail = run(history.get_game_detail(gid))
self.assertIsNotNone(detail)
self.assertEqual(len(detail["rounds"]), 4) # 4 hraci x 1 kolo
by_seat = {r["player_id"]: r for r in detail["rounds"]}
# Body podla get_points_summary; won = points > 0.
self.assertEqual(by_seat[ids[0]]["points"], 12)
self.assertTrue(by_seat[ids[0]]["won"])
self.assertEqual(by_seat[ids[1]]["points"], 0)
self.assertFalse(by_seat[ids[1]]["won"])
self.assertIsNotNone(detail["ended_at"]) # core.is_completed() -> ended
def test_rebuild_core_position(self):
# Z dvoch cisel sa obnovi spravna pozicia a karty sa rozdaju nanovo.
core = history.rebuild_core(2, 3)
self.assertEqual(core.series[-1].series_number, 2)
last_round = core.series[-1].get_last_round()
self.assertEqual(last_round.round_number, 3)
# V kole 3 ma kazdy hrac 8 - 3 = 5 kariet.
self.assertEqual(len(core.get_player_cards(0)), 5)
def test_restore_game_from_db(self):
ids = self._make_players()
gid = str(uuid.uuid4())
run(history.record_game_started(gid, "Test", ids))
run(history.record_completed_rounds(gid, make_core())) # zapise poziciu
restored = run(history.restore_game_core(gid))
self.assertIsNotNone(restored)
# make_core() je seria 0, kolo 0 -> pozicia 0/0
self.assertEqual(restored.series[-1].series_number, 0)
self.assertEqual(restored.series[-1].get_last_round().round_number, 0)
def test_unfinished_games_listed(self):
ids = self._make_players()
gid = str(uuid.uuid4())
run(history.record_game_started(gid, "Nedohrata", ids))
run(history.record_completed_rounds(gid, make_core(completed=False)))
rows = run(history.get_unfinished_games())
mine = next((g for g in rows if g["gid"] == gid), None)
self.assertIsNotNone(mine)
self.assertEqual(mine["name"], "Nedohrata")
self.assertEqual(len(mine["seats"]), 4)
self.assertEqual(mine["seats"][0][0], ids[0]) # player_id na sedadle 0
self.assertIsNotNone(mine["core"]) # uz postaveny Bridzik
def test_mark_game_ended_removes_from_unfinished(self):
ids = self._make_players()
gid = str(uuid.uuid4())
run(history.record_game_started(gid, "Vzdana", ids))
run(history.record_completed_rounds(gid, make_core(completed=False)))
self.assertTrue(any(g["gid"] == gid for g in run(history.get_unfinished_games())))
run(history.mark_game_ended(gid))
self.assertFalse(any(g["gid"] == gid for g in run(history.get_unfinished_games())))
def test_reopen_prematurely_ended_game(self):
ids = self._make_players()
gid = str(uuid.uuid4())
run(history.record_game_started(gid, "Vzdana", ids))
run(history.record_completed_rounds(gid, make_core(completed=False)))
run(history.mark_game_ended(gid))
# V historii sa ukazuje ako predcasne ukoncena (nie naplno dohrana).
rows = run(history.get_player_history(ids[0]))
mine = next(g for g in rows if g["gid"] == gid)
self.assertFalse(mine["completed"])
# Cudzi hrac ju obnovit nemoze.
outsider = self._make_players(1)[0]
self.assertIsNone(run(history.reopen_game(gid, outsider)))
# Clen ju obnovi -> ended_at sa zmaze a hra je zas medzi nedohratymi.
info = run(history.reopen_game(gid, ids[0]))
self.assertIsNotNone(info)
self.assertEqual(len(info["seats"]), 4)
self.assertTrue(any(g["gid"] == gid for g in run(history.get_unfinished_games())))
# A teda uz nie je v historii (zobrazuju sa iba ukoncene hry).
self.assertFalse(any(g["gid"] == gid for g in run(history.get_player_history(ids[0]))))
def test_standings_from_db(self):
ids = self._make_players()
gid = str(uuid.uuid4())
run(history.record_game_started(gid, "Test", ids))
run(history.record_completed_rounds(gid, make_core()))
standings, guesses = run(history.get_standings(gid))
# 1 seria, 1 kolo. Body podla sedadiel zo stubu [12, 0, 10, 11],
# tipy zo stubu {0: 2, 1: 1, 2: 0, 3: 1}.
self.assertEqual(standings, [[[12, 0, 10, 11]]])
self.assertEqual(guesses, [[[2, 1, 0, 1]]])
def test_player_history(self):
ids = self._make_players()
gid = str(uuid.uuid4())
run(history.record_game_started(gid, "Test", ids))
run(history.record_completed_rounds(gid, make_core()))
rows = run(history.get_player_history(ids[0]))
self.assertTrue(any(g["gid"] == gid for g in rows))
mine = next(g for g in rows if g["gid"] == gid)
self.assertEqual(mine["my_points"], 12)
self.assertEqual(len(mine["players"]), 4)
def test_totp_secret_stored_encrypted_not_plaintext(self):
username = "dave_" + uuid.uuid4().hex[:6]
data = run(auth.register_account(username))
async def _raw_secret():
async with async_session() as session:
return await session.scalar(
select(Player.totp_secret).where(Player.username == username)
)
stored = run(_raw_secret())
self.assertNotEqual(stored, data["secret"])
self.assertEqual(crypto.decrypt(stored), data["secret"])
def test_auth_token_stored_hashed_not_plaintext(self):
username = "erin_" + uuid.uuid4().hex[:6]
data = run(auth.register_account(username))
ident = run(auth.confirm_account(username, pyotp.TOTP(data["secret"]).now()))
async def _raw_token():
async with async_session() as session:
return await session.scalar(
select(Player.auth_token).where(Player.username == username)
)
stored = run(_raw_token())
self.assertNotEqual(stored, ident["token"])
self.assertEqual(stored, crypto.hash_token(ident["token"]))
class LeaderboardCase(unittest.TestCase):
"""Pravidla rebrickov (api/leaderboard.py): radenie nad syntetickymi datami,
plus jeden integracny test cez DB."""
UN = {1: "ann", 2: "bob", 3: "cyd", 4: "dee", 5: "bot:heuristic-1", 6: "bot:heuristic-2"}
@staticmethod
def game(created_h, ended_h, has_bot=False, day=1):
base = datetime(2026, 10, day)
return {
"created_at": base + timedelta(hours=created_h),
"ended_at": base + timedelta(hours=ended_h),
"has_bot": has_bot,
}
def board(self, results, games, me=1):
return leaderboard.build_leaderboard(results, games, self.UN, me)
def test_points_split_humans_and_bots_and_hide_bots(self):
games = {"h": self.game(10, 11), "b": self.game(12, 13, has_bot=True)}
results = [
("h", 1, 200), ("h", 2, 150), ("h", 3, 100), ("h", 4, 50),
("b", 1, 300), ("b", 2, 120), ("b", 3, 90), ("b", 5, 400),
]
lb = self.board(results, games)
humans = lb["points"]["humans"]["top"]
self.assertEqual([(e["username"], e["points"]) for e in humans],
[("ann", 200), ("bob", 150), ("cyd", 100), ("dee", 50)])
bots = lb["points"]["bots"]["top"]
# bot (400) sa v rebricku nezobrazi, hrac ann ma v botej hre svojich 300
self.assertEqual([(e["username"], e["points"]) for e in bots],
[("ann", 300), ("bob", 120), ("cyd", 90)])
self.assertEqual([e["rank"] for e in bots], [1, 2, 3])
def test_points_one_row_per_player_best_game(self):
games = {"a": self.game(10, 11), "b": self.game(12, 13, day=2)}
results = [("a", 1, 100), ("b", 1, 180), ("a", 2, 150)]
top = self.board(results, games)["points"]["humans"]["top"]
self.assertEqual([(e["username"], e["points"]) for e in top],
[("ann", 180), ("bob", 150)])
def test_points_tie_goes_to_earlier_game(self):
games = {"early": self.game(10, 11, day=1), "late": self.game(10, 11, day=2)}
results = [("late", 1, 150), ("early", 2, 150)]
top = self.board(results, games)["points"]["humans"]["top"]
self.assertEqual([e["username"] for e in top], ["bob", "ann"])
def test_me_outside_top_n_gets_real_rank(self):
games = {"g": self.game(10, 11)}
usernames = {i: f"p{i}" for i in range(1, 16)}
results = [("g", i, 1000 - i) for i in range(1, 16)]
lb = leaderboard.build_leaderboard(results, games, usernames, 12)
board = lb["points"]["humans"]
self.assertEqual(len(board["top"]), leaderboard.TOP_N)
self.assertEqual(board["me"]["rank"], 12)
self.assertEqual(board["me"]["points"], 988)
self.assertNotIn(12, [e["player_id"] for e in board["top"]])
def test_me_is_none_without_a_game(self):
games = {"g": self.game(10, 11)}
lb = self.board([("g", 2, 100)], games, me=1)
self.assertIsNone(lb["points"]["humans"]["me"])
self.assertIsNone(lb["points"]["bots"]["me"])
self.assertIsNone(lb["hours"]["me"])
def test_hours_sum_includes_bot_games_and_counts_games(self):
games = {
"a": self.game(10, 11), # 1 h
"b": self.game(12, 12.5, has_bot=True), # 30 min, s botom sa pocita
}
results = [("a", 1, 10), ("b", 1, 10), ("b", 5, 10), ("a", 2, 10)]
hours = self.board(results, games)["hours"]
self.assertEqual([e["username"] for e in hours["top"]], ["ann", "bob"])
self.assertEqual(hours["top"][0]["seconds"], 5400)
self.assertEqual(hours["top"][0]["games"], 2)
self.assertEqual(hours["top"][1]["seconds"], 3600)
def test_hours_game_across_midnight_counts_but_long_game_does_not(self):
games = {
"night": { # 23:30 -> 00:15 nasledujuceho dna = 45 min
"created_at": datetime(2026, 10, 1, 23, 30),
"ended_at": datetime(2026, 10, 2, 0, 15),
"has_bot": False,
},
"resumed": { # obnovena o den neskor -> pauza by skreslila cas
"created_at": datetime(2026, 10, 1, 20, 0),
"ended_at": datetime(2026, 10, 2, 20, 30),
"has_bot": False,
},
}
results = [("night", 1, 10), ("resumed", 1, 10), ("resumed", 2, 10)]
hours = self.board(results, games)["hours"]
self.assertEqual([(e["username"], e["seconds"]) for e in hours["top"]],
[("ann", 2700)])
def test_hours_game_at_the_limit_counts(self):
games = {"g": self.game(10, 13)}
hours = self.board([("g", 1, 10)], games)["hours"]
self.assertEqual(hours["top"][0]["seconds"], 3 * 3600)
def test_bots_never_in_hours(self):
games = {"g": self.game(10, 11, has_bot=True)}
hours = self.board([("g", 1, 10), ("g", 5, 10), ("g", 6, 10)], games)["hours"]
self.assertEqual([e["username"] for e in hours["top"]], ["ann"])
# --- integracia s DB -----------------------------------------------------
@classmethod
def setUpClass(cls):
run(init_db())
@staticmethod
async def _add_game(gid, ids, totals, created, ended, rounds=history.FULL_GAME_ROUNDS):
"""Hra so `rounds` dohratymi kolami; cely sucet bodov hraca je v kole 0."""
async with async_session() as session:
session.add(Game(
id=gid, name="LB", player0_id=ids[0], player1_id=ids[1],
player2_id=ids[2], player3_id=ids[3], created_at=created, ended_at=ended,
))
for n in range(rounds):
for seat in range(4):
session.add(Guess(
game_id=gid, player_id=ids[seat], series_number=n // ROUNDS_PER_SERIES,
round_number=n % ROUNDS_PER_SERIES, guess=0, points=totals[seat] if n == 0 else 0,
))
await session.commit()
def test_get_leaderboard_from_db(self):
humans = make_players(4)
bot = run(bots_module.ensure_bot_account("heuristic", set()))
mixed = humans[:3] + [bot["player_id"]]
start = datetime(2026, 11, 1, 10, 0)
# hry s obrovskymi bodmi, nech su urcite na vrchu aj popri inych testoch
run(self._add_game(str(uuid.uuid4()), humans, [9001, 9002, 9003, 9004],
start, start + timedelta(hours=1)))
run(self._add_game(str(uuid.uuid4()), mixed, [9101, 9102, 9103, 99999],
start, start + timedelta(hours=2)))
# nedohrana hra (len 1 kolo) sa nepocita
run(self._add_game(str(uuid.uuid4()), humans, [99990, 0, 0, 0],
start, start + timedelta(hours=1), rounds=1))
lb = run(leaderboard.get_leaderboard(humans[0]))
hum = lb["points"]["humans"]["top"]
self.assertEqual([e["player_id"] for e in hum[:4]],
[humans[3], humans[2], humans[1], humans[0]])
self.assertEqual(hum[0]["points"], 9004)
bts = lb["points"]["bots"]["top"]
self.assertEqual([e["player_id"] for e in bts[:3]],
[humans[2], humans[1], humans[0]])
self.assertNotIn(bot["player_id"], [e["player_id"] for e in bts])
self.assertEqual(lb["points"]["humans"]["me"]["points"], 9001)
# hodiny: humans[0] odohral 1 h + 2 h, humans[3] len 1 h
me_hours = lb["hours"]["me"]
self.assertEqual(me_hours["seconds"], 3 * 3600)
self.assertEqual(me_hours["games"], 2)
class CryptoCase(unittest.TestCase):
def test_encrypt_decrypt_roundtrip(self):
secret = "JBSWY3DPEHPK3PXP"
ciphertext = crypto.encrypt(secret)
self.assertNotEqual(ciphertext, secret)
self.assertEqual(crypto.decrypt(ciphertext), secret)
def test_hash_token_is_deterministic_and_distinct(self):
self.assertEqual(crypto.hash_token("abc"), crypto.hash_token("abc"))
self.assertNotEqual(crypto.hash_token("abc"), crypto.hash_token("abd"))
def test_missing_key_raises(self):
saved = os.environ.pop("ENCRYPTION_KEY")
try:
with self.assertRaises(RuntimeError):
crypto.encrypt("x")
finally:
os.environ["ENCRYPTION_KEY"] = saved
def tearDownModule():
from db.db import engine
run(engine.dispose())
try:
os.remove(_DB_FILE)
except OSError:
pass
if __name__ == "__main__":
unittest.main()