Add persistence layer: TOTP auth, game history, restore
- db/ package: async SQLAlchemy engine + Player/Game/Guess models - api/auth.py: passwordless TOTP login (pyotp), session token via socket auth - api/history.py: record guesses/points, DB-backed standings, restore unfinished games on startup, host-only end_game - api/__init__.py: auth-gated handlers, accounts map, rejoin via account - frontend: Auth (QR + code) and History pages, resume/end-game in lobby/table - docker-compose: real PostgreSQL service wired via DATABASE_URL - tests_history.py for the persistence/auth layer; refresh CLAUDE.md Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,180 @@
|
||||
"""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 types import SimpleNamespace
|
||||
|
||||
# Nastav DB 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("\\", "/")
|
||||
|
||||
import pyotp # noqa: E402
|
||||
|
||||
from api import auth, history # noqa: E402
|
||||
from db.db import init_db # 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)
|
||||
|
||||
|
||||
class HistoryCase(unittest.TestCase):
|
||||
@classmethod
|
||||
def setUpClass(cls):
|
||||
run(init_db())
|
||||
|
||||
def _make_players(self, n=4):
|
||||
ids = []
|
||||
for _ in range(n):
|
||||
username = "u_" + uuid.uuid4().hex[:8]
|
||||
data = run(auth.register_account(username))
|
||||
ident = run(auth.login(username, pyotp.TOTP(data["secret"]).now()))
|
||||
ids.append(ident["player_id"])
|
||||
return ids
|
||||
|
||||
def test_register_login_token(self):
|
||||
username = "alice_" + uuid.uuid4().hex[:6]
|
||||
data = run(auth.register_account(username))
|
||||
self.assertIn("otpauth_uri", data)
|
||||
|
||||
# Zle meno je obsadene
|
||||
with self.assertRaises(auth.AuthError):
|
||||
run(auth.register_account(username))
|
||||
|
||||
code = pyotp.TOTP(data["secret"]).now()
|
||||
ident = run(auth.login(username, code))
|
||||
self.assertEqual(ident["username"], username)
|
||||
self.assertTrue(ident["token"])
|
||||
|
||||
# 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_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_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 = run(history.get_standings(gid))
|
||||
# 1 seria, 1 kolo, body podla sedadiel zo stubu [12, 0, 10, 11].
|
||||
self.assertEqual(standings, [[[12, 0, 10, 11]]])
|
||||
|
||||
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 tearDownModule():
|
||||
from db.db import engine
|
||||
|
||||
run(engine.dispose())
|
||||
try:
|
||||
os.remove(_DB_FILE)
|
||||
except OSError:
|
||||
pass
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user