tim.mccarthy/pirats
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
2de4648288a29a4f9ea2aa75185772299f497da8
pirats/tests/test_game.py
Tim McCarthy 692c6d26c1 Cap request body size (HTTP 413) 
LimitRequestBodyMiddleware (pure ASGI, registered outermost) rejects request
bodies larger than PIRATS_MAX_BODY_BYTES (default 1 MiB) before they're buffered
into memory: it checks the declared Content-Length first, then counts the bytes
actually streamed so a chunked/length-omitting client can't bypass the header
check. Exposed as services.pirats.maxBodyBytes and documented in the README.

Tested in isolation (Content-Length fast path + streamed path) and through the
real app.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-15 15:59:48 -07:00

2572 lines
94 KiB
Python
Raw Blame History

This file contains invisible Unicode characters
This file contains invisible Unicode characters that are indistinguishable to humans but may be processed differently by a computer. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
import pytest
import json
from sqlmodel import SQLModel, create_engine, Session, select
from pirats import cards
from pirats import crud
from pirats.models import Game, Obstacle, Player, Challenge, GameEvent, Checkpoint
# In-memory database for testing
@pytest.fixture(name="session")
def session_fixture():
engine = create_engine("sqlite://", connect_args={"check_same_thread": False})
SQLModel.metadata.create_all(engine)
with Session(engine) as session:
yield session
def test_card_parsing():
c_10d = cards.parse_card("10D")
assert c_10d["value"] == "10"
assert c_10d["suit"] == "D"
assert c_10d["color"] == "red"
assert c_10d["numeric_value"] == 10
assert not c_10d["is_joker"]
c_ac = cards.parse_card("AC")
assert c_ac["value"] == "A"
assert c_ac["suit"] == "C"
assert c_ac["color"] == "black"
assert c_ac["numeric_value"] == 1
c_kh = cards.parse_card("KH")
assert c_kh["value"] == "K"
assert c_kh["numeric_value"] == 13
c_joker = cards.parse_card("Joker1")
assert c_joker["is_joker"]
assert c_joker["color"] == "black"
def test_obstacle_info():
obs = cards.get_obstacle_info("7C")
assert obs["title"] == "Lustful Bean Whale"
assert "bean" in obs["description"].lower()
assert obs["initial_value"] == 7
assert obs["color"] == "black"
def test_game_creation(session):
game = crud.create_game(session)
assert game.id is not None
assert game.phase == "lobby"
assert len(crud.get_game_deck(game)) == 54
def test_character_creation_and_swap(session):
game = crud.create_game(session)
p1 = crud.add_player(session, game.id, "Carlos", is_creator=True)
p2 = crud.add_player(session, game.id, "Barry")
p3 = crud.add_player(session, game.id, "Jaspar")
session.refresh(game)
assert len(game.players) == 3
# Fill basic details
p1.avatar_look = "Bandana"
p2.avatar_look = "Eyepatch"
p3.avatar_look = "Pegleg"
session.add_all([p1, p2, p3])
session.commit()
# Submit techniques
crud.submit_techniques(session, p1.id, "Carlos Strike", "Hide", "Roll")
crud.submit_techniques(session, p2.id, "Barry Leap", "Bite", "Scream")
crud.submit_techniques(session, p3.id, "Jaspar Blast", "Shoot", "Run")
session.refresh(game)
# The swap phase should have automatically begun because all 3 submitted
assert game.phase == "swap_techniques"
session.refresh(p1)
session.refresh(p2)
session.refresh(p3)
# Everyone starts the swap phase holding their own creations
held_p1 = json.loads(p1.held_techniques)
assert [t["text"] for t in held_p1] == ["Carlos Strike", "Hide", "Roll"]
assert all(t["creator_id"] == p1.id for t in held_p1)
assert json.loads(p1.swapped_techniques) == []
# Can't ready up while still holding your own techniques
ok, msg = crud.set_swap_ready(session, p1.id, True)
assert not ok
# A scripted set of blind swaps that leaves nobody holding their own work:
# (offerer, offered technique, responder, technique traded back)
swaps = [
(p1, "Carlos Strike", p2, "Barry Leap"),
(p1, "Hide", p3, "Jaspar Blast"),
(p1, "Roll", p2, "Bite"),
(p2, "Scream", p3, "Shoot"),
(p3, "Run", p2, "Carlos Strike"),
]
for offerer, offered, responder, returned in swaps:
ok, msg = crud.offer_swap(session, offerer.id, responder.id, offered)
assert ok, msg
ok, msg = crud.respond_swap_offer(session, responder.id, offerer.id, True, returned)
assert ok, msg
for p in (p1, p2, p3):
session.refresh(p)
held = json.loads(p.held_techniques)
assert len(held) == 3
assert all(t["creator_id"] != p.id for t in held)
ok, msg = crud.set_swap_ready(session, p.id, True)
assert ok, msg
session.refresh(game)
assert game.phase == "assign_techniques"
# Everyone's swapped techniques were locked in from their held set
session.refresh(p1)
techs_p1 = json.loads(p1.swapped_techniques)
assert len(techs_p1) == 3
assert "Carlos Strike" not in techs_p1
# Now assign to face cards (must use exactly the swapped techniques)
ok, msg = crud.assign_face_card_techniques(session, p1.id, "T2", "T3", "T4")
assert not ok
for p in (p1, p2, p3):
session.refresh(p)
techs = json.loads(p.swapped_techniques)
ok, msg = crud.assign_face_card_techniques(session, p.id, techs[0], techs[1], techs[2])
assert ok, msg
session.refresh(game)
# All are ready, should transition to scene_setup and assign starting ranks!
assert game.phase == "scene_setup"
session.refresh(p1)
session.refresh(p2)
session.refresh(p3)
ranks = [p1.rank, p2.rank, p3.rank]
assert 3 in ranks
assert 1 in ranks
assert 2 in ranks
roles = [p1.role, p2.role, p3.role]
assert "deep" in roles
assert "pirat" in roles
def make_scene_game(session, num_pirats=1):
"""Helper: a game in scene phase with 1 Deep player and num_pirats Pi-Rats."""
game = crud.create_game(session)
deep = crud.add_player(session, game.id, "Deep1", is_creator=True)
deep.rank = 3
deep.role = "deep"
deep.previous_role = "deep"
session.add(deep)
pirats = []
for i in range(num_pirats):
p = crud.add_player(session, game.id, f"Rat{i+1}")
p.rank = 1 if i == 0 else 2
p.role = "pirat"
p.previous_role = "pirat"
session.add(p)
pirats.append(p)
session.commit()
# Stub shuffle so the rebuilt deck keeps its construction order (Jokers last)
# and no Joker can be drawn as an Obstacle, which would bump extra_obstacles
import random
orig_shuffle = random.shuffle
random.shuffle = lambda x: None
try:
success, msg = crud.confirm_scene_setup(session, game.id)
finally:
random.shuffle = orig_shuffle
assert success, msg
session.refresh(game)
return game, deep, pirats
def test_scene_start_and_challenges(session):
game, deep, (p2,) = make_scene_game(session)
assert game.phase == "scene"
assert len(game.obstacles) == 2 # Deeps (1) + 1 = 2 obstacles
obs = game.obstacles[0]
# Give p2 a known hand for testing
p2.hand_cards = json.dumps(["10D", "2H", "JS", "Joker1"])
obs.current_value = 5
session.add_all([p2, obs])
session.commit()
# Playing a card without an open Challenge must be rejected
ok, msg, res = crud.play_challenge_card(session, p2.id, obs.id, "10D")
assert not ok
assert "Challenge" in msg
# The Deep calls a Challenge applying the obstacle
ok, msg = crud.create_challenge(session, game.id, deep.id, p2.id, [obs.id], stakes_success="The cheese is yours", stakes_failure="The cheese is on the line")
assert ok, msg
session.refresh(game)
challenge = game.challenges[0]
assert challenge.status == "open"
assert challenge.acting_player_id == p2.id
assert challenge.stakes_success == "The cheese is yours"
assert challenge.stakes_failure == "The cheese is on the line"
orig_is_red = cards.parse_card(obs.original_card)["color"] == "red"
ok, msg, res = crud.play_challenge_card(session, p2.id, obs.id, "10D")
assert ok
assert res["success"]
assert "Success" in res["details"]
session.refresh(p2)
session.refresh(obs)
# Obstacle current value should update to 10
assert obs.current_value == 10
# Check if hand size matches card draw rule
# 10D was played (-1 card). If colors matched, drew 1 back (+1 card).
hand = crud.get_player_hand(p2)
if orig_is_red:
assert len(hand) == 4
assert res["drew_card"] is not None
else:
assert len(hand) == 3
assert res["drew_card"] is None
# The Deep resolves the challenge: at least one success -> succeeded
ok, msg = crud.resolve_challenge(session, challenge.id, deep.id)
assert ok
session.refresh(challenge)
assert challenge.status == "succeeded"
def test_assistant_does_not_draw(session):
game, deep, (p2, p3) = make_scene_game(session, num_pirats=2)
obs = game.obstacles[0]
obs.current_value = 5
# Helper card guaranteed to match the obstacle's color
helper_card = "9D" if cards.parse_card(obs.original_card)["color"] == "red" else "9S"
p3.hand_cards = json.dumps([helper_card])
session.add_all([obs, p3])
session.commit()
ok, msg = crud.create_challenge(session, game.id, deep.id, p2.id, [obs.id])
assert ok
session.refresh(game)
# p3 assists: plays a color-matching card but must NOT draw back
ok, msg, res = crud.play_challenge_card(session, p3.id, obs.id, helper_card)
assert ok
assert res["success"]
assert res["drew_card"] is None
session.refresh(p3)
assert crud.get_player_hand(p3) == []
def test_secret_technique_auto_success(session):
game, deep, (p1,) = make_scene_game(session)
obs = game.obstacles[0]
p1.tech_jack = "Pocket Sand"
p1.hand_cards = json.dumps(["JS"])
obs.current_value = 10
session.add_all([p1, obs])
session.commit()
ok, msg = crud.create_challenge(session, game.id, deep.id, p1.id, [obs.id])
assert ok
# Play Jack. J is a face card -> auto success!
ok, msg, res = crud.play_challenge_card(session, p1.id, obs.id, "JS")
assert ok
assert res["success"]
assert res["is_technique"]
assert res["tech_name"] == "Pocket Sand"
def test_joker_play(session):
game = crud.create_game(session)
# Remove Joker1 and 10C from deck to prevent flakiness
deck = json.loads(game.deck_cards)
if "Joker1" in deck:
deck.remove("Joker1")
if "10C" in deck:
deck.remove("10C")
game.deck_cards = json.dumps(deck)
session.add(game)
p1 = crud.add_player(session, game.id, "P1")
p1.role = "pirat"
p1.hand_cards = json.dumps(["Joker1"])
session.add(p1)
obs = Obstacle(
game_id=game.id,
original_card="10C",
suit="C",
title="Knights",
current_value=10
)
session.add(obs)
session.commit()
# Play Joker: discards the obstacle and draws a replacement
ok, msg = crud.play_joker(session, p1.id, obs.id, "Joker1")
assert ok
# Verify obstacle was deleted and replaced
session.refresh(game)
assert len(game.obstacles) == 1
assert game.obstacles[0].original_card != "10C"
# Joker left the player's hand
session.refresh(p1)
assert crud.get_player_hand(p1) == []
def test_obstacle_success_count(session):
game = crud.create_game(session)
obs = Obstacle(
game_id=game.id,
original_card="10C",
suit="C",
title="Knights",
current_value=10,
played_cards=json.dumps([
{"card": "9H", "player_id": "p1", "player_name": "P1", "success": True},
{"card": "8S", "player_id": "p2", "player_name": "P2", "success": False},
{"card": "AH", "player_id": "p1", "player_name": "P1", "success": True}
])
)
session.add(obs)
session.commit()
assert obs.success_count == 2
def test_captaincy_and_hand_sizes(session):
game = crud.create_game(session)
p1 = crud.add_player(session, game.id, "Captain Barnaby")
p2 = crud.add_player(session, game.id, "Crewmate Pip")
p3 = crud.add_player(session, game.id, "Lowly Larry")
p1.role = "deep"
p1.rank = 2
p2.role = "pirat"
p2.rank = 2
p3.role = "pirat"
p3.rank = 1
session.add_all([p1, p2, p3])
session.commit()
# No captain until the crew controls a ship
assert game.captain_player_id is None
# Highest pirat (p2) -> 4 cards, lowest (p3) -> 2 cards
assert crud.calculate_max_hand_size(p2, game.players, game.captain_player_id) == 4
assert crud.calculate_max_hand_size(p3, game.players, game.captain_player_id) == 2
# If everyone shares a Rank, they are all 'highest' -> 4 cards
p3.rank = 2
session.add(p3)
session.commit()
assert crud.calculate_max_hand_size(p3, game.players, game.captain_player_id) == 4
# Stealing a ship does NOT auto-assign a Captain — the crew must earn and
# choose one. Completing Crew Objective 1 leaves the captaincy vacant.
crud.toggle_objective(session, game.id, p1.id, "crew_1", True)
session.refresh(game)
assert game.captain_player_id is None
# The crew chooses a Captain explicitly (via the Captaincy UI / set_captain).
crud.set_captain(session, game, p2.id)
session.refresh(game)
assert game.captain_player_id == p2.id
# Captain privilege: +1 hand size
assert crud.calculate_max_hand_size(p2, game.players, game.captain_player_id) == 5
def test_captain_tax_on_failed_challenge(session):
game, deep, (p2, p3) = make_scene_game(session, num_pirats=2)
# p3 (rank 2) becomes captain
crud.set_captain(session, game, p3.id)
session.refresh(game)
assert game.captain_player_id == p3.id
start_rank = p3.rank
obs = game.obstacles[0]
obs.current_value = 13
p3.hand_cards = json.dumps(["2C"])
session.add_all([obs, p3])
session.commit()
ok, msg = crud.create_challenge(session, game.id, deep.id, p3.id, [obs.id])
assert ok
session.refresh(game)
challenge = [c for c in game.challenges if c.status == "open"][0]
ok, msg, res = crud.play_challenge_card(session, p3.id, obs.id, "2C")
assert ok
assert not res["success"]
ok, msg = crud.resolve_challenge(session, challenge.id, deep.id)
assert ok
session.refresh(p3)
# Captain Tax: lost a rank for personally failing a Challenge
assert p3.rank == start_rank - 1
def test_set_role_endpoint():
from fastapi.testclient import TestClient
from sqlalchemy.pool import StaticPool
from sqlmodel import SQLModel, create_engine, Session
from pirats.main import app
from pirats import crud
# Setup in-memory DB with StaticPool to share connection across threads
engine = create_engine(
"sqlite://",
connect_args={"check_same_thread": False},
poolclass=StaticPool,
)
SQLModel.metadata.create_all(engine)
session = Session(engine)
# Create game and player
game = crud.create_game(session)
player = crud.add_player(session, game.id, "TestPirate")
game.phase = "scene_setup"
session.add(game)
session.commit()
def get_session_override():
yield session
from pirats.database import get_session
app.dependency_overrides[get_session] = get_session_override
client = TestClient(app)
try:
response = client.post(f"/api/game/{game.id}/player/{player.id}/set-role", data={"role": "pirat"})
assert response.status_code == 200
assert response.json()["status"] == "ok"
# The database should be updated
session.refresh(player)
assert player.role == "pirat"
finally:
app.dependency_overrides.clear()
def test_game_creation_with_crew_name(session):
game = crud.create_game(session, crew_name="The Black Gats")
assert game.id is not None
assert game.crew_name == "The Black Gats"
assert game.phase == "lobby"
def test_create_game_endpoint():
from fastapi.testclient import TestClient
from sqlalchemy.pool import StaticPool
from sqlmodel import SQLModel, create_engine, Session, select
from pirats.main import app
# Setup in-memory DB
engine = create_engine(
"sqlite://",
connect_args={"check_same_thread": False},
poolclass=StaticPool,
)
SQLModel.metadata.create_all(engine)
session = Session(engine)
def get_session_override():
yield session
from pirats.database import get_session
app.dependency_overrides[get_session] = get_session_override
client = TestClient(app, base_url="http://testserver")
try:
response = client.post("/api/game", data={"crew_name": "The Math Mutineers"})
assert response.status_code == 200
# Verify the database has the game with the crew name
from pirats.models import Game
games = session.exec(select(Game)).all()
assert len(games) == 1
assert games[0].crew_name == "The Math Mutineers"
finally:
app.dependency_overrides.clear()
def test_transition_to_deep_upkeep(session):
game = crud.create_game(session)
p1 = crud.add_player(session, game.id, "P1")
p2 = crud.add_player(session, game.id, "P2")
p1.previous_role = "deep"
p2.previous_role = "pirat"
session.add_all([p1, p2])
session.commit()
crud.transition_to_deep_upkeep(session, game.id)
session.refresh(game)
assert game.phase == "deep_upkeep"
def test_confirm_deep_refresh(session):
game = crud.create_game(session)
p1 = crud.add_player(session, game.id, "P1")
p1.previous_role = "deep"
p1.role = "deep"
p1.rank = 2
# Hand has 3 cards: ['2C', '3C', '4C']
p1.hand_cards = json.dumps(["2C", "3C", "4C"])
# Rest of deck has some known cards
game.deck_cards = json.dumps(["5H", "6H", "7H", "8H"])
session.add_all([game, p1])
session.commit()
# Stub shuffle to keep card order deterministic
import random
orig_shuffle = random.shuffle
random.shuffle = lambda x: None
try:
# Discard 2C and 3C
crud.confirm_deep_refresh(session, p1.id, ["2C", "3C"])
finally:
random.shuffle = orig_shuffle
session.refresh(p1)
session.refresh(game)
# Hand should not contain 2C and 3C, and should have drawn back up to max
# hand size (4: the only player counts as highest rank)
hand = json.loads(p1.hand_cards)
assert "2C" not in hand
assert "3C" not in hand
assert "4C" in hand
assert "5H" in hand
assert "6H" in hand
assert "7H" in hand
assert len(hand) == 4
# Discarded cards should be at the end of the deck
deck = json.loads(game.deck_cards)
assert "2C" in deck
assert "3C" in deck
assert deck == ["8H", "2C", "3C"]
# Since P1 was the only resting deep player and is now ready, phase should have advanced to scene_setup!
assert game.phase == "scene_setup"
def test_submit_vote_endpoint():
from fastapi.testclient import TestClient
from sqlalchemy.pool import StaticPool
from sqlmodel import SQLModel, create_engine, Session
from pirats.main import app
from pirats import crud
# Setup in-memory DB
engine = create_engine(
"sqlite://",
connect_args={"check_same_thread": False},
poolclass=StaticPool,
)
SQLModel.metadata.create_all(engine)
session = Session(engine)
# Create game and players
game = crud.create_game(session)
p1 = crud.add_player(session, game.id, "P1")
p2 = crud.add_player(session, game.id, "P2")
game.phase = "between_scenes"
session.add(game)
session.commit()
def get_session_override():
yield session
from pirats.database import get_session
app.dependency_overrides[get_session] = get_session_override
client = TestClient(app)
try:
response = client.post(f"/api/game/{game.id}/player/{p1.id}/submit-vote", data={"nominated_id": p2.id})
assert response.status_code == 200
assert response.json()["status"] == "ok"
finally:
app.dependency_overrides.clear()
def test_single_eligible_deep_player(session):
game = crud.create_game(session)
p1 = crud.add_player(session, game.id, "P1")
p2 = crud.add_player(session, game.id, "P2")
p3 = crud.add_player(session, game.id, "P3")
# Scene number > 1
game.current_scene_number = 2
# 2 players were Deep in the last scene, 1 was Pi-Rat
p1.previous_role = "deep"
p2.previous_role = "deep"
p3.previous_role = "pirat"
# Set their current roles to something invalid first
# P3 is the only one eligible to be Deep, but plays Pi-Rat. This should fail!
p1.role = "pirat"
p2.role = "pirat"
p3.role = "pirat"
session.add_all([game, p1, p2, p3])
session.commit()
success, msg = crud.confirm_scene_setup(session, game.id)
assert not success
assert "is the only player eligible to play the Deep" in msg
# Now set P3 to play Deep. This should succeed!
p3.role = "deep"
session.add(p3)
session.commit()
success, msg = crud.confirm_scene_setup(session, game.id)
assert success
def test_player_death_toggle(session):
game = crud.create_game(session)
player = crud.add_player(session, game.id, "DeadRat")
assert not player.is_dead
assert not player.completed_personal_3
# Toggle objective 3
crud.toggle_objective(session, game.id, player.id, "personal_3", True)
session.refresh(player)
assert player.is_dead
assert player.completed_personal_3
# Toggle off
crud.toggle_objective(session, game.id, player.id, "personal_3", False)
session.refresh(player)
assert not player.is_dead
assert not player.completed_personal_3
def test_story_bonus_rank_grant(session):
game, deep, (granter, mate) = make_scene_game(session, num_pirats=2)
crud.toggle_objective(session, game.id, granter.id, "personal_3", True)
session.refresh(granter)
assert granter.needs_rank_3_bonus
start_rank = mate.rank
ok, _ = crud.grant_story_bonus_rank(session, game, granter, mate)
assert ok
session.refresh(granter)
session.refresh(mate)
assert not granter.needs_rank_3_bonus
assert mate.rank == start_rank + 1
def test_story_bonus_rank_guards_and_forfeit(session):
game, deep, (granter, mate) = make_scene_game(session, num_pirats=2)
# No bonus pending yet -> refused.
ok, _ = crud.grant_story_bonus_rank(session, game, granter, mate)
assert not ok
crud.toggle_objective(session, game.id, granter.id, "personal_3", True)
session.refresh(granter)
mate_rank = mate.rank
# Can't grant to yourself, nor to the Deep -> refused, prompt stays up.
for bad_target in (granter, deep):
ok, _ = crud.grant_story_bonus_rank(session, game, granter, bad_target)
assert not ok
session.refresh(granter)
assert granter.needs_rank_3_bonus
# Forfeit (no eligible crewmate) clears the prompt without changing any Rank.
ok, _ = crud.grant_story_bonus_rank(session, game, granter, None)
assert ok
session.refresh(granter)
session.refresh(mate)
assert not granter.needs_rank_3_bonus
assert mate.rank == mate_rank
def test_become_ghost_flow(session):
game = crud.create_game(session)
player = crud.add_player(session, game.id, "DeadRat")
player.is_dead = True
session.add(player)
session.commit()
from fastapi.testclient import TestClient
from pirats.main import app
from pirats.database import get_session
def get_session_override():
yield session
app.dependency_overrides[get_session] = get_session_override
client = TestClient(app)
try:
response = client.post(f"/api/game/{game.id}/player/{player.id}/become-ghost")
assert response.status_code == 200
session.refresh(player)
assert player.is_ghost
assert not player.is_dead
finally:
app.dependency_overrides.clear()
def test_recruit_creation_full_flow(session):
"""A dead Pi-Rat re-rolls: queued in between_scenes, created cooperatively
in the recruit_creation phase, then the game advances to scene_setup."""
game = crud.create_game(session)
p1 = crud.add_player(session, game.id, "P1")
p2 = crud.add_player(session, game.id, "P2")
p3 = crud.add_player(session, game.id, "P3")
game.phase = "between_scenes"
game.current_scene_number = 2
p1.is_dead = True
p1.rank = 3
p1.hand_cards = json.dumps(["2C", "3D", "4H"])
p1.completed_personal_1 = True
p1.completed_personal_2 = True
p1.completed_personal_3 = True
p1.name = "Dread Pirate Roberts"
session.add_all([game, p1, p2, p3])
session.commit()
# 1. The dead player opts to re-roll (a living player can't)
ok, msg = crud.choose_reroll(session, p2.id)
assert not ok
ok, msg = crud.choose_reroll(session, p1.id)
assert ok, msg
session.refresh(p1)
assert p1.needs_reroll
assert p1.is_dead # still dead until the recruit is created
# 2. Upkeep ends (no resting Deep) -> detours into recruit_creation
crud.transition_to_deep_upkeep(session, game.id)
session.refresh(game)
session.refresh(p1)
assert game.phase == "recruit_creation"
assert game.current_scene_number == 2 # not incremented yet
# Activation reset the sheet and returned the old hand to the deck
assert p1.rank == 1
assert not p1.is_dead
assert not p1.completed_personal_1
assert p1.avatar_look == ""
assert p1.tech_jack == ""
assert json.loads(p1.hand_cards) == []
deck = json.loads(game.deck_cards)
assert all(c in deck for c in ["2C", "3D", "4H"])
# Crewmates were assigned to answer Like/Hate and write 3 techniques
assert p1.other_like_from_player_id in (p2.id, p3.id)
assert p1.other_hate_from_player_id in (p2.id, p3.id)
assert p1.other_like_from_player_id != p1.other_hate_from_player_id
slots = json.loads(p1.incoming_techniques)
assert len(slots) == 3
assert all(s["from_id"] in (p2.id, p3.id) for s in slots)
assert all(s["text"] == "" for s in slots)
# 3. Can't finalize before the crew has done its part
ok, msg = crud.finalize_recruit(session, p1.id, "a", "b", "c")
assert not ok
# 4. The recruit fills in their basic records
p1.avatar_look = "Shiny eyes"
p1.avatar_smell = "Spiced Rum"
p1.first_words = "No fear!"
p1.good_at_math = "Yes"
p1.name = crud.recruit_name(p1)
session.add(p1)
session.commit()
assert p1.name == "Recruit Spiced Rum"
# 5. Crewmates answer Like/Hate and write the techniques
crud.submit_delegated_answer(session, p1.id, "like", "Great hat", p1.other_like_from_player_id)
crud.submit_delegated_answer(session, p1.id, "hate", "Bad jokes", p1.other_hate_from_player_id)
techs = ["Plank Pirouette", "Bilge Bomb", "Rat King Roar"]
for i, s in enumerate(slots):
# The wrong crewmate can't fill someone else's slot
wrong = p3.id if s["from_id"] == p2.id else p2.id
ok, msg = crud.contribute_recruit_technique(session, wrong, p1.id, i, techs[i])
assert not ok
ok, msg = crud.contribute_recruit_technique(session, s["from_id"], p1.id, i, techs[i])
assert ok, msg
# 6. The recruit assigns J/Q/K and joins the crew
ok, msg = crud.finalize_recruit(session, p1.id, techs[1], techs[0], techs[2])
assert ok, msg
session.refresh(p1)
session.refresh(game)
assert not p1.needs_reroll
assert p1.tech_jack == techs[1]
assert p1.tech_queen == techs[0]
assert p1.tech_king == techs[2]
assert json.loads(p1.swapped_techniques) == techs
assert len(json.loads(p1.hand_cards)) > 0 # starting hand dealt
# Last recruit done -> next scene's setup begins
assert game.phase == "scene_setup"
assert game.current_scene_number == 3
def test_gat_tax_accept(session):
game, deep, (p2, p3) = make_scene_game(session, num_pirats=2)
obs = game.obstacles[0]
# p2 is gatless; p3 has a gat
p3.completed_personal_1 = True
p2.hand_cards = json.dumps(["2C", "3C"])
p3.hand_cards = json.dumps(["10C"])
session.add_all([p2, p3])
session.commit()
ok, msg = crud.create_challenge(session, game.id, deep.id, p2.id, [obs.id])
assert ok
session.refresh(game)
challenge = game.challenges[0]
ok, msg = crud.request_tax(session, challenge.id, p2.id, p3.id)
assert ok, msg
session.refresh(challenge)
assert challenge.tax_state == "requested"
assert challenge.tax_type == "gat"
# Plays are blocked while the tax is pending
ok, msg, res = crud.play_challenge_card(session, p2.id, obs.id, "2C")
assert not ok
ok, msg = crud.respond_tax(session, challenge.id, p3.id, accept=True)
assert ok
session.refresh(challenge)
session.refresh(p2)
session.refresh(p3)
# The gatted pi-rat takes over and received one random card from the requester
assert challenge.acting_player_id == p3.id
assert len(crud.get_player_hand(p2)) == 1
assert len(crud.get_player_hand(p3)) == 2
def test_gat_tax_refuse_and_fail(session):
game, deep, (p2, p3) = make_scene_game(session, num_pirats=2)
obs = game.obstacles[0]
obs.current_value = 13
p3.completed_personal_1 = True
p2.hand_cards = json.dumps(["2C"])
session.add_all([obs, p2, p3])
session.commit()
p2_start_rank = p2.rank
ok, msg = crud.create_challenge(session, game.id, deep.id, p2.id, [obs.id])
assert ok
session.refresh(game)
challenge = game.challenges[0]
ok, msg = crud.request_tax(session, challenge.id, p2.id, p3.id)
assert ok
ok, msg = crud.respond_tax(session, challenge.id, p3.id, accept=False)
assert ok
session.refresh(p2)
session.refresh(p3)
# Refusal: the gat changes paws immediately, requester completes Objective 1 (+1 Rank)
assert p2.completed_personal_1
assert not p3.completed_personal_1
assert p2.rank == p2_start_rank + 1
# The requester attempts the challenge themselves... and fails
ok, msg, res = crud.play_challenge_card(session, p2.id, obs.id, "2C")
assert ok
assert not res["success"]
ok, msg = crud.resolve_challenge(session, challenge.id, deep.id)
assert ok
session.refresh(p2)
session.refresh(p3)
# Failure: the gat goes back, the rank reverts, and no more taxes this scene
assert not p2.completed_personal_1
assert p3.completed_personal_1
assert p2.rank == p2_start_rank
assert p2.tax_banned
def test_name_tax_refuse_steals_name_string(session):
game, deep, (p2, p3) = make_scene_game(session, num_pirats=2)
obs = game.obstacles[0]
obs.current_value = 13
# p2 has a gat but no Name; p3 has both and a fancy name
p2.completed_personal_1 = True
p2.avatar_smell = "smells like burnt cheese"
p3.completed_personal_1 = True
p3.completed_personal_2 = True
p3.name = "Bloodfang the Unforgiven"
p3.avatar_smell = "faintly of gunpowder"
p2.hand_cards = json.dumps(["2C"])
session.add_all([obs, p2, p3])
session.commit()
ok, msg = crud.create_challenge(session, game.id, deep.id, p2.id, [obs.id])
assert ok
session.refresh(game)
challenge = game.challenges[0]
ok, msg = crud.request_tax(session, challenge.id, p2.id, p3.id)
assert ok, msg
assert challenge.tax_type == "name"
ok, msg = crud.respond_tax(session, challenge.id, p3.id, accept=False)
assert ok
session.refresh(p2)
session.refresh(p3)
# The name string itself is stolen; the refuser is demoted to their smell
assert p2.name == "Bloodfang the Unforgiven"
assert p2.completed_personal_2
assert not p2.needs_name
assert p3.name == "Recruit Gunpowder"
assert not p3.completed_personal_2
# The thief fails the challenge: the name returns and the thief is demoted
ok, msg, res = crud.play_challenge_card(session, p2.id, obs.id, "2C")
assert ok and not res["success"]
ok, msg = crud.resolve_challenge(session, challenge.id, deep.id)
assert ok
session.refresh(p2)
session.refresh(p3)
assert p3.name == "Bloodfang the Unforgiven"
assert p3.completed_personal_2
assert p2.name == "Recruit Burnt cheese"
assert not p2.completed_personal_2
assert p2.tax_banned
def test_name_tax_refuse_success_keeps_stolen_name(session):
game, deep, (p2, p3) = make_scene_game(session, num_pirats=2)
obs = game.obstacles[0]
obs.current_value = 1
p2.completed_personal_1 = True
p2.avatar_smell = "old rum"
p3.completed_personal_1 = True
p3.completed_personal_2 = True
p3.name = "Captain Calculus"
p3.avatar_smell = "chalk dust"
p2.hand_cards = json.dumps(["10C"])
session.add_all([obs, p2, p3])
session.commit()
ok, _ = crud.create_challenge(session, game.id, deep.id, p2.id, [obs.id])
assert ok
session.refresh(game)
challenge = game.challenges[0]
ok, _ = crud.request_tax(session, challenge.id, p2.id, p3.id)
assert ok
ok, _ = crud.respond_tax(session, challenge.id, p3.id, accept=False)
assert ok
ok, msg, res = crud.play_challenge_card(session, p2.id, obs.id, "10C")
assert ok and res["success"]
ok, _ = crud.resolve_challenge(session, challenge.id, deep.id)
assert ok
session.refresh(p2)
session.refresh(p3)
# Success: the stolen name stays stolen
assert p2.name == "Captain Calculus"
assert p2.completed_personal_2
assert p3.name == "Recruit Chalk dust"
assert not p3.completed_personal_2
def test_recruit_name_helper():
assert crud.recruit_name_from_smell("smells like burnt cheese") == "Recruit Burnt cheese"
assert crud.recruit_name_from_smell("faintly of gunpowder") == "Recruit Gunpowder"
assert crud.recruit_name_from_smell("") == "Recruit"
assert crud.recruit_name_from_smell("", fallback="Tim") == "Recruit Tim"
def test_pvp_challenge(session):
game, deep, (p2, p3) = make_scene_game(session, num_pirats=2)
p2.hand_cards = json.dumps(["7C"])
p3.hand_cards = json.dumps(["9C"])
session.add_all([p2, p3])
session.commit()
ok, msg = crud.create_pvp_challenge(session, game.id, p2.id, p3.id, "7C")
assert ok, msg
session.refresh(p2)
assert crud.get_player_hand(p2) == [] # temp obstacle card left the hand
session.refresh(game)
duel = [c for c in game.challenges if c.challenge_type == "pvp"][0]
assert duel.temp_card == "7C"
# Defender beats 7 with 9; matching color (both black) -> draws a card back
ok, msg, res = crud.play_pvp_defense(session, duel.id, p3.id, "9C")
assert ok
assert res["success"]
assert res["drew_card"] is not None
session.refresh(duel)
assert duel.status == "succeeded"
def test_end_scene_requires_every_pirat_challenged(session):
game, deep, (p2,) = make_scene_game(session)
game.dev_mode = False
session.add(game)
session.commit()
# Obstacles remain and p2 has not been challenged yet -> blocked.
ok, msg = crud.end_scene_and_transition(session, game.id)
assert not ok
assert p2.name in msg
session.refresh(game)
assert game.phase == "scene"
# Once p2 has faced a Deep Challenge, the scene can end.
obs = game.obstacles[0]
ok, msg = crud.create_challenge(session, game.id, deep.id, p2.id, [obs.id])
assert ok, msg
ok, msg = crud.end_scene_and_transition(session, game.id)
assert ok, msg
session.refresh(game)
assert game.phase == "between_scenes"
def test_end_scene_allowed_when_obstacle_list_empty(session):
game, deep, (p2,) = make_scene_game(session)
game.dev_mode = False
session.add(game)
# Clear the Obstacle List; nobody has been challenged.
for obs in list(game.obstacles):
session.delete(obs)
session.commit()
ok, msg = crud.end_scene_and_transition(session, game.id)
assert ok, msg
session.refresh(game)
assert game.phase == "between_scenes"
def test_end_scene_dev_mode_bypasses_challenge_requirement(session):
game, deep, (p2,) = make_scene_game(session)
assert game.dev_mode # defaults on for a local checkout
ok, msg = crud.end_scene_and_transition(session, game.id)
assert ok, msg
session.refresh(game)
assert game.phase == "between_scenes"
def test_obstacles_persist_across_scenes(session):
game, deep, (p2,) = make_scene_game(session)
assert len(game.obstacles) == 2
surviving_ids = {o.id for o in game.obstacles}
# End the scene, vote, and set up scene 2 with swapped roles
crud.end_scene_and_transition(session, game.id)
crud.transition_to_deep_upkeep(session, game.id)
crud.confirm_deep_refresh(session, deep.id, [])
session.refresh(game)
assert game.phase == "scene_setup"
assert game.current_scene_number == 2
deep.role = "pirat"
p2.role = "deep"
session.add_all([deep, p2])
session.commit()
success, msg = crud.confirm_scene_setup(session, game.id)
assert success, msg
session.refresh(game)
# Unbeaten obstacles remain; the list is already at Deep(1)+1=2, so nothing new is drawn
assert {o.id for o in game.obstacles} == surviving_ids
def test_joker_obstacle_increase_is_permanent(session):
game = crud.create_game(session)
deep = crud.add_player(session, game.id, "Deep1")
deep.rank = 3
deep.role = "deep"
p2 = crud.add_player(session, game.id, "Rat1")
p2.rank = 1
p2.role = "pirat"
game.extra_obstacles = 1 # a Joker was drawn as an Obstacle earlier
session.add_all([game, deep, p2])
session.commit()
# Stub shuffle so the rebuilt deck keeps its construction order (Jokers last)
# and no Joker can be drawn as an Obstacle here
import random
orig_shuffle = random.shuffle
random.shuffle = lambda x: None
try:
success, msg = crud.confirm_scene_setup(session, game.id)
finally:
random.shuffle = orig_shuffle
assert success, msg
session.refresh(game)
# Deep(1) + 1 + permanent increase(1) = 3 obstacles
assert len(game.obstacles) == 3
# The increase is permanent, not consumed
assert game.extra_obstacles == 1
def test_first_scene_role_choice(session):
game = crud.create_game(session)
p1 = crud.add_player(session, game.id, "P1")
p2 = crud.add_player(session, game.id, "P2")
p3 = crud.add_player(session, game.id, "P3")
p1.rank = 3
p2.rank = 1
p3.rank = 2
session.add_all([p1, p2, p3])
session.commit()
# Rank 3 must play the Deep, Rank 1 must play their Pi-Rat, Rank 2 may choose either
assert crud.get_allowed_roles(session, game.id, p1.id) == ["deep"]
assert crud.get_allowed_roles(session, game.id, p2.id) == ["pirat"]
assert set(crud.get_allowed_roles(session, game.id, p3.id)) == {"pirat", "deep"}
# A Rank 2 player choosing Deep alongside the Rank 3 player is legal
p1.role = "deep"
p2.role = "pirat"
p3.role = "deep"
session.add_all([p1, p2, p3])
session.commit()
success, msg = crud.confirm_scene_setup(session, game.id)
assert success, msg
def test_vote_validation(session):
game = crud.create_game(session)
p1 = crud.add_player(session, game.id, "P1")
p2 = crud.add_player(session, game.id, "P2")
p3 = crud.add_player(session, game.id, "P3")
p1.role = "deep"
p2.role = "pirat"
p3.role = "pirat"
session.add_all([p1, p2, p3])
session.commit()
# Self-nomination is rejected
ok, msg = crud.submit_rank_vote(session, game.id, p2.id, p2.id)
assert not ok
# Nominating a previous-scene Deep player is rejected
ok, msg = crud.submit_rank_vote(session, game.id, p2.id, p1.id)
assert not ok
# But the Deep player may vote
ok, msg = crud.submit_rank_vote(session, game.id, p1.id, p2.id)
assert ok
# Dead and awaiting-recruit Pi-Rats can't be nominated
p3.is_dead = True
session.add(p3)
session.commit()
ok, msg = crud.submit_rank_vote(session, game.id, p1.id, p3.id)
assert not ok
p3.is_dead = False
p3.needs_reroll = True
session.add(p3)
session.commit()
ok, msg = crud.submit_rank_vote(session, game.id, p1.id, p3.id)
assert not ok
def test_vote_skips_when_only_dead_options(session):
# 3-player crew, one living Pi-Rat whose only crewmates are the previous
# Deep and a dead rat: they have no eligible nominee and skip voting.
game = crud.create_game(session)
deep = crud.add_player(session, game.id, "Deep")
living = crud.add_player(session, game.id, "Living")
dead = crud.add_player(session, game.id, "Dead")
deep.role = "deep"
living.role = "pirat"
dead.role = "pirat"
dead.is_dead = True
session.add_all([deep, living, dead])
session.commit()
# The living Pi-Rat has no one valid to nominate
assert crud.eligible_vote_nominees(game, living) == []
# The Deep can still nominate the living Pi-Rat
nominees = crud.eligible_vote_nominees(game, deep)
assert [p.id for p in nominees] == [living.id]
def test_rank_up_draws_immediately(session):
game, deep, (p2, p3) = make_scene_game(session, num_pirats=2)
# p2 is rank 1 (lowest -> max 2), p3 is rank 2 (highest -> max 4)
hand_before = len(crud.get_player_hand(p2))
crud.toggle_objective(session, game.id, p2.id, "personal_1", True)
session.refresh(p2)
assert p2.rank == 2
# p2 jumped from lowest (2 cards) to highest-tied (4 cards): draws 2 immediately
assert len(crud.get_player_hand(p2)) == hand_before + 2
def test_auto_delegate_all(session):
game = crud.create_game(session)
p1 = crud.add_player(session, game.id, "P1", is_creator=True)
p2 = crud.add_player(session, game.id, "P2")
p3 = crud.add_player(session, game.id, "P3")
session.refresh(game)
crud.auto_delegate_all(session, game)
players = [p1, p2, p3]
for p in players:
session.refresh(p)
assert p.other_like_from_player_id and p.other_like_from_player_id != p.id
assert p.other_hate_from_player_id and p.other_hate_from_player_id != p.id
# With 3+ players, like and hate go to different crewmates
assert p.other_like_from_player_id != p.other_hate_from_player_id
# Workload is even: each player answers exactly one like and one hate
like_answerers = sorted(p.other_like_from_player_id for p in players)
hate_answerers = sorted(p.other_hate_from_player_id for p in players)
assert like_answerers == sorted(p.id for p in players)
assert hate_answerers == sorted(p.id for p in players)
def test_auto_delegate_all_two_players(session):
game = crud.create_game(session)
p1 = crud.add_player(session, game.id, "P1", is_creator=True)
p2 = crud.add_player(session, game.id, "P2")
session.refresh(game)
crud.auto_delegate_all(session, game)
session.refresh(p1)
session.refresh(p2)
assert p1.other_like_from_player_id == p2.id
assert p1.other_hate_from_player_id == p2.id
assert p2.other_like_from_player_id == p1.id
assert p2.other_hate_from_player_id == p1.id
def test_submit_techniques_rejects_collisions(session):
game = crud.create_game(session)
p1 = crud.add_player(session, game.id, "P1", is_creator=True)
p2 = crud.add_player(session, game.id, "P2")
# Duplicates within a single submission are rejected
err = crud.submit_techniques(session, p1.id, "Tail Whip", "tail whip", "Cheese Decoy")
assert err is not None
session.refresh(p1)
assert json.loads(p1.created_techniques) == []
# Valid submission goes through
err = crud.submit_techniques(session, p1.id, "Tail Whip", "Pocket Sand", "Cheese Decoy")
assert err is None
# A second player colliding (case-insensitively) with the pool is rejected
err = crud.submit_techniques(session, p2.id, "POCKET SAND", "Bilge Bomb", "Rat King Roar")
assert err is not None
session.refresh(p2)
assert json.loads(p2.created_techniques) == []
# And a unique set is accepted, triggering the swap
err = crud.submit_techniques(session, p2.id, "Plank Pirouette", "Bilge Bomb", "Rat King Roar")
assert err is None
session.refresh(game)
assert game.phase == "swap_techniques"
def test_unsubmit_techniques(session):
game = crud.create_game(session)
p1 = crud.add_player(session, game.id, "P1", is_creator=True)
p2 = crud.add_player(session, game.id, "P2")
game.phase = "character_creation"
session.add(game)
session.commit()
# Nothing to unlock before submitting
ok, _ = crud.unsubmit_techniques(session, p1.id)
assert not ok
err = crud.submit_techniques(session, p1.id, "Tail Whip", "Pocket Sand", "Cheese Decoy")
assert err is None
# Unlock clears the submission so it can be revised
ok, msg = crud.unsubmit_techniques(session, p1.id)
assert ok, msg
session.refresh(p1)
assert json.loads(p1.created_techniques) == []
# Resubmitting (with everyone else done) still triggers the swap
err = crud.submit_techniques(session, p2.id, "Plank Pirouette", "Bilge Bomb", "Rat King Roar")
assert err is None
err = crud.submit_techniques(session, p1.id, "Tail Whip", "Pocket Sand", "Cheese Decoy")
assert err is None
session.refresh(game)
assert game.phase == "swap_techniques"
# Once swapping has begun, revision is locked out
ok, _ = crud.unsubmit_techniques(session, p1.id)
assert not ok
def test_unassign_face_techniques(session):
game, p1, p2, p3 = make_swap_game(session)
crud.auto_assign_techniques(session, game)
session.refresh(game)
assert game.phase == "scene_setup"
# Wrong phase is rejected
ok, _ = crud.unassign_face_techniques(session, p1.id)
assert not ok
# Fresh game stopped in assign_techniques: two assign, one revises
game, p1, p2, p3 = make_swap_game(session)
game.phase = "assign_techniques"
for p in (p1, p2, p3):
p.swapped_techniques = p.created_techniques
session.add(p)
session.add(game)
session.commit()
# Not assigned yet -> nothing to unlock
ok, _ = crud.unassign_face_techniques(session, p1.id)
assert not ok
t1 = json.loads(p1.swapped_techniques)
ok, msg = crud.assign_face_card_techniques(session, p1.id, t1[0], t1[1], t1[2])
assert ok, msg
ok, msg = crud.unassign_face_techniques(session, p1.id)
assert ok, msg
session.refresh(p1)
assert not p1.tech_jack and not p1.tech_queen and not p1.tech_king
assert not p1.is_ready
# Others finishing while p1 is revising must not advance the game
for p in (p2, p3):
t = json.loads(p.swapped_techniques)
ok, msg = crud.assign_face_card_techniques(session, p.id, t[0], t[1], t[2])
assert ok, msg
session.refresh(game)
assert game.phase == "assign_techniques"
# p1 reassigns (swapped order) and the game moves on
ok, msg = crud.assign_face_card_techniques(session, p1.id, t1[2], t1[0], t1[1])
assert ok, msg
session.refresh(game)
assert game.phase == "scene_setup"
def make_swap_game(session):
"""Helper: 3 players in the swap_techniques phase, each holding their own techniques."""
game = crud.create_game(session)
p1 = crud.add_player(session, game.id, "Carlos", is_creator=True)
p2 = crud.add_player(session, game.id, "Barry")
p3 = crud.add_player(session, game.id, "Jaspar")
crud.submit_techniques(session, p1.id, "A1", "A2", "A3")
crud.submit_techniques(session, p2.id, "B1", "B2", "B3")
crud.submit_techniques(session, p3.id, "C1", "C2", "C3")
session.refresh(game)
assert game.phase == "swap_techniques"
return game, p1, p2, p3
def test_swap_offer_validation(session):
game, p1, p2, p3 = make_swap_game(session)
# Can't offer a technique you don't hold, or swap with yourself
ok, _ = crud.offer_swap(session, p1.id, p2.id, "B1")
assert not ok
ok, _ = crud.offer_swap(session, p1.id, p1.id, "A1")
assert not ok
# One pending offer at a time
ok, msg = crud.offer_swap(session, p1.id, p2.id, "A1")
assert ok, msg
ok, _ = crud.offer_swap(session, p1.id, p3.id, "A2")
assert not ok
ok, msg = crud.cancel_swap_offer(session, p1.id)
assert ok, msg
session.refresh(p1)
assert p1.swap_offer_to_id is None and p1.swap_offer_technique == ""
# Accepting must not trade back a technique the offerer created
ok, msg = crud.offer_swap(session, p1.id, p2.id, "A1")
assert ok, msg
crud.respond_swap_offer(session, p2.id, p1.id, True, "B1")
session.refresh(p2)
# p2 now holds A1 (created by p1) and can't offer it back to p1
ok, _ = crud.offer_swap(session, p2.id, p1.id, "A1")
assert not ok
ok, msg = crud.offer_swap(session, p1.id, p2.id, "A2")
assert ok, msg
ok, _ = crud.respond_swap_offer(session, p2.id, p1.id, True, "A1")
assert not ok # A1 was created by the offerer
# Declining clears the offer
ok, msg = crud.respond_swap_offer(session, p2.id, p1.id, False)
assert ok, msg
session.refresh(p1)
assert p1.swap_offer_to_id is None
# Only the actual recipient can respond
ok, msg = crud.offer_swap(session, p1.id, p2.id, "A2")
assert ok, msg
ok, _ = crud.respond_swap_offer(session, p3.id, p1.id, True, "C1")
assert not ok
def test_swap_accept_exchanges_and_withdraws_stale_offer(session):
game, p1, p2, p3 = make_swap_game(session)
# p2 promises B1 to p3, then trades B1 away to p1 instead
ok, msg = crud.offer_swap(session, p2.id, p3.id, "B1")
assert ok, msg
ok, msg = crud.offer_swap(session, p1.id, p2.id, "A1")
assert ok, msg
ok, msg = crud.respond_swap_offer(session, p2.id, p1.id, True, "B1")
assert ok, msg
session.refresh(p1)
session.refresh(p2)
held_p1 = json.loads(p1.held_techniques)
held_p2 = json.loads(p2.held_techniques)
# The techniques moved with their creator tags intact
assert {"text": "B1", "creator_id": p2.id} in held_p1
assert {"text": "A1", "creator_id": p1.id} in held_p2
assert not any(t["text"] == "A1" for t in held_p1)
# p2's now-impossible offer to p3 was withdrawn
assert p2.swap_offer_to_id is None and p2.swap_offer_technique == ""
def test_auto_assign_techniques(session):
game, p1, p2, p3 = make_swap_game(session)
ok, msg = crud.auto_assign_techniques(session, game)
assert ok, msg
session.refresh(game)
assert game.phase == "scene_setup"
for p in (p1, p2, p3):
session.refresh(p)
assert p.tech_jack and p.tech_queen and p.tech_king
own = {t.lower() for t in json.loads(p.created_techniques)}
held = {p.tech_jack.lower(), p.tech_queen.lower(), p.tech_king.lower()}
assert not (own & held)
# Ranks were rolled too
assert sorted(p.rank for p in (p1, p2, p3)) == [1, 2, 3]
# Wrong phase is rejected
ok, _ = crud.auto_assign_techniques(session, game)
assert not ok
def test_swap_offer_technique_hidden_in_state(session):
from fastapi.testclient import TestClient
from pirats.main import app
from pirats.database import get_session
game, p1, p2, p3 = make_swap_game(session)
ok, msg = crud.offer_swap(session, p1.id, p2.id, "A1")
assert ok, msg
def get_session_override():
yield session
app.dependency_overrides[get_session] = get_session_override
client = TestClient(app, base_url="http://testserver")
try:
# The recipient sees that an offer exists, but not which technique
state = client.get(f"/api/game/{game.id}/player/{p2.id}/state").json()
p1_blob = next(p for p in state["players"] if p["id"] == p1.id)
assert p1_blob["swap_offer_to_id"] == p2.id
assert "swap_offer_technique" not in p1_blob
# The offerer still sees their own offered technique
state = client.get(f"/api/game/{game.id}/player/{p1.id}/state").json()
assert state["player"]["swap_offer_technique"] == "A1"
p1_blob = next(p for p in state["players"] if p["id"] == p1.id)
assert p1_blob["swap_offer_technique"] == "A1"
finally:
app.dependency_overrides.clear()
def test_recruit_smell_name_normalization(session):
game = crud.create_game(session)
p1 = crud.add_player(session, game.id, "P1")
p1.avatar_smell = "suspiciously smelling of ink and sour lemons"
assert crud.recruit_name(p1) == "Recruit Ink and sour lemons"
def test_contribute_technique_rejects_collisions(session):
game = crud.create_game(session)
p1 = crud.add_player(session, game.id, "P1")
p2 = crud.add_player(session, game.id, "P2")
p2.tech_jack = "Parabolic Boarding Bounce"
p1.is_dead = True
p1.needs_reroll = True
session.add_all([p1, p2])
session.commit()
crud.start_recruit_creation(session, game)
session.refresh(p1)
slots = json.loads(p1.incoming_techniques)
assert all(s["from_id"] == p2.id for s in slots)
# A name already on another sheet is rejected (case-insensitively)
ok, msg = crud.contribute_recruit_technique(session, p2.id, p1.id, 0, "parabolic boarding bounce")
assert not ok
ok, msg = crud.contribute_recruit_technique(session, p2.id, p1.id, 0, "Bilge Bomb")
assert ok, msg
# A name already written into another of the recruit's slots is rejected too
ok, msg = crud.contribute_recruit_technique(session, p2.id, p1.id, 1, "Bilge Bomb")
assert not ok
def test_solo_recruit_gets_pool_techniques(session):
"""With no crewmates to ask, the recruit's techniques come from the pool."""
game = crud.create_game(session)
p1 = crud.add_player(session, game.id, "P1")
p1.is_dead = True
p1.needs_reroll = True
session.add(p1)
session.commit()
crud.start_recruit_creation(session, game)
session.refresh(p1)
slots = json.loads(p1.incoming_techniques)
assert len(slots) == 3
assert all(s["from_id"] is None for s in slots)
assert all(s["text"] in cards.TECHNIQUE_SUGGESTIONS for s in slots)
assert p1.other_like_from_player_id is None
def test_late_joiner_queued_for_recruit_creation(session):
game, deep, (p2,) = make_scene_game(session)
assert game.phase == "scene"
newbie = crud.add_player(session, game.id, "Latecomer")
assert newbie.needs_reroll
assert newbie.role is None
assert json.loads(newbie.hand_cards) == []
# Scene ends, everyone votes/readies, resting Deep refreshes...
crud.end_scene_and_transition(session, game.id)
crud.transition_to_deep_upkeep(session, game.id)
session.refresh(game)
assert game.phase == "deep_upkeep"
crud.confirm_deep_refresh(session, deep.id, [])
session.refresh(game)
# ...and instead of scene_setup we land in recruit_creation
assert game.phase == "recruit_creation"
session.refresh(newbie)
slots = json.loads(newbie.incoming_techniques)
assert len(slots) == 3
assert all(s["from_id"] in (deep.id, p2.id) for s in slots)
def test_reshuffle_keeps_open_pvp_temp_card_out_of_deck(session):
game, deep, (p2, p3) = make_scene_game(session, num_pirats=2)
p2.hand_cards = json.dumps(["7C"])
session.add(p2)
session.commit()
ok, msg = crud.create_pvp_challenge(session, game.id, p2.id, p3.id, "7C")
assert ok, msg
# A mid-duel reshuffle must not pull the temporary Obstacle back into the deck
session.refresh(game)
crud.reshuffle_discard_pile(session, game)
assert "7C" not in crud.get_game_deck(game)
# Once the duel resolves, the card is discarded and reshuffles normally
duel = [c for c in game.challenges if c.challenge_type == "pvp"][0]
p3.hand_cards = json.dumps(["9C"])
session.add(p3)
session.commit()
ok, _, _ = crud.play_pvp_defense(session, duel.id, p3.id, "9C")
assert ok
crud.reshuffle_discard_pile(session, game)
assert "7C" in crud.get_game_deck(game)
def test_technique_suggestions_endpoint():
from fastapi.testclient import TestClient
from pirats.main import app
client = TestClient(app, base_url="http://testserver")
response = client.get("/api/suggestions/techniques")
assert response.status_code == 200
techs = response.json()["techniques"]
assert techs == cards.TECHNIQUE_SUGGESTIONS
assert len(techs) >= 100
def test_admin_key_returned_at_create_but_hidden_from_state():
from fastapi.testclient import TestClient
from sqlalchemy.pool import StaticPool
from sqlmodel import SQLModel, create_engine, Session
from pirats.main import app
engine = create_engine(
"sqlite://",
connect_args={"check_same_thread": False},
poolclass=StaticPool,
)
SQLModel.metadata.create_all(engine)
session = Session(engine)
def get_session_override():
yield session
from pirats.database import get_session
app.dependency_overrides[get_session] = get_session_override
client = TestClient(app, base_url="http://testserver")
try:
created = client.post("/api/game", data={"crew_name": "Leak Test"}).json()
assert created["admin_key"]
game_id = created["id"]
# First joiner is the creator (an Admin), so they DO see the key
creator_id = client.post(f"/api/game/{game_id}/join", data={"name": "Carlos"}).json()["id"]
state = client.get(f"/api/game/{game_id}/player/{creator_id}/state").json()
assert state["game"]["admin_key"] == created["admin_key"]
# Regular players don't
player_id = client.post(f"/api/game/{game_id}/join", data={"name": "Swabbie"}).json()["id"]
state = client.get(f"/api/game/{game_id}/player/{player_id}/state").json()
assert "admin_key" not in state["game"]
finally:
app.dependency_overrides.clear()
def test_event_log_pagination(session):
from pirats.models import GameEvent
game = crud.create_game(session)
for i in range(120):
# Explicit timestamps keep ordering deterministic
session.add(GameEvent(game_id=game.id, timestamp=1000.0 + i, message=f"event {i}", kind="info"))
session.commit()
# Newest page
events, have_more = crud.get_events_page(session, game.id, limit=50)
assert have_more
assert len(events) == 50
assert events[0].message == "event 70"
assert events[-1].message == "event 119"
# Page back from the oldest loaded event
older, have_more = crud.get_events_page(session, game.id, before=events[0].timestamp, limit=50)
assert have_more
assert older[0].message == "event 20"
assert older[-1].message == "event 69"
# Final page is short and reports no more history
oldest, have_more = crud.get_events_page(session, game.id, before=older[0].timestamp, limit=50)
assert not have_more
assert len(oldest) == 20
assert oldest[0].message == "event 0"
assert oldest[-1].message == "event 19"
# Exact page boundary still reports no remaining history
exact, have_more = crud.get_events_page(session, game.id, before=1020.0, limit=20)
assert not have_more
assert [e.message for e in exact] == [f"event {i}" for i in range(20)]
def test_websocket_state_push():
"""A successful POST under /api/game/{gid}/ pings that game's websockets."""
from fastapi.testclient import TestClient
from sqlmodel.pool import StaticPool
from pirats.main import app
from pirats.database import get_session
engine = create_engine(
"sqlite://",
connect_args={"check_same_thread": False},
poolclass=StaticPool,
)
SQLModel.metadata.create_all(engine)
def override_session():
with Session(engine) as s:
yield s
app.dependency_overrides[get_session] = override_session
try:
client = TestClient(app)
gid = client.post("/api/game", data={"crew_name": "Testers"}).json()["id"]
with client.websocket_connect(f"/api/game/{gid}/ws") as ws:
res = client.post(f"/api/game/{gid}/join", data={"name": "Carlos"})
assert res.status_code == 200
assert ws.receive_json() == {"type": "state_changed"}
# A POST for a different game must not ping this socket
other_gid = client.post("/api/game", data={"crew_name": "Others"}).json()["id"]
client.post(f"/api/game/{other_gid}/join", data={"name": "Barry"})
client.post(f"/api/game/{gid}/join", data={"name": "Jaspar"})
assert ws.receive_json() == {"type": "state_changed"}
finally:
app.dependency_overrides.clear()
def test_skip_character_creation(session):
game = crud.create_game(session)
p1 = crud.add_player(session, game.id, "Carlos", is_creator=True)
p2 = crud.add_player(session, game.id, "Barry")
p3 = crud.add_player(session, game.id, "Jaspar")
session.refresh(game)
game.phase = "character_creation"
game.dev_mode = True
session.add(game)
session.commit()
# p1 did part of the work by hand; the skip must not overwrite it
p1.avatar_look = "Hand-written bandana"
session.add(p1)
session.commit()
crud.submit_techniques(session, p1.id, "Carlos Strike", "Hide", "Roll")
fills = {p.id: {
"avatar_look": f"Look for {p.name}",
"avatar_smell": "of test cheese",
"first_words": "Test!",
"good_at_math": "Only in tests",
"like": "Their test coverage",
"hate": "Their flaky tests",
} for p in game.players}
ok, msg = crud.skip_character_creation(session, game, fills)
assert ok, msg
session.refresh(game)
assert game.phase == "scene_setup"
for p in (p1, p2, p3):
session.refresh(p)
assert p.avatar_look and p.avatar_smell and p.first_words and p.good_at_math
assert p.other_like and p.other_hate
assert p.tech_jack and p.tech_queen and p.tech_king
# Everyone holds techniques created by someone else
own = {t.lower() for t in json.loads(p.created_techniques)}
held = {p.tech_jack.lower(), p.tech_queen.lower(), p.tech_king.lower()}
assert not (own & held)
session.refresh(p1)
assert p1.avatar_look == "Hand-written bandana"
assert json.loads(p1.created_techniques) == ["Carlos Strike", "Hide", "Roll"]
# Ranks were rolled: one rank-3 Deep, one rank-1 Pi-Rat, rest rank 2
ranks = sorted(p.rank for p in (p1, p2, p3))
assert ranks == [1, 2, 3]
def test_skip_character_creation_wrong_phase(session):
game = crud.create_game(session)
crud.add_player(session, game.id, "Carlos", is_creator=True)
session.refresh(game)
ok, msg = crud.skip_character_creation(session, game, {})
assert not ok
def test_dev_mode_default_env(session, monkeypatch):
# Unset = local checkout, defaults on
monkeypatch.delenv("PIRATS_DEV_MODE", raising=False)
assert crud.create_game(session).dev_mode is True
# Set = deployment controls it explicitly
monkeypatch.setenv("PIRATS_DEV_MODE", "false")
assert crud.create_game(session).dev_mode is False
monkeypatch.setenv("PIRATS_DEV_MODE", "true")
assert crud.create_game(session).dev_mode is True
def test_grant_and_revoke_admin():
from fastapi.testclient import TestClient
from sqlalchemy.pool import StaticPool
from sqlmodel import SQLModel, create_engine, Session
from pirats.main import app
from pirats import crud
engine = create_engine(
"sqlite://",
connect_args={"check_same_thread": False},
poolclass=StaticPool,
)
SQLModel.metadata.create_all(engine)
session = Session(engine)
game = crud.create_game(session)
creator = crud.add_player(session, game.id, "Creator", is_creator=True)
mate = crud.add_player(session, game.id, "Mate")
assert creator.is_admin
assert not mate.is_admin
def get_session_override():
yield session
from pirats.database import get_session
app.dependency_overrides[get_session] = get_session_override
client = TestClient(app)
try:
# Admin-only endpoints reject non-admins
response = client.post(f"/api/game/{game.id}/player/{mate.id}/dev-mode", data={"enabled": True})
assert response.status_code == 403
# Wrong key can't grant
response = client.post(f"/api/game/{game.id}/admin/set-admin",
data={"key": "nope", "target_player_id": mate.id, "is_admin": True})
assert response.status_code == 403
# Grant with the real key
response = client.post(f"/api/game/{game.id}/admin/set-admin",
data={"key": game.admin_key, "target_player_id": mate.id, "is_admin": True})
assert response.status_code == 200
session.refresh(mate)
assert mate.is_admin
# The new admin can use admin-only endpoints and sees the admin key in state
response = client.post(f"/api/game/{game.id}/player/{mate.id}/dev-mode", data={"enabled": True})
assert response.status_code == 200
response = client.get(f"/api/game/{game.id}/player/{mate.id}/state")
assert response.json()["game"]["admin_key"] == game.admin_key
# ...but non-admins still don't
response = client.get(f"/api/game/{game.id}/player/{creator.id}/state")
assert "admin_key" in response.json()["game"] # creator is admin too
# Revoke works on regular admins, never on the creator
response = client.post(f"/api/game/{game.id}/admin/set-admin",
data={"key": game.admin_key, "target_player_id": mate.id, "is_admin": False})
assert response.status_code == 200
session.refresh(mate)
assert not mate.is_admin
response = client.post(f"/api/game/{game.id}/admin/set-admin",
data={"key": game.admin_key, "target_player_id": creator.id, "is_admin": False})
assert response.status_code == 400
session.refresh(creator)
assert creator.is_admin
# Once revoked, the state blob hides the key again
response = client.get(f"/api/game/{game.id}/player/{mate.id}/state")
assert "admin_key" not in response.json()["game"]
finally:
app.dependency_overrides.clear()
def test_player_minimum(session):
game = crud.create_game(session)
game.dev_mode = False
session.add(game)
session.commit()
crud.add_player(session, game.id, "P1", is_creator=True)
crud.add_player(session, game.id, "P2")
session.refresh(game)
# Two players can't start a scene
assert not crud.has_min_players(game)
ok, msg = crud.confirm_scene_setup(session, game.id)
assert not ok
assert "at least 3" in msg
# Dev Mode bypasses the minimum for testing
game.dev_mode = True
session.add(game)
session.commit()
assert crud.has_min_players(game)
# Three players is enough regardless
game.dev_mode = False
session.add(game)
session.commit()
crud.add_player(session, game.id, "P3")
session.refresh(game)
assert crud.has_min_players(game)
def test_lobby_start_requires_min_players():
from fastapi.testclient import TestClient
from sqlalchemy.pool import StaticPool
from sqlmodel import SQLModel, create_engine, Session
from pirats.main import app
from pirats import crud
engine = create_engine(
"sqlite://",
connect_args={"check_same_thread": False},
poolclass=StaticPool,
)
SQLModel.metadata.create_all(engine)
session = Session(engine)
game = crud.create_game(session)
game.dev_mode = False
session.add(game)
session.commit()
crud.add_player(session, game.id, "P1", is_creator=True)
crud.add_player(session, game.id, "P2")
def get_session_override():
yield session
from pirats.database import get_session
app.dependency_overrides[get_session] = get_session_override
client = TestClient(app)
try:
response = client.post(f"/api/game/{game.id}/lobby/start")
assert response.status_code == 400
assert "at least 3" in response.json()["error"]
session.refresh(game)
assert game.phase == "lobby"
crud.add_player(session, game.id, "P3")
response = client.post(f"/api/game/{game.id}/lobby/start")
assert response.status_code == 200
session.refresh(game)
assert game.phase == "character_creation"
finally:
app.dependency_overrides.clear()
def test_kick_unblocks_technique_submission(session):
"""A vanished player who never submitted techniques can be kicked, and the
swap phase then begins for the rest of the crew. Their hand goes to discard."""
game = crud.create_game(session)
p1 = crud.add_player(session, game.id, "P1", is_creator=True)
p2 = crud.add_player(session, game.id, "P2")
p3 = crud.add_player(session, game.id, "P3")
game.phase = "character_creation"
session.add(game)
session.commit()
crud.submit_techniques(session, p1.id, "Alpha", "Beta", "Gamma")
crud.submit_techniques(session, p2.id, "Delta", "Epsilon", "Zeta")
# p3 vanished without submitting; give them a hand so we can track the discard
p3.hand_cards = json.dumps(["2C", "3D", "4H"])
session.add(p3)
session.commit()
session.refresh(game)
assert game.phase == "character_creation" # still blocked on p3
ok, msg = crud.kick_player(session, game, p3)
assert ok, msg
session.refresh(game)
assert game.phase == "swap_techniques" # p1 & p2 submitted -> swap begins
assert crud.get_player(session, p3.id) is None
assert len(game.players) == 2
# p3's cards return to the discard: the next reshuffle pulls them into the deck
crud.reshuffle_discard_pile(session, game)
deck = crud.get_game_deck(game)
assert all(c in deck for c in ["2C", "3D", "4H"])
def test_kick_cleans_up_references(session):
"""Kicking a player drops their captaincy, their votes, and any open
Challenge they're part of."""
from pirats.models import Vote
game, deep, (p2, p3) = make_scene_game(session, num_pirats=2)
crud.set_captain(session, game, p2.id, reason="test")
session.refresh(game)
assert game.captain_player_id == p2.id
session.add(Vote(game_id=game.id, voter_player_id=p3.id, nominated_player_id=p2.id))
session.commit()
obstacle = game.obstacles[0]
ok, msg = crud.create_challenge(session, game.id, deep.id, p2.id, [obstacle.id])
assert ok, msg
assert any(c.target_player_id == p2.id for c in game.challenges)
ok, msg = crud.kick_player(session, game, p2)
assert ok, msg
session.refresh(game)
assert crud.get_player(session, p2.id) is None
assert game.captain_player_id is None
assert game.votes == []
assert len(game.challenges) == 0
def test_kick_reassigns_recruit_obligations(session):
"""If a crewmate who owes a recruit Like/Hate answers or technique slots
vanishes, kicking them hands those duties to a present crewmate so the
recruit can still be finalized."""
game = crud.create_game(session)
p1 = crud.add_player(session, game.id, "P1", is_creator=True)
p2 = crud.add_player(session, game.id, "P2")
p3 = crud.add_player(session, game.id, "P3")
game.phase = "between_scenes"
game.current_scene_number = 2
p1.is_dead = True
session.add_all([game, p1])
session.commit()
ok, msg = crud.choose_reroll(session, p1.id)
assert ok, msg
crud.transition_to_deep_upkeep(session, game.id)
session.refresh(game)
assert game.phase == "recruit_creation"
# Stack every recruit duty onto p2 so kicking them exercises the reassignment
session.refresh(p1)
p1.other_like_from_player_id = p2.id
p1.other_hate_from_player_id = p2.id
p1.other_like = ""
p1.other_hate = ""
slots = json.loads(p1.incoming_techniques)
for s in slots:
s["from_id"] = p2.id
s["text"] = ""
p1.incoming_techniques = json.dumps(slots)
session.add(p1)
session.commit()
ok, msg = crud.kick_player(session, game, p2)
assert ok, msg
session.refresh(game)
session.refresh(p1)
assert game.phase == "recruit_creation" # recruit still pending
assert crud.get_player(session, p2.id) is None
# p2's duties were handed to the only remaining crewmate, p3
assert p1.other_like_from_player_id == p3.id
assert p1.other_hate_from_player_id == p3.id
assert all(s["from_id"] == p3.id for s in json.loads(p1.incoming_techniques))
# p3 can now complete everything and the recruit finalizes
crud.submit_delegated_answer(session, p1.id, "like", "Sharp wit", p3.id)
crud.submit_delegated_answer(session, p1.id, "hate", "Slow walkers", p3.id)
techs = ["Cutlass Cartwheel", "Powder Keg Punt", "Crow's Nest Cry"]
for i, t in enumerate(techs):
ok, msg = crud.contribute_recruit_technique(session, p3.id, p1.id, i, t)
assert ok, msg
p1.avatar_look = "Scar"
p1.avatar_smell = "Brine"
p1.first_words = "Arr"
p1.good_at_math = "Nope"
session.add(p1)
session.commit()
ok, msg = crud.finalize_recruit(session, p1.id, techs[0], techs[1], techs[2])
assert ok, msg
session.refresh(game)
assert game.phase == "scene_setup"
def test_kick_endpoint_and_last_admin_guard():
from fastapi.testclient import TestClient
from sqlalchemy.pool import StaticPool
from sqlmodel import SQLModel, create_engine, Session
from pirats.main import app
from pirats import crud
engine = create_engine(
"sqlite://",
connect_args={"check_same_thread": False},
poolclass=StaticPool,
)
SQLModel.metadata.create_all(engine)
session = Session(engine)
game = crud.create_game(session)
creator = crud.add_player(session, game.id, "Creator", is_creator=True)
mate = crud.add_player(session, game.id, "Mate")
extra = crud.add_player(session, game.id, "Extra")
def get_session_override():
yield session
from pirats.database import get_session
app.dependency_overrides[get_session] = get_session_override
client = TestClient(app)
try:
# Wrong key can't kick
r = client.post(f"/api/game/{game.id}/admin/kick",
data={"key": "nope", "target_player_id": mate.id})
assert r.status_code == 403
# The sole Admin (creator) can't be kicked — the table would be locked out
r = client.post(f"/api/game/{game.id}/admin/kick",
data={"key": game.admin_key, "target_player_id": creator.id})
assert r.status_code == 400
assert "last Admin" in r.json()["error"]
# A regular player can be kicked
r = client.post(f"/api/game/{game.id}/admin/kick",
data={"key": game.admin_key, "target_player_id": mate.id})
assert r.status_code == 200
assert crud.get_player(session, mate.id) is None
# With a co-admin present, even an admin (the creator) can be kicked
client.post(f"/api/game/{game.id}/admin/set-admin",
data={"key": game.admin_key, "target_player_id": extra.id, "is_admin": True})
r = client.post(f"/api/game/{game.id}/admin/kick",
data={"key": game.admin_key, "target_player_id": creator.id})
assert r.status_code == 200
assert crud.get_player(session, creator.id) is None
finally:
app.dependency_overrides.clear()
def test_leave_game_unblocks_phase(session):
"""A player who leaves on their own is removed like a kick, and the phase
advances if they were the last blocker. The log shows a leave, not a kick."""
game = crud.create_game(session)
p1 = crud.add_player(session, game.id, "P1", is_creator=True)
p2 = crud.add_player(session, game.id, "P2")
p3 = crud.add_player(session, game.id, "P3")
game.phase = "character_creation"
session.add(game)
session.commit()
crud.submit_techniques(session, p1.id, "Alpha", "Beta", "Gamma")
crud.submit_techniques(session, p2.id, "Delta", "Epsilon", "Zeta")
session.refresh(game)
assert game.phase == "character_creation" # still blocked on p3
ok, msg = crud.leave_game(session, game, p3)
assert ok, msg
session.refresh(game)
assert game.phase == "swap_techniques"
assert crud.get_player(session, p3.id) is None
assert any("P3 left the crew" in e.message for e in game.events)
def test_leave_endpoint_and_last_admin_guard():
from fastapi.testclient import TestClient
from sqlalchemy.pool import StaticPool
from sqlmodel import SQLModel, create_engine, Session
from pirats.main import app
from pirats import crud
engine = create_engine(
"sqlite://",
connect_args={"check_same_thread": False},
poolclass=StaticPool,
)
SQLModel.metadata.create_all(engine)
session = Session(engine)
game = crud.create_game(session)
creator = crud.add_player(session, game.id, "Creator", is_creator=True)
mate = crud.add_player(session, game.id, "Mate")
extra = crud.add_player(session, game.id, "Extra")
def get_session_override():
yield session
from pirats.database import get_session
app.dependency_overrides[get_session] = get_session_override
client = TestClient(app)
try:
# The sole Admin can't abandon the table without handing off Admin first
r = client.post(f"/api/game/{game.id}/player/{creator.id}/leave")
assert r.status_code == 400
assert "last Admin" in r.json()["error"]
# A regular player can leave themselves (no admin key needed)
r = client.post(f"/api/game/{game.id}/player/{mate.id}/leave")
assert r.status_code == 200
assert crud.get_player(session, mate.id) is None
# With a co-admin present, even the creator can leave
client.post(f"/api/game/{game.id}/admin/set-admin",
data={"key": game.admin_key, "target_player_id": extra.id, "is_admin": True})
r = client.post(f"/api/game/{game.id}/player/{creator.id}/leave")
assert r.status_code == 200
assert crud.get_player(session, creator.id) is None
finally:
app.dependency_overrides.clear()
# --- Rollback ---
def _checkpoints(session, game):
return session.exec(
select(Checkpoint).where(Checkpoint.game_id == game.id).order_by(Checkpoint.id)
).all()
def _events(session, game):
return session.exec(select(GameEvent).where(GameEvent.game_id == game.id)).all()
def test_rollback_capture_tags_only_new_events(session):
game, deep, (p2,) = make_scene_game(session)
# Simulate the middleware having sealed the lobby/setup events (it does so on
# every non-scene mutation); now only fresh scene actions get tagged.
crud.seal_and_purge(session, game)
assert all(e.checkpoint_id == 0 for e in _events(session, game))
crud.add_game_event(session, game.id, "Action A", kind="info")
cp = crud.capture_checkpoint(session, game)
tagged = [e for e in _events(session, game) if e.message == "Action A"]
assert tagged and all(e.checkpoint_id == cp.id for e in tagged)
# The previously-sealed events are untouched.
assert all(e.checkpoint_id == 0 for e in _events(session, game) if e.message != "Action A")
def test_rollback_round_trip_restores_state(session):
game, deep, (p2,) = make_scene_game(session)
crud.seal_and_purge(session, game)
obs = game.obstacles[0]
p2.hand_cards = json.dumps(["10D", "2H", "JS", "3C"])
obs.current_value = 5
session.add_all([p2, obs])
session.commit()
saved_hand = json.loads(p2.hand_cards)
saved_deck = crud.get_game_deck(game)
crud.add_game_event(session, game.id, "Known state", kind="info")
cp_known = crud.capture_checkpoint(session, game)
# A destructive action: open a challenge and play a card against the obstacle.
crud.create_challenge(session, game.id, deep.id, p2.id, [obs.id], stakes_failure="x")
crud.capture_checkpoint(session, game)
crud.play_challenge_card(session, p2.id, obs.id, "10D")
crud.capture_checkpoint(session, game)
session.refresh(obs)
assert obs.current_value == 10 # changed
assert len(_checkpoints(session, game)) == 3
ok, msg = crud.rollback_to_checkpoint(session, game, deep, cp_known.id)
assert ok, msg
# State restored exactly; the challenge (created after cp_known) is gone.
obs_after = session.get(Obstacle, obs.id)
p2_after = session.get(Player, p2.id)
session.refresh(game)
assert obs_after.current_value == 5
assert crud.get_player_hand(p2_after) == saved_hand
assert crud.get_game_deck(game) == saved_deck
assert session.exec(select(Challenge).where(Challenge.game_id == game.id)).all() == []
def _scene_with_two_checkpoints(session, num_pirats=1):
"""A scene with sealed history and two action checkpoints (cp1 'First', cp2 'Second')."""
game, deep, pirats = make_scene_game(session, num_pirats=num_pirats)
crud.seal_and_purge(session, game)
crud.add_game_event(session, game.id, "First", kind="info")
cp1 = crud.capture_checkpoint(session, game)
crud.add_game_event(session, game.id, "Second", kind="info")
cp2 = crud.capture_checkpoint(session, game)
return game, deep, pirats, cp1, cp2
def test_rollback_keeps_future_and_sets_head(session):
game, deep, _, cp1, cp2 = _scene_with_two_checkpoints(session)
v0 = game.rollback_timeline_version
ok, msg = crud.rollback_to_checkpoint(session, game, deep, cp1.id)
assert ok, msg
session.refresh(game)
# Non-destructive: the later checkpoint and its event are kept (the frontend greys
# them); head points at cp1 and the version is unchanged.
assert {c.id for c in _checkpoints(session, game)} == {cp1.id, cp2.id}
assert {e.message for e in _events(session, game)} >= {"First", "Second"}
assert game.rollback_head_checkpoint_id == cp1.id
assert game.rollback_timeline_version == v0
def test_latest_checkpoint_id(session):
game, deep, _, cp1, cp2 = _scene_with_two_checkpoints(session)
# Drives the "you are here" button-hiding: during live play this is the current point.
assert crud.latest_checkpoint_id(session, game.id) == cp2.id
assert crud.latest_checkpoint_id(session, crud.create_game(session).id) is None
def test_redo_after_rollback_clears_head(session):
game, deep, _, cp1, cp2 = _scene_with_two_checkpoints(session)
crud.rollback_to_checkpoint(session, game, deep, cp1.id)
session.refresh(game)
assert game.rollback_head_checkpoint_id == cp1.id
# Rolling forward to the latest checkpoint redoes the rollback and returns to live play.
ok, msg = crud.rollback_to_checkpoint(session, game, deep, cp2.id)
assert ok, msg
session.refresh(game)
assert game.rollback_head_checkpoint_id is None
def test_action_after_rollback_truncates_and_bumps(session):
game, deep, _, cp1, cp2 = _scene_with_two_checkpoints(session)
v0 = game.rollback_timeline_version
crud.rollback_to_checkpoint(session, game, deep, cp1.id)
# A new action from the rolled-back state commits the new timeline: maintain_rollback_history
# discards the abandoned future, clears the head, bumps the version, and captures anew.
crud.add_game_event(session, game.id, "New branch", kind="info")
crud.maintain_rollback_history(session, game)
session.refresh(game)
events = _events(session, game)
msgs = {e.message for e in events}
assert "Second" not in msgs # abandoned future discarded
assert {"First", "New branch"} <= msgs
# cp1 + the freshly captured checkpoint (note: SQLite may reuse cp2's freed id, which
# is safe — every event that referenced it was deleted in the same step).
assert len(_checkpoints(session, game)) == 2
new_event = next(e for e in events if e.message == "New branch")
assert new_event.checkpoint_id and new_event.checkpoint_id > 0
assert game.rollback_head_checkpoint_id is None
assert game.rollback_timeline_version == v0 + 1
def test_rollback_permissions(session):
game, deep, (p2, p3) = make_scene_game(session, num_pirats=2)
crud.seal_and_purge(session, game)
crud.add_game_event(session, game.id, "Floor", kind="info")
cp = crud.capture_checkpoint(session, game)
# A plain Pi-Rat (not admin, not Deep) cannot roll back.
assert not p2.is_admin and p2.role == "pirat"
ok, msg = crud.rollback_to_checkpoint(session, game, p2, cp.id)
assert not ok
# A Deep player who is not an admin can.
p3.role = "deep"
p3.is_admin = False
session.add(p3)
session.commit()
ok, msg = crud.rollback_to_checkpoint(session, game, p3, cp.id)
assert ok, msg
# Not during a scene: rejected.
game.phase = "between_scenes"
session.add(game)
session.commit()
ok, msg = crud.rollback_to_checkpoint(session, game, deep, cp.id)
assert not ok
def test_scene_end_seals_and_purges(session):
game, deep, (p2,) = make_scene_game(session)
crud.seal_and_purge(session, game)
crud.add_game_event(session, game.id, "Mid-scene", kind="info")
cp = crud.capture_checkpoint(session, game)
crud.add_game_event(session, game.id, "Later", kind="info")
crud.capture_checkpoint(session, game) # a later checkpoint so the rollback sets a head
crud.rollback_to_checkpoint(session, game, deep, cp.id) # leave with a head set
session.refresh(game)
assert game.rollback_head_checkpoint_id == cp.id
assert len(_checkpoints(session, game)) == 2
# Leaving the scene runs the non-scene branch: history is purged and sealed.
game.phase = "between_scenes"
session.add(game)
session.commit()
crud.maintain_rollback_history(session, game)
session.refresh(game)
assert _checkpoints(session, game) == []
assert all(e.checkpoint_id == 0 for e in _events(session, game))
assert game.rollback_head_checkpoint_id is None
# The now-stale checkpoint id is rejected (back in a scene, even).
game.phase = "scene"
session.add(game)
session.commit()
ok, msg = crud.rollback_to_checkpoint(session, game, deep, cp.id)
assert not ok
def test_rollback_middleware_capture_and_route_end_to_end():
"""Exercises the real middleware capture path: a normal scene mutation over HTTP
must create a checkpoint, and the rollback endpoint must restore the prior state.
The middleware uses pirats.main.engine directly, so point it at the test engine."""
from fastapi.testclient import TestClient
from sqlalchemy.pool import StaticPool
import pirats.main as main_module
from pirats.database import get_session
engine = create_engine(
"sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool,
)
SQLModel.metadata.create_all(engine)
session = Session(engine)
game, deep, (p2,) = make_scene_game(session)
obs = game.obstacles[0]
p2.hand_cards = json.dumps(["10D", "2H", "JS", "3C"])
obs.current_value = 5
session.add_all([p2, obs])
session.commit()
crud.create_challenge(session, game.id, deep.id, p2.id, [obs.id], stakes_failure="x")
# A floor checkpoint representing the pre-play state.
floor = crud.capture_checkpoint(session, game)
def get_session_override():
yield session
app = main_module.app
app.dependency_overrides[get_session] = get_session_override
saved_engine = main_module.engine
main_module.engine = engine # so the capture middleware writes to the test DB
client = TestClient(app)
try:
# Normal scene mutation over HTTP -> the middleware should capture a checkpoint.
r = client.post(f"/api/game/{game.id}/player/{p2.id}/play-card",
data={"obstacle_id": obs.id, "card_code": "10D"})
assert r.status_code == 200
session.expire_all()
assert session.get(Obstacle, obs.id).current_value == 10
assert len(_checkpoints(session, game)) == 2 # floor + the one the middleware took
# Roll back to the floor over HTTP: state reverts, but the later checkpoint is
# kept (greyed/undoable) and the head points at the floor.
r = client.post(f"/api/game/{game.id}/player/{deep.id}/rollback",
data={"checkpoint_id": floor.id})
assert r.status_code == 200
session.expire_all()
assert session.get(Obstacle, obs.id).current_value == 5
assert len(_checkpoints(session, game)) == 2
session.refresh(game)
assert game.rollback_head_checkpoint_id == floor.id
v_before = game.rollback_timeline_version
# A new action from the rolled-back state (the floor restored the open challenge
# and the played card to p2's hand): the middleware truncates the abandoned future,
# clears the head, and bumps the version.
r = client.post(f"/api/game/{game.id}/player/{p2.id}/play-card",
data={"obstacle_id": obs.id, "card_code": "10D"})
assert r.status_code == 200
session.expire_all()
session.refresh(game)
assert session.get(Obstacle, obs.id).current_value == 10
assert game.rollback_head_checkpoint_id is None
assert game.rollback_timeline_version == v_before + 1
assert len(_checkpoints(session, game)) == 2 # floor + the new branch's checkpoint
finally:
main_module.engine = saved_engine
app.dependency_overrides.clear()
def test_rollback_route_enforces_permission():
from fastapi.testclient import TestClient
from sqlalchemy.pool import StaticPool
from pirats.main import app
from pirats.database import get_session
engine = create_engine(
"sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool,
)
SQLModel.metadata.create_all(engine)
session = Session(engine)
game, deep, pirats = make_scene_game(session, num_pirats=2)
p2 = pirats[0]
crud.seal_and_purge(session, game)
crud.add_game_event(session, game.id, "Floor", kind="info")
cp = crud.capture_checkpoint(session, game)
def get_session_override():
yield session
app.dependency_overrides[get_session] = get_session_override
client = TestClient(app)
try:
# A plain Pi-Rat is rejected by the endpoint.
r = client.post(f"/api/game/{game.id}/player/{p2.id}/rollback",
data={"checkpoint_id": cp.id})
assert r.status_code == 400
# The Deep/admin succeeds.
r = client.post(f"/api/game/{game.id}/player/{deep.id}/rollback",
data={"checkpoint_id": cp.id})
assert r.status_code == 200
assert r.json()["status"] == "ok"
finally:
app.dependency_overrides.clear()
def _seed_game(session, crew_name, phase, last_event_ts):
"""A game whose most-recent GameEvent is at `last_event_ts`."""
game = Game(crew_name=crew_name, phase=phase)
session.add(game)
session.commit()
session.refresh(game)
session.add(GameEvent(game_id=game.id, message="move", timestamp=last_event_ts - 10))
session.add(GameEvent(game_id=game.id, message="move", timestamp=last_event_ts))
session.commit()
return game
def test_purge_old_games(session):
from pirats import maintenance
now = 2_000_000_000.0
DAY = 86400.0
active = _seed_game(session, "Active", "scene", now - 2 * DAY)
finished_old = _seed_game(session, "DoneOld", "ended", now - 20 * DAY)
finished_recent = _seed_game(session, "DoneNew", "ended", now - 3 * DAY)
inactive = _seed_game(session, "Ghosted", "scene", now - 40 * DAY)
# A game with no events at all is undatable and must be left alone.
empty = Game(crew_name="Empty", phase="lobby")
session.add(empty)
session.commit()
session.refresh(empty)
# Give the finished-old game a player + checkpoint to prove children are cleared.
session.add(Player(game_id=finished_old.id, name="Crewmate"))
session.add(Checkpoint(game_id=finished_old.id, state_json="{}"))
session.commit()
purged = maintenance.purge_old_games(session, finished_days=14, inactive_days=30, now=now)
purged_ids = {p["id"] for p in purged}
assert finished_old.id in purged_ids # ended > 14 days ago
assert inactive.id in purged_ids # no activity for > 30 days
assert active.id not in purged_ids # recently active
assert finished_recent.id not in purged_ids # ended only 3 days ago
assert empty.id not in purged_ids # undatable
remaining = {g.id for g in session.exec(select(Game)).all()}
assert finished_old.id not in remaining
assert inactive.id not in remaining
assert active.id in remaining
assert empty.id in remaining
# Children of a purged game are gone.
assert session.exec(select(GameEvent).where(GameEvent.game_id == finished_old.id)).first() is None
assert session.exec(select(Player).where(Player.game_id == finished_old.id)).first() is None
assert session.exec(select(Checkpoint).where(Checkpoint.game_id == finished_old.id)).first() is None
# Summaries carry the reason and a positive byte estimate.
summary = next(p for p in purged if p["id"] == finished_old.id)
assert summary["reason"] == "finished"
assert summary["crew_name"] == "DoneOld"
assert summary["estimated_bytes"] > 0
def test_sanitize_text_and_name():
from pirats.validation import (
sanitize_text, sanitize_name, MAX_TEXT_LENGTH, MAX_NAME_LENGTH,
)
# Control characters dropped, surrounding whitespace trimmed.
assert sanitize_text(" hi\x00\x07there ") == "hithere"
# Tabs/newlines kept inside the text.
assert sanitize_text("a\nb\tc") == "a\nb\tc"
# Zero-width joiner (emoji sequences) survives.
assert sanitize_text("a‍b") == "a‍b"
# Lone surrogate removed without raising.
assert sanitize_text("a" + chr(0xD800) + "b") == "ab"
# Length capped.
assert len(sanitize_text("x" * (MAX_TEXT_LENGTH + 50))) == MAX_TEXT_LENGTH
# Empty / falsy in -> empty out.
assert sanitize_text("") == ""
# Names collapse internal whitespace and use the tighter cap.
assert sanitize_name(" Captain Redbeard\n ") == "Captain Redbeard"
assert len(sanitize_name("y" * (MAX_NAME_LENGTH + 10))) == MAX_NAME_LENGTH
def test_create_and_join_sanitize_names():
from fastapi.testclient import TestClient
from sqlalchemy.pool import StaticPool
from pirats.main import app
from pirats.database import get_session
from sqlmodel import SQLModel, create_engine, Session
engine = create_engine(
"sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool,
)
SQLModel.metadata.create_all(engine)
session = Session(engine)
def override():
yield session
app.dependency_overrides[get_session] = override
client = TestClient(app)
try:
r = client.post("/api/game", data={"crew_name": " The\x00 Salty\x07 Dogs "})
gid = r.json()["id"]
assert session.get(Game, gid).crew_name == "The Salty Dogs"
r = client.post(f"/api/game/{gid}/join", data={"name": " Bad\x01name "})
player = session.get(Player, r.json()["id"])
assert player.player_name == "Badname"
assert "\x01" not in player.name
finally:
app.dependency_overrides.clear()
def test_request_body_size_limit_middleware():
from fastapi import FastAPI, Request
from fastapi.testclient import TestClient
from pirats.main import LimitRequestBodyMiddleware
app = FastAPI()
app.add_middleware(LimitRequestBodyMiddleware, max_bytes=1000)
@app.post("/echo")
async def echo(request: Request):
body = await request.body()
return {"len": len(body)}
client = TestClient(app)
# Under the limit: passes through and the handler sees the full body.
r = client.post("/echo", content=b"x" * 500)
assert r.status_code == 200
assert r.json()["len"] == 500
# Over the limit with a declared Content-Length: rejected up front.
r = client.post("/echo", content=b"x" * 5000)
assert r.status_code == 413
# Over the limit via a streamed (length-omitting) body: rejected once the
# counted bytes exceed the cap.
def oversized_chunks():
for _ in range(10):
yield b"x" * 500 # 5000 bytes total
r = client.post("/echo", content=oversized_chunks())
assert r.status_code == 413
def test_request_body_size_limit_on_real_app():
from fastapi.testclient import TestClient
from pirats.main import app, DEFAULT_MAX_BODY_BYTES
# No DB override needed: the middleware rejects before routing/dependencies.
client = TestClient(app)
r = client.post("/api/game", content=b"x" * (DEFAULT_MAX_BODY_BYTES + 1))
assert r.status_code == 413
Reference in New Issue View Git Blame Copy Permalink