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 # Completing Crew Objective 1 (Steal a Ship) makes the highest-ranked Pi-Rat the Captain crud.toggle_objective(session, game.id, p1.id, "crew_1", True) session.refresh(game) assert game.captain_player_id in (p2.id, p3.id) captain = crud.get_player(session, game.captain_player_id) assert crud.is_player_captain(captain, game) # Captain privilege: +1 hand size assert crud.calculate_max_hand_size(captain, game.players, game.captain_player_id) == 5 # Losing the ship vacates the Captaincy crud.toggle_objective(session, game.id, p1.id, "crew_1", False) session.refresh(game) assert game.captain_player_id is None 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()