Tim McCarthy
ceaf88ef22
Completing Crew Objective 1 (Steal a Ship) no longer anoints the highest-ranked Pi-Rat as Captain (nor vacates the seat when the ship is lost). Captaincy is now earned: the crew chooses one via the Captaincy UI / set_captain and Crew Objective 2. Decoupled captaincy from crew_1 in toggle_objective, dropped the now-unused random import, and updated test_captaincy_and_hand_sizes accordingly. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2333 lines
85 KiB
Python
2333 lines
85 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 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
|
|
assert not crud.is_player_captain(p2, game)
|
|
|
|
# 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
|
|
assert crud.is_player_captain(p2, game)
|
|
# 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_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_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()
|