Show live player presence and Captain/Gat roster badges

tests/test_presence.py

@@ -0,0 +1,86 @@
+"""Live presence tracks connections rather than persistent character state."""
+import asyncio
+from contextlib import asynccontextmanager
+
+from fastapi.testclient import TestClient
+from sqlmodel import Session, SQLModel, create_engine
+from sqlmodel.pool import StaticPool
+
+from pirats.database import get_session
+from pirats.main import app
+from pirats.ws import GameConnectionManager
+
+
+def test_presence_tracks_multiple_tabs_and_rejects_other_games():
+    engine = create_engine('sqlite://', connect_args={'check_same_thread': False}, poolclass=StaticPool)
+    SQLModel.metadata.create_all(engine)
+
+    def session_override():
+        with Session(engine) as db:
+            yield db
+
+    app.dependency_overrides[get_session] = session_override
+    original_lifespan = app.router.lifespan_context
+    @asynccontextmanager
+    async def test_lifespan(app):
+        yield
+    app.router.lifespan_context = test_lifespan
+    try:
+        with TestClient(app) as client:
+            gid = client.post('/api/game', data={'crew_name': 'Presence'}).json()['id']
+            pid = client.post(f'/api/game/{gid}/join', data={'name': 'Rat'}).json()['id']
+            other = client.post('/api/game', data={'crew_name': 'Other'}).json()['id']
+            from starlette.websockets import WebSocketDisconnect
+            import pytest
+            with pytest.raises(WebSocketDisconnect) as rejected:
+                with client.websocket_connect(f'/api/game/{other}/ws?player_id={pid}'):
+                    pass
+            assert rejected.value.code == 1008
+
+            with client.websocket_connect(f'/api/game/{gid}/ws') as observer:
+                with client.websocket_connect(f'/api/game/{gid}/ws?player_id={pid}') as first:
+                    expected = {'type': 'presence', 'player_ids': [pid]}
+                    assert first.receive_json() == expected
+                    assert observer.receive_json() == expected
+                    with client.websocket_connect(f'/api/game/{gid}/ws?player_id={pid}') as second:
+                        assert second.receive_json() == expected
+                        assert first.receive_json() == expected
+                        assert observer.receive_json() == expected
+                        second.close()
+                        assert first.receive_json() == expected
+                        assert observer.receive_json() == expected
+                    first.close()
+                    assert observer.receive_json() == {'type': 'presence', 'player_ids': []}
+    finally:
+        app.router.lifespan_context = original_lifespan
+        app.dependency_overrides.clear()
+        engine.dispose()
+
+
+def test_failed_socket_is_removed_from_presence():
+    class Socket:
+        def __init__(self):
+            self.messages = []
+            self.failed = False
+
+        async def accept(self):
+            pass
+
+        async def send_json(self, message):
+            if self.failed:
+                raise RuntimeError('Connection lost')
+            self.messages.append(message)
+
+    async def scenario():
+        manager = GameConnectionManager()
+        first, second = Socket(), Socket()
+        await manager.connect('game', first, 'a')
+        await manager.connect('game', second, 'b')
+        first.failed = True
+        await manager.broadcast('game', {'type': 'state_changed'})
+        assert second.messages[-1] == {'type': 'presence', 'player_ids': ['b']}
+        manager.disconnect('game', first)  # Endpoint cleanup remains idempotent.
+        manager.disconnect('game', second)
+        assert not manager._connections
+
+    asyncio.run(scenario())