87 lines
3.5 KiB
Python
87 lines
3.5 KiB
Python
"""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())
|