tim
b8e2d15e27
- leave_game: explicit exit from the lobby/game. Before start it frees the seat; after start it keeps the seat (token reconnect still works) and marks the player offline, dropping the game once empty. - Seat assignment now picks the lowest free order, so leaving a lobby before start no longer collides order numbers on the next join. - game_status is now self-contained: adds players roster (order/name/connected), series_number, round_number and cards_in_round (= trick count / max bid), so the game view no longer has to stitch the lobby snapshot. - Add Game.player_by_order helper. Tests: +5 (roster/round meta, leave-before-start frees seat, leave drops empty game, leave started game keeps seat offline, leave noop). Suite: 34. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
235 lines
9.7 KiB
Python
235 lines
9.7 KiB
Python
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()
|