Tim McCarthy
ccb32e7103
New maintenance.py runs a background asyncio loop (started from the app lifespan) that purges games which either finished more than PIRATS_PURGE_ FINISHED_DAYS (default 14) ago or have had no activity for PIRATS_PURGE_ INACTIVE_DAYS (default 30). "Activity" is a game's most-recent GameEvent timestamp; games with no events are undatable and left alone. Each purge is logged with the crew name, uuid, reason, idle days and an estimated byte footprint; every run that removes anything VACUUMs the SQLite file and logs the disk space actually reclaimed. Child rows go via the ORM cascade already declared on Game's relationships. Configurable via PIRATS_PURGE_ENABLED / _INTERVAL_HOURS / _FINISHED_DAYS / _INACTIVE_DAYS, exposed as services.pirats.purge.* in the NixOS module and documented in the README. Unit test covers the selection logic and cascade; smoke-tested the VACUUM/reclaim path against a file DB. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2474 lines
91 KiB
Python
2474 lines
91 KiB
Python
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
|