Tidy up root-level files: drop dead config.py, group tests into tests/
- Remove config.py, an unused Flask SECRET_KEY leftover from before the legacy HTTP backend was replaced by the Socket.IO/ASGI server. - Move tests.py / tests_history.py / test_socket.py into a tests/ package as test_engine.py / test_history.py / test_socket.py, and update CLAUDE.md's documented commands to match. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,356 @@
|
||||
import unittest
|
||||
from bridzik import Stash, cards, Card_colors, Card_values,\
|
||||
RuleException, BridzikException, Card, Round
|
||||
|
||||
class StashCase(unittest.TestCase):
|
||||
def test_first_card(self):
|
||||
heart_7 = Card(Card_colors['HEARTS'], Card_values['C7'])
|
||||
s = Stash(3)
|
||||
s.add_card(3, heart_7)
|
||||
self.assertEqual(s.get_first_card(), heart_7)
|
||||
|
||||
heart_8 = Card(Card_colors['HEARTS'], Card_values['C8'])
|
||||
s.add_card(1, heart_8)
|
||||
self.assertNotEqual(s.get_first_card(), heart_8)
|
||||
self.assertEqual(s.get_first_card(), heart_7)
|
||||
|
||||
def test_add_cart(self):
|
||||
heart_7 = Card(Card_colors['HEARTS'], Card_values['C7'])
|
||||
leaves_7 = Card(Card_colors['LEAVES'], Card_values['C7'])
|
||||
leaves_ace = Card(Card_colors['LEAVES'], Card_values['ACE'])
|
||||
bells_10 = Card(Card_colors['BELLS'], Card_values['C10'])
|
||||
acorns_10 = Card(Card_colors['ACORNS'], Card_values['C10'])
|
||||
s = Stash(1)
|
||||
|
||||
s.add_card(1, heart_7)
|
||||
self.assertEqual(s._cards[1], heart_7)
|
||||
s.add_card(2, leaves_7)
|
||||
self.assertEqual(s._cards[2], leaves_7)
|
||||
s.add_card(3, leaves_ace)
|
||||
self.assertEqual(s._cards[3], leaves_ace)
|
||||
s.add_card(0, bells_10)
|
||||
self.assertEqual(s._cards[0], bells_10)
|
||||
|
||||
self.assertRaises(BridzikException, s.add_card, card=bells_10, player=4)
|
||||
self.assertRaises(BridzikException, s.add_card, card=acorns_10, player=0)
|
||||
|
||||
def test_get_winner(self):
|
||||
heart_7 = Card(Card_colors['HEARTS'], Card_values['C7'])
|
||||
leaves_7 = Card(Card_colors['LEAVES'], Card_values['C7'])
|
||||
acorns_ace = Card(Card_colors['ACORNS'], Card_values['ACE'])
|
||||
bells_10 = Card(Card_colors['BELLS'], Card_values['C10'])
|
||||
|
||||
# only heart takes stash
|
||||
c1 = [leaves_7, heart_7, acorns_ace, bells_10]
|
||||
s1 = Stash(0)
|
||||
for i in range(4):
|
||||
s1.add_card(i, c1[i])
|
||||
self.assertEqual(s1.get_winner(), 1)
|
||||
|
||||
# highest color takes the stash
|
||||
leaves_8 = Card(Card_colors['LEAVES'], Card_values['C8'])
|
||||
leaves_lower = Card(Card_colors['LEAVES'], Card_values['LOWER'])
|
||||
c2 = [leaves_8, leaves_7, acorns_ace, leaves_lower]
|
||||
s2 = Stash(0)
|
||||
for i in range(4):
|
||||
s2.add_card(i, c2[i])
|
||||
self.assertEqual(s2.get_winner(), 3)
|
||||
|
||||
# no matching color and no heart in stash
|
||||
c3 = [bells_10, leaves_lower, acorns_ace, leaves_8]
|
||||
s3 = Stash(0)
|
||||
for i in range(4):
|
||||
s3.add_card(i, c3[i])
|
||||
self.assertEqual(s3.get_winner(), 0)
|
||||
|
||||
# highest heart takes the stash
|
||||
heart_upper = Card(Card_colors['HEARTS'], Card_values['UPPER'])
|
||||
c4 = [leaves_8, leaves_lower, heart_upper, heart_7]
|
||||
s4 = Stash(0)
|
||||
for i in range(4):
|
||||
s4.add_card(i, c4[i])
|
||||
self.assertEqual(s4.get_winner(), 2)
|
||||
|
||||
# test exceptions
|
||||
s5 = Stash(3)
|
||||
self.assertRaises(BridzikException, s5.get_winner)
|
||||
|
||||
def test_get_cards(self):
|
||||
heart_7 = Card(Card_colors['HEARTS'], Card_values['C7'])
|
||||
leaves_7 = Card(Card_colors['LEAVES'], Card_values['C7'])
|
||||
acorns_ace = Card(Card_colors['ACORNS'], Card_values['ACE'])
|
||||
bells_10 = Card(Card_colors['BELLS'], Card_values['C10'])
|
||||
s = Stash(0)
|
||||
self.assertEqual(s.get_cards(), {})
|
||||
s.add_card(2, heart_7)
|
||||
self.assertEqual(s.get_cards(), {2: heart_7})
|
||||
s.add_card(3, leaves_7)
|
||||
self.assertEqual(s.get_cards(), {2: heart_7, 3: leaves_7})
|
||||
s.add_card(0, acorns_ace)
|
||||
self.assertEqual(s.get_cards(), {2: heart_7, 3: leaves_7, 0: acorns_ace})
|
||||
s.add_card(1, bells_10)
|
||||
self.assertEqual(s.get_cards(), {2: heart_7, 3: leaves_7, 0: acorns_ace, 1: bells_10})
|
||||
|
||||
def test_is_complete(self):
|
||||
heart_7 = Card(Card_colors['HEARTS'], Card_values['C7'])
|
||||
leaves_7 = Card(Card_colors['LEAVES'], Card_values['C7'])
|
||||
acorns_ace = Card(Card_colors['ACORNS'], Card_values['ACE'])
|
||||
bells_10 = Card(Card_colors['BELLS'], Card_values['C10'])
|
||||
s = Stash(0)
|
||||
c = [heart_7, leaves_7, acorns_ace, bells_10]
|
||||
for i in range(4):
|
||||
self.assertFalse(s.is_completed())
|
||||
s.add_card(i, c[0])
|
||||
self.assertTrue(s.is_completed())
|
||||
|
||||
def test_get_active_player(self):
|
||||
heart_7 = Card(Card_colors['HEARTS'], Card_values['C7'])
|
||||
leaves_7 = Card(Card_colors['LEAVES'], Card_values['C7'])
|
||||
leaves_ace = Card(Card_colors['LEAVES'], Card_values['ACE'])
|
||||
bells_10 = Card(Card_colors['BELLS'], Card_values['C10'])
|
||||
s = Stash(1)
|
||||
|
||||
self.assertEqual(s.get_active_player(), 1)
|
||||
s.add_card(1, heart_7)
|
||||
self.assertEqual(s.get_active_player(), 2)
|
||||
s.add_card(2, leaves_7)
|
||||
self.assertEqual(s.get_active_player(), 3)
|
||||
s.add_card(3, leaves_ace)
|
||||
self.assertEqual(s.get_active_player(), 0)
|
||||
s.add_card(0, bells_10)
|
||||
self.assertRaises(BridzikException, s.get_active_player)
|
||||
|
||||
|
||||
class RoundCase(unittest.TestCase):
|
||||
def test_round_constructor(self):
|
||||
self.assertRaises(BridzikException, Round, round_number=8, first_player=0)
|
||||
self.assertRaises(BridzikException, Round, round_number=0, first_player=4)
|
||||
|
||||
for first_player in range(4):
|
||||
for round_n in range(8):
|
||||
r = Round(round_n, first_player)
|
||||
self.assertEqual(r.round_number, round_n)
|
||||
self.assertEqual(r.first_player, first_player)
|
||||
self.assertEqual(r.guesses, {})
|
||||
self.assertEqual(r.stashes, [])
|
||||
self.assertEqual(len(r.round_cards), 4*(8-round_n))
|
||||
for player_n in range(4):
|
||||
self.assertEqual(
|
||||
r.round_cards[player_n * (8-round_n) : (player_n+1) * (8-round_n)],
|
||||
r.player_cards[player_n]
|
||||
)
|
||||
|
||||
def test_add_player_guess(self):
|
||||
r = Round(0, 1)
|
||||
|
||||
self.assertRaises(BridzikException, r.add_player_guess, player=2, guess=0)
|
||||
|
||||
r.add_player_guess(1, 1)
|
||||
self.assertEqual(r.guesses[1], 1)
|
||||
self.assertRaises(BridzikException, r.add_player_guess, player=1, guess=0)
|
||||
|
||||
r.add_player_guess(2, 0)
|
||||
self.assertEqual(r.guesses[2], 0)
|
||||
r.add_player_guess(3, 2)
|
||||
self.assertEqual(r.guesses[3], 2)
|
||||
|
||||
self.assertEqual(r.stashes, [])
|
||||
|
||||
self.assertRaises(BridzikException, r.add_player_guess, player=0, guess=5)
|
||||
|
||||
r.add_player_guess(0, 4)
|
||||
self.assertEqual(r.guesses[0], 4)
|
||||
self.assertEqual(r.stashes[0].first_player, 0)
|
||||
|
||||
def test_get_highest_guessing_player(self):
|
||||
r1 = Round(0, 0)
|
||||
r1.add_player_guess(0, 2)
|
||||
r1.add_player_guess(1, 1)
|
||||
r1.add_player_guess(2, 3)
|
||||
self.assertRaises(BridzikException, r1.get_highest_guessing_player)
|
||||
r1.add_player_guess(3, 4)
|
||||
self.assertEqual(r1.get_highest_guessing_player(), 3)
|
||||
|
||||
r2 = Round(0, 2)
|
||||
r2.add_player_guess(2, 5)
|
||||
r2.add_player_guess(3, 0)
|
||||
r2.add_player_guess(0, 1)
|
||||
r2.add_player_guess(1, 1)
|
||||
self.assertEqual(r2.get_highest_guessing_player(), 2)
|
||||
|
||||
def test_get_active_player(self):
|
||||
r = Round(6, 1)
|
||||
self.assertEqual(r.get_active_player(), 1)
|
||||
r.add_player_guess(1, 1)
|
||||
self.assertEqual(r.get_active_player(), 2)
|
||||
r.add_player_guess(2, 0)
|
||||
self.assertEqual(r.get_active_player(), 3)
|
||||
r.add_player_guess(3, 1)
|
||||
self.assertEqual(r.get_active_player(), 0)
|
||||
r.add_player_guess(0, 2)
|
||||
self.assertEqual(r.get_active_player(), 0)
|
||||
|
||||
r.play_card(0, r.player_cards[0][0])
|
||||
self.assertEqual(r.get_active_player(), 1)
|
||||
|
||||
def test_play_card(self):
|
||||
shuffler = lambda list: None
|
||||
c0 = [
|
||||
Card(Card_colors['BELLS'], Card_values['UPPER']),
|
||||
Card(Card_colors['HEARTS'], Card_values['UPPER'])
|
||||
]
|
||||
c1 = [
|
||||
Card(Card_colors['BELLS'], Card_values['C7']),
|
||||
Card(Card_colors['HEARTS'], Card_values['C10'])
|
||||
]
|
||||
c2 = [
|
||||
Card(Card_colors['BELLS'], Card_values['ACE']),
|
||||
Card(Card_colors['BELLS'], Card_values['C8'])
|
||||
]
|
||||
c3 = [
|
||||
Card(Card_colors['LEAVES'], Card_values['C7']),
|
||||
Card(Card_colors['BELLS'], Card_values['LOWER'])
|
||||
]
|
||||
c = ['dummy']*24 + c0 + c1 + c2 + c3
|
||||
r = Round(6, 1, c, shuffler)
|
||||
r.add_player_guess(1, 0)
|
||||
r.add_player_guess(2, 0)
|
||||
r.add_player_guess(3, 1)
|
||||
self.assertRaises(BridzikException, r.play_card, player=0, card=c0[0]) # neukoncene tipovanie
|
||||
r.add_player_guess(0, 2)
|
||||
self.assertRaises(BridzikException, r.play_card, player=1, card=c1[0]) # mimo poradia
|
||||
|
||||
r.play_card(0, c0[0])
|
||||
self.assertRaises(BridzikException, r.play_card, player=1, card=c1[1]) # cerven namiesto farby
|
||||
r.play_card(1, c1[0])
|
||||
r.play_card(2, c2[0])
|
||||
self.assertRaises(BridzikException, r.play_card, player=3, card=c3[0]) # ina farba
|
||||
r.play_card(3, c3[1])
|
||||
|
||||
r.play_card(2, c2[1])
|
||||
r.play_card(3, c3[0])
|
||||
self.assertRaises(BridzikException, r.play_card, player=0, card=c0[0]) # uz zahrana karta
|
||||
r.play_card(0, c0[1])
|
||||
r.play_card(1, c1[1])
|
||||
|
||||
self.assertTrue(len(r.stashes), 2)
|
||||
|
||||
def test_is_completed(self):
|
||||
r = Round(7, 1)
|
||||
self.assertFalse(r.is_completed())
|
||||
r.add_player_guess(1, 0)
|
||||
r.add_player_guess(2, 0)
|
||||
r.add_player_guess(3, 1)
|
||||
r.add_player_guess(0, 1)
|
||||
|
||||
r.play_card(3, r.player_cards[3][0])
|
||||
r.play_card(0, r.player_cards[0][0])
|
||||
r.play_card(1, r.player_cards[1][0])
|
||||
self.assertFalse(r.is_completed())
|
||||
r.play_card(2, r.player_cards[2][0])
|
||||
self.assertTrue(r.is_completed())
|
||||
|
||||
def test_is_guessing_completed(self):
|
||||
r = Round(6, 1)
|
||||
self.assertFalse(r.is_guessing_completed())
|
||||
r.add_player_guess(1, 1)
|
||||
r.add_player_guess(2, 0)
|
||||
r.add_player_guess(3, 2)
|
||||
r.add_player_guess(0, 1)
|
||||
self.assertTrue(r.is_guessing_completed())
|
||||
|
||||
def test_get_last_stash(self):
|
||||
r = Round(7, 0)
|
||||
self.assertIsNone(r.get_last_stash())
|
||||
r.add_player_guess(0, 1)
|
||||
r.add_player_guess(1, 1)
|
||||
r.add_player_guess(2, 0)
|
||||
r.add_player_guess(3, 0)
|
||||
self.assertEqual(r.get_last_stash(), r.stashes[0])
|
||||
r.play_card(0, r.player_cards[0][0])
|
||||
r.play_card(1, r.player_cards[1][0])
|
||||
r.play_card(2, r.player_cards[2][0])
|
||||
r.play_card(3, r.player_cards[3][0])
|
||||
self.assertEqual(r.get_last_stash(), r.stashes[0])
|
||||
r.stashes.append(Stash(2))
|
||||
self.assertEqual(r.get_last_stash(), r.stashes[1])
|
||||
|
||||
def test_get_stashes_winner_summary(self):
|
||||
shuffler = lambda list: None
|
||||
c0 = [
|
||||
Card(Card_colors['BELLS'], Card_values['UPPER']),
|
||||
Card(Card_colors['HEARTS'], Card_values['UPPER'])
|
||||
]
|
||||
c1 = [
|
||||
Card(Card_colors['BELLS'], Card_values['C7']),
|
||||
Card(Card_colors['HEARTS'], Card_values['C10'])
|
||||
]
|
||||
c2 = [
|
||||
Card(Card_colors['BELLS'], Card_values['ACE']),
|
||||
Card(Card_colors['BELLS'], Card_values['C8'])
|
||||
]
|
||||
c3 = [
|
||||
Card(Card_colors['LEAVES'], Card_values['C7']),
|
||||
Card(Card_colors['BELLS'], Card_values['LOWER'])
|
||||
]
|
||||
c = ['dummy']*24 + c0 + c1 + c2 + c3
|
||||
r = Round(6, 1, c, shuffler)
|
||||
self.assertEqual(r.get_stashes_winner_summary(), [0]*4)
|
||||
r.add_player_guess(1, 0)
|
||||
r.add_player_guess(2, 0)
|
||||
r.add_player_guess(3, 1)
|
||||
r.add_player_guess(0, 2)
|
||||
|
||||
r.play_card(0, c0[0])
|
||||
r.play_card(1, c1[0])
|
||||
r.play_card(2, c2[0])
|
||||
self.assertEqual(r.get_stashes_winner_summary(), [0]*4)
|
||||
r.play_card(3, c3[1])
|
||||
self.assertEqual(r.get_stashes_winner_summary(), [0, 0, 1, 0])
|
||||
|
||||
r.play_card(2, c2[1])
|
||||
r.play_card(3, c3[0])
|
||||
r.play_card(0, c0[1])
|
||||
r.play_card(1, c1[1])
|
||||
self.assertEqual(r.get_stashes_winner_summary(), [1, 0, 1, 0])
|
||||
|
||||
def test_get_points_summary(self):
|
||||
shuffler = lambda list: None
|
||||
c0 = [
|
||||
Card(Card_colors['BELLS'], Card_values['UPPER']),
|
||||
Card(Card_colors['HEARTS'], Card_values['UPPER'])
|
||||
]
|
||||
c1 = [
|
||||
Card(Card_colors['BELLS'], Card_values['C7']),
|
||||
Card(Card_colors['HEARTS'], Card_values['C10'])
|
||||
]
|
||||
c2 = [
|
||||
Card(Card_colors['BELLS'], Card_values['ACE']),
|
||||
Card(Card_colors['BELLS'], Card_values['C8'])
|
||||
]
|
||||
c3 = [
|
||||
Card(Card_colors['LEAVES'], Card_values['C7']),
|
||||
Card(Card_colors['BELLS'], Card_values['LOWER'])
|
||||
]
|
||||
c = ['dummy']*24 + c0 + c1 + c2 + c3
|
||||
r = Round(6, 1, c, shuffler)
|
||||
r.add_player_guess(1, 0)
|
||||
r.add_player_guess(2, 1)
|
||||
r.add_player_guess(3, 0)
|
||||
r.add_player_guess(0, 2)
|
||||
|
||||
r.play_card(0, c0[0])
|
||||
r.play_card(1, c1[0])
|
||||
r.play_card(2, c2[0])
|
||||
r.play_card(3, c3[1])
|
||||
|
||||
r.play_card(2, c2[1])
|
||||
r.play_card(3, c3[0])
|
||||
r.play_card(0, c0[1])
|
||||
r.play_card(1, c1[1])
|
||||
self.assertEqual(r.get_points_summary(), [0, 10, 11, 10])
|
||||
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main(verbosity=2)
|
||||
@@ -0,0 +1,206 @@
|
||||
"""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_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 tearDownModule():
|
||||
from db.db import engine
|
||||
|
||||
run(engine.dispose())
|
||||
try:
|
||||
os.remove(_DB_FILE)
|
||||
except OSError:
|
||||
pass
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,234 @@
|
||||
import unittest
|
||||
|
||||
import api
|
||||
|
||||
|
||||
class SocketLayerCase(unittest.IsolatedAsyncioTestCase):
|
||||
"""Drives the socket handlers directly with a fake emit, asserting the
|
||||
identity-binding, privacy, lifecycle and error-handling guarantees."""
|
||||
|
||||
def setUp(self):
|
||||
api.games.clear()
|
||||
api.sessions.clear()
|
||||
self.emits = []
|
||||
|
||||
async def fake_emit(event, data=None, room=None, to=None, **kw):
|
||||
self.emits.append({"event": event, "data": data, "room": room, "to": to})
|
||||
|
||||
async def noop(*a, **kw):
|
||||
pass
|
||||
|
||||
# Patch the network-facing primitives; logic under test is untouched.
|
||||
api.sio.emit = fake_emit
|
||||
api.sio.enter_room = noop
|
||||
api.sio.leave_room = noop
|
||||
|
||||
# --- helpers ----------------------------------------------------------
|
||||
|
||||
def last(self, event):
|
||||
return next(e for e in reversed(self.emits) if e["event"] == event)
|
||||
|
||||
def events(self, event):
|
||||
return [e for e in self.emits if e["event"] == event]
|
||||
|
||||
async def make_game(self):
|
||||
await api.create_game("host", "Game")
|
||||
return list(api.games.keys())[0]
|
||||
|
||||
async def make_started_game(self):
|
||||
gid = await self.make_game()
|
||||
for i in range(4):
|
||||
await api.register_player(f"s{i}", gid, f"P{i}")
|
||||
await api.start_game("s0", gid)
|
||||
return gid
|
||||
|
||||
# --- creation & lobby -------------------------------------------------
|
||||
|
||||
async def test_create_game_mints_uuid(self):
|
||||
gid = await self.make_game()
|
||||
self.assertNotEqual(gid, "a")
|
||||
self.assertEqual(len(gid), 36)
|
||||
self.assertEqual(self.last("create_game")["data"]["gid"], gid)
|
||||
|
||||
async def test_lobby_payload_hides_sids_and_tokens(self):
|
||||
gid = await self.make_game()
|
||||
await api.register_player("s0", gid, "P0")
|
||||
lobby = self.last("get_games")["data"]["games"]
|
||||
blob = repr(lobby)
|
||||
self.assertNotIn("sid", blob)
|
||||
self.assertNotIn("token", blob)
|
||||
self.assertEqual(lobby[0]["players"][0], {"order": 0, "name": "P0", "connected": True})
|
||||
|
||||
async def test_register_returns_private_token_to_caller_only(self):
|
||||
gid = await self.make_game()
|
||||
await api.register_player("s0", gid, "P0")
|
||||
reg = self.last("register_player")
|
||||
self.assertEqual(reg["to"], "s0")
|
||||
self.assertIn("token", reg["data"])
|
||||
self.assertEqual(reg["data"]["player"], {"order": 0, "name": "P0"})
|
||||
|
||||
async def test_register_rejects_unknown_game_fifth_and_started(self):
|
||||
await api.register_player("x", "nope", "P") # unknown gid
|
||||
self.assertEqual(self.last("error")["data"]["error"], "Hra neexistuje.")
|
||||
|
||||
gid = await self.make_game()
|
||||
for i in range(4):
|
||||
await api.register_player(f"s{i}", gid, f"P{i}")
|
||||
await api.register_player("s4", gid, "P4") # fifth
|
||||
self.assertEqual(self.last("error")["data"]["error"], "Prekroceny pocet hracov.")
|
||||
|
||||
await api.start_game("s0", gid)
|
||||
await api.register_player("s5", gid, "late") # already started
|
||||
self.assertEqual(self.last("error")["data"]["error"], "Hra uz zacala.")
|
||||
|
||||
# --- start lifecycle --------------------------------------------------
|
||||
|
||||
async def test_start_deals_and_emits_private_hands(self):
|
||||
gid = await self.make_started_game()
|
||||
self.assertTrue(api.games[gid].started)
|
||||
self.assertEqual(self.last("game_status")["room"], gid)
|
||||
hands = self.events("player_cards")
|
||||
self.assertEqual(sorted(h["to"] for h in hands), ["s0", "s1", "s2", "s3"])
|
||||
|
||||
async def test_start_game_twice_is_rejected_and_does_not_reset(self):
|
||||
gid = await self.make_started_game()
|
||||
core_before = api.games[gid].bridzik_core
|
||||
await api.start_game("s0", gid)
|
||||
self.assertEqual(self.last("error")["data"]["error"], "Hra uz zacala.")
|
||||
self.assertIs(api.games[gid].bridzik_core, core_before)
|
||||
|
||||
# --- identity binding (the anti-cheat guarantee) ----------------------
|
||||
|
||||
async def test_player_cards_serves_only_your_own_seat(self):
|
||||
gid = await self.make_started_game()
|
||||
# The handler takes no player argument: a connection can only ever
|
||||
# request the hand bound to its own session.
|
||||
self.emits.clear()
|
||||
await api.player_cards("s2")
|
||||
sent = self.last("player_cards")
|
||||
self.assertEqual(sent["to"], "s2")
|
||||
expected = api.games[gid].bridzik_core.get_player_cards(2)
|
||||
self.assertEqual(len(sent["data"]["cards"]), len(expected))
|
||||
|
||||
async def test_action_without_session_is_rejected(self):
|
||||
await self.make_started_game()
|
||||
await api.add_guess("stranger", 1)
|
||||
self.assertEqual(self.last("error")["data"]["error"], "Nie ste v rozohratej hre.")
|
||||
|
||||
# --- full play flow ---------------------------------------------------
|
||||
|
||||
async def test_guess_and_play_flow(self):
|
||||
gid = await self.make_started_game()
|
||||
core = api.games[gid].bridzik_core
|
||||
# Round 0 leads with seat 0; guess 0 all round (sum 0 != 8 tricks).
|
||||
for i in range(4):
|
||||
await api.add_guess(f"s{i}", 0)
|
||||
rnd = core.series[-1].get_last_round()
|
||||
self.assertTrue(rnd.is_guessing_completed())
|
||||
leader = rnd.get_active_player()
|
||||
|
||||
before = len(core.get_player_cards(leader))
|
||||
self.emits.clear()
|
||||
await api.play_card(f"s{leader}", 0)
|
||||
after = len(core.get_player_cards(leader))
|
||||
self.assertEqual(after, before - 1)
|
||||
status = self.last("game_status")["data"]
|
||||
self.assertIn("completed", status)
|
||||
self.assertIn("active_stash", status["status"])
|
||||
|
||||
async def test_bad_input_does_not_crash(self):
|
||||
await self.make_started_game()
|
||||
await api.add_guess("s0", "not-a-number")
|
||||
self.assertEqual(self.last("error")["data"]["error"], "Neplatny tip.")
|
||||
await api.play_card("s0", "xx")
|
||||
self.assertEqual(self.last("error")["data"]["error"], "Neplatna karta.")
|
||||
|
||||
async def test_engine_error_message_is_forwarded(self):
|
||||
gid = await self.make_started_game()
|
||||
# Seat 0 leads guessing; seat 1 acting out of turn -> engine rejects,
|
||||
# and the real (Slovak) reason is forwarded, not a generic message.
|
||||
await api.add_guess("s1", 1)
|
||||
self.assertEqual(self.last("error")["data"]["error"], "Hrac nie je na tahu")
|
||||
|
||||
# --- reconnect --------------------------------------------------------
|
||||
|
||||
async def test_secure_reconnect_with_token_rebinds_seat(self):
|
||||
gid = await self.make_game()
|
||||
await api.register_player("s0", gid, "P0")
|
||||
token = self.last("register_player")["data"]["token"]
|
||||
for i in range(1, 4):
|
||||
await api.register_player(f"s{i}", gid, f"P{i}")
|
||||
await api.start_game("s0", gid)
|
||||
|
||||
await api.reconnect_to_game("s0-new", gid, token)
|
||||
self.assertEqual(api.games[gid].player_by_token(token).sid, "s0-new")
|
||||
self.assertEqual(api.sessions["s0-new"], {"gid": gid, "order": 0})
|
||||
|
||||
self.emits.clear()
|
||||
await api.reconnect_to_game("zzz", gid, "wrong-token")
|
||||
self.assertEqual(self.last("error")["data"]["error"], "Neplatny token pre pripojenie.")
|
||||
|
||||
# --- self-contained game_status --------------------------------------
|
||||
|
||||
async def test_game_status_carries_roster_and_round_meta(self):
|
||||
gid = await self.make_started_game()
|
||||
self.emits.clear()
|
||||
await api.game_status("s0")
|
||||
data = self.last("game_status")["data"]
|
||||
self.assertEqual(data["round_number"], 0)
|
||||
self.assertEqual(data["cards_in_round"], 8)
|
||||
self.assertEqual(data["series_number"], 0)
|
||||
roster = data["players"]
|
||||
self.assertEqual([p["order"] for p in roster], [0, 1, 2, 3])
|
||||
self.assertEqual([p["name"] for p in roster], ["P0", "P1", "P2", "P3"])
|
||||
self.assertTrue(all(p["connected"] for p in roster))
|
||||
|
||||
# --- leave_game -------------------------------------------------------
|
||||
|
||||
async def test_leave_before_start_frees_the_seat(self):
|
||||
gid = await self.make_game()
|
||||
await api.register_player("s0", gid, "P0")
|
||||
await api.register_player("s1", gid, "P1")
|
||||
await api.leave_game("s0") # order 0 leaves
|
||||
self.assertNotIn("s0", api.sessions)
|
||||
self.assertEqual([p.order for p in api.games[gid].players], [1])
|
||||
# the freed seat 0 is reused by the next joiner
|
||||
await api.register_player("s2", gid, "P2")
|
||||
self.assertEqual(api.sessions["s2"]["order"], 0)
|
||||
|
||||
async def test_leave_last_lobby_player_drops_game(self):
|
||||
gid = await self.make_game()
|
||||
await api.register_player("s0", gid, "P0")
|
||||
await api.leave_game("s0")
|
||||
self.assertNotIn(gid, api.games)
|
||||
|
||||
async def test_leave_started_game_keeps_seat_marks_offline(self):
|
||||
gid = await self.make_started_game()
|
||||
await api.leave_game("s1")
|
||||
self.assertNotIn("s1", api.sessions)
|
||||
player = api.games[gid].player_by_order(1)
|
||||
self.assertIsNotNone(player) # seat kept for reconnect
|
||||
self.assertFalse(player.connected)
|
||||
self.assertIn(gid, api.games) # others still in
|
||||
|
||||
async def test_leave_without_session_is_a_noop(self):
|
||||
await api.leave_game("ghost")
|
||||
# no exception, and no error emitted to a non-participant
|
||||
self.assertEqual(self.events("error"), [])
|
||||
|
||||
# --- disconnect & cleanup --------------------------------------------
|
||||
|
||||
async def test_disconnect_marks_player_then_drops_empty_game(self):
|
||||
gid = await self.make_started_game()
|
||||
await api.disconnect("s0")
|
||||
self.assertFalse(api.games[gid].player_by_sid("s0").connected)
|
||||
self.assertNotIn("s0", api.sessions)
|
||||
self.assertIn(gid, api.games) # others still connected
|
||||
|
||||
for i in range(1, 4):
|
||||
await api.disconnect(f"s{i}")
|
||||
self.assertNotIn(gid, api.games) # everyone gone -> game removed
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user