import pytest import json from sqlmodel import SQLModel, create_engine, Session from pirats import cards from pirats import crud from pirats.models import Obstacle # 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="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 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 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 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()