77 lines
3.3 KiB
Python
77 lines
3.3 KiB
Python
"""Notebooks persist independently of public gameplay and rollback history."""
|
||
import pytest
|
||
from fastapi.testclient import TestClient
|
||
from sqlmodel import Session, SQLModel, create_engine, select
|
||
from sqlmodel.pool import StaticPool
|
||
|
||
from pirats.database import get_session
|
||
from pirats.main import app
|
||
from pirats.models import Game, Player, PlayerNotebook, Checkpoint
|
||
from pirats.crud_rollback import serialize_game_state, apply_game_state
|
||
|
||
|
||
@pytest.fixture
|
||
def notebook_env(monkeypatch):
|
||
engine = create_engine('sqlite://', connect_args={'check_same_thread': False}, poolclass=StaticPool)
|
||
SQLModel.metadata.create_all(engine)
|
||
with Session(engine) as db:
|
||
game = Game(phase='scene')
|
||
other = Game()
|
||
db.add(game)
|
||
db.add(other)
|
||
db.commit()
|
||
players = [Player(game_id=game.id, player_name=name) for name in ('One', 'Two')]
|
||
db.add_all(players)
|
||
db.commit()
|
||
ids = game.id, other.id, players[0].id, players[1].id
|
||
def session_override():
|
||
with Session(engine) as db:
|
||
yield db
|
||
app.dependency_overrides[get_session] = session_override
|
||
def unexpected(*args, **kwargs):
|
||
pytest.fail('Notes must not checkpoint gameplay or broadcast to the crew')
|
||
monkeypatch.setattr('pirats.main._checkpoint_after_mutation', unexpected)
|
||
monkeypatch.setattr('pirats.main.manager.broadcast', unexpected)
|
||
try:
|
||
yield TestClient(app), engine, ids
|
||
finally:
|
||
app.dependency_overrides.clear()
|
||
engine.dispose()
|
||
|
||
|
||
def test_notes_round_trip_isolation_and_validation(notebook_env):
|
||
client, engine, (gid, other, pid, second) = notebook_env
|
||
path = f'/api/game/{gid}/player/{pid}/notes'
|
||
assert client.get(path).json() == {'text': ''}
|
||
text = 'Remember the captain’s map 🐀\n Meet at dusk.\n'
|
||
assert client.put(path, data={'text': text}).json() == {'text': text}
|
||
assert client.get(path).json() == {'text': text}
|
||
assert client.get(f'/api/game/{gid}/player/{second}/notes').json() == {'text': ''}
|
||
assert text not in client.get(f'/api/game/{gid}/player/{second}/state').text
|
||
wrong = f'/api/game/{other}/player/{pid}/notes'
|
||
assert client.get(wrong).status_code == 404
|
||
assert client.put(wrong, data={'text': 'bad'}).status_code == 404
|
||
assert client.put(path, data={'text': 'x' * 50001}).status_code == 422
|
||
assert client.get(path).json()['text'] == text
|
||
assert client.put(path, data={'text': ''}).json() == {'text': ''}
|
||
assert client.get(path).json() == {'text': ''}
|
||
|
||
|
||
def test_notes_survive_rollback_and_game_cleanup(notebook_env):
|
||
client, engine, (gid, _, pid, _) = notebook_env
|
||
path = f'/api/game/{gid}/player/{pid}/notes'
|
||
with Session(engine) as db:
|
||
game = db.get(Game, gid)
|
||
snapshot = serialize_game_state(game)
|
||
client.put(path, data={'text': 'Keep this after rewinding'})
|
||
with Session(engine) as db:
|
||
game = db.get(Game, gid)
|
||
assert 'Keep this after rewinding' not in serialize_game_state(game)
|
||
apply_game_state(db, game, snapshot)
|
||
assert db.exec(select(Checkpoint)).all() == []
|
||
assert client.get(path).json()['text'] == 'Keep this after rewinding'
|
||
with Session(engine) as db:
|
||
db.delete(db.get(Game, gid))
|
||
db.commit()
|
||
assert db.exec(select(PlayerNotebook)).all() == []
|