Migrate Socket.IO server to ASGI, fix bugs, harden, add socket tests

Replace Flask-SocketIO + eventlet with python-socketio AsyncServer on an
ASGI app served by uvicorn (Python 3.14). The server is no longer started
as an import side-effect; `python -m app` runs uvicorn for dev and the
Docker image runs `uvicorn api:app`.

Bug fixes:
- create_game now mints a real uuid gid and returns it to the creator
  (was hardcoded 'a').
- play_card resolves the player's hand and plays the selected Card (was
  indexing a method and crashing).

Hardening:
- Identity binding: every action derives the seat from the connection
  (sid -> {gid, order}); clients no longer pass a player number, closing
  the hidden-cards cheat where any client could request any hand.
- Secure token-based reconnect (per-player secret token).
- disconnect handler marks players offline and drops empty games (no
  more leaked games), notifying the room via player_connection.
- Guards for unknown gid, double start_game, and bad input; engine
  exception messages are forwarded instead of swallowed.
- Lobby payload is public-only (no sids/tokens); game_status carries a
  completed flag.
- /health endpoint via other_asgi_app; env-driven CORS and logging.

Infra:
- Dockerfile -> python:3.14-slim, uvicorn CMD, drop dead venv lines.
- requirements.txt -> python-socketio/engineio + uvicorn; drop eventlet,
  Flask-SocketIO, Flask-Session.
- docker-compose: drop unused debugpy port and obsolete version key.
- Remove redundant start.py; gitignore /.venv.

Tests: test_socket.py drives the handlers (identity binding, lobby
privacy, reconnect, disconnect cleanup, error handling, play flow).
Full suite: 29 passing.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Tim
2026-06-13 23:40:32 +02:00
parent aa1b037c1a
commit d47eb03bce
8 changed files with 529 additions and 100 deletions
+186
View File
@@ -0,0 +1,186 @@
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.")
# --- 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()