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 should have automatically triggered because all 3 submitted assert game.phase == "swap_techniques" session.refresh(p1) session.refresh(p2) session.refresh(p3) techs_p1 = json.loads(p1.swapped_techniques) assert len(techs_p1) == 3 # Check that Carlos didn't get any of his own techniques assert "Carlos Strike" not in techs_p1 # Now assign to face cards crud.assign_face_card_techniques(session, p1.id, techs_p1[0], techs_p1[1], techs_p1[2]) crud.assign_face_card_techniques(session, p2.id, "T2", "T3", "T4") crud.assign_face_card_techniques(session, p3.id, "T5", "T6", "T7") 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_roll_new_character(session): game = crud.create_game(session) p1 = crud.add_player(session, game.id, "P1") p2 = crud.add_player(session, game.id, "P2") # helper player for delegation target 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 session.add_all([p1, p2]) session.commit() crud.roll_new_character( session, game.id, p1.id, avatar_look="Shiny eyes", avatar_smell="Spiced Rum", first_words="No fear!", good_at_math="Yes" ) session.refresh(p1) session.refresh(game) # Verify properties assert not p1.is_dead assert not p1.is_ghost assert not p1.completed_personal_1 assert not p1.completed_personal_2 assert not p1.completed_personal_3 assert p1.avatar_look == "Shiny eyes" assert p1.avatar_smell == "Spiced Rum" assert p1.first_words == "No fear!" assert p1.good_at_math == "Yes" # New recruits always start at Rank 1 assert p1.rank == 1 # Verify name is "Recruit Spiced Rum" assert p1.name == "Recruit Spiced Rum" # Verify techniques auto-assigned assert p1.tech_jack != "" assert p1.tech_queen != "" assert p1.tech_king != "" # Verify delegated targets exist (p2 is the only other player) assert p1.other_like_from_player_id == p2.id assert p1.other_hate_from_player_id == p2.id # Verify hand was refreshed up to new max hand size hand = json.loads(p1.hand_cards) assert len(hand) > 0 # Old cards returned to deck deck = json.loads(game.deck_cards) assert "2C" in deck or "2C" in hand assert "3D" in deck or "3D" in hand assert "4H" in deck or "4H" in hand 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 test_roll_new_character_smell_name_normalization(session): game = crud.create_game(session) p1 = crud.add_player(session, game.id, "P1") crud.add_player(session, game.id, "P2") p1.is_dead = True session.add(p1) session.commit() crud.roll_new_character( session, game.id, p1.id, avatar_look="Shiny eyes", avatar_smell="suspiciously smelling of ink and sour lemons", first_words="No fear!", good_at_math="Yes" ) session.refresh(p1) assert p1.name == "Recruit Ink and sour lemons" def test_roll_new_character_avoids_technique_collisions(session): game = crud.create_game(session) p1 = crud.add_player(session, game.id, "P1") p2 = crud.add_player(session, game.id, "P2") # Give the surviving player techniques straight from the suggestion pool p2.tech_jack = cards.TECHNIQUE_SUGGESTIONS[0] p2.tech_queen = cards.TECHNIQUE_SUGGESTIONS[1] p2.tech_king = cards.TECHNIQUE_SUGGESTIONS[2] p1.is_dead = True session.add_all([p1, p2]) session.commit() # Shrink the pool so a collision would be guaranteed without the exclusion logic import pirats.crud_character # noqa: F401 (module uses cards.TECHNIQUE_SUGGESTIONS) original = cards.TECHNIQUE_SUGGESTIONS cards.TECHNIQUE_SUGGESTIONS = original[:6] try: crud.roll_new_character( session, game.id, p1.id, avatar_look="Shiny eyes", avatar_smell="Spiced Rum", first_words="No fear!", good_at_math="Yes" ) finally: cards.TECHNIQUE_SUGGESTIONS = original session.refresh(p1) recruit_techs = {p1.tech_jack, p1.tech_queen, p1.tech_king} assert recruit_techs == set(original[3:6]) 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_late_joiner_is_dealt_a_hand(session): game, deep, (p2,) = make_scene_game(session) assert game.phase == "scene" newbie = crud.add_player(session, game.id, "Stowaway") assert newbie.role == "pirat" session.refresh(game) expected = crud.calculate_max_hand_size(newbie, game.players, game.captain_player_id) assert expected > 0 assert len(crud.get_player_hand(newbie)) == expected 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"] player_id = client.post(f"/api/game/{game_id}/join", data={"name": "Carlos"}).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()