From c0bc5384f4f8650e845ca9801141c99bade45898 Mon Sep 17 00:00:00 2001 From: Qian Rui Date: Sat, 23 May 2026 22:11:45 +0800 Subject: [PATCH] feat: add hand detail API and enrich hand summary fields - HandSummary: add hole_cards, starting_stacks, ending_stacks, pot_contributions - Engine: capture all players' hole cards (not just showdown), pre/post hand stacks, per-level pot contributions - Server: new GET /game//hands/ route - Service: add get_hand_state() method - Tests: add ServerTests for new endpoint, update existing tests - Existing GET /game/ auto-inherits new fields via shared to_dict() --- tests/test_engine.py | 40 ++++++++++++++++++++++ tests/test_server.py | 76 +++++++++++++++++++++++++++++++++++++++++ tests/test_service.py | 10 +++++- texas_holdem/engine.py | 61 ++++++++++++++++++++++++++++++--- texas_holdem/models.py | 37 ++++++++++++++++++++ texas_holdem/server.py | 8 +++++ texas_holdem/service.py | 7 ++++ 7 files changed, 233 insertions(+), 6 deletions(-) create mode 100644 tests/test_server.py diff --git a/tests/test_engine.py b/tests/test_engine.py index 4641532..4b7fcaa 100644 --- a/tests/test_engine.py +++ b/tests/test_engine.py @@ -100,10 +100,50 @@ class EngineTests(unittest.TestCase): awards = game._award_pots() self.assertEqual([award.amount for award in awards], [150, 100]) + self.assertEqual( + [contribution["amount"] for contribution in game._last_pot_contributions], + [150, 100], + ) + self.assertEqual( + game._last_pot_contributions[0]["contributors"], + {"p1": 50, "p2": 50, "p3": 50}, + ) + self.assertEqual( + game._last_pot_contributions[1]["contributors"], + {"p2": 50, "p3": 50}, + ) self.assertEqual(game.players[0].stack, 150) self.assertEqual(game.players[1].stack, 100) self.assertEqual(game.players[2].stack, 0) + def test_hand_summary_includes_full_hand_snapshots(self) -> None: + players = [ + ("p1", "Player 1", CallingStationAgent()), + ("p2", "Player 2", CallingStationAgent()), + ("p3", "Player 3", CallingStationAgent()), + ] + game = TableGame("g5", players, starting_stack=100, small_blind=5, big_blind=10, rng=Random(23)) + + summary = game.run_hand() + payload = summary.to_dict() + + self.assertEqual(set(summary.hole_cards), {"p1", "p2", "p3"}) + self.assertTrue(all(len(cards) == 2 for cards in summary.hole_cards.values())) + self.assertEqual(summary.starting_stacks, {"p1": 100, "p2": 100, "p3": 100}) + self.assertEqual(set(summary.ending_stacks), {"p1", "p2", "p3"}) + self.assertEqual(sum(summary.starting_stacks.values()), sum(summary.ending_stacks.values())) + self.assertGreaterEqual(len(summary.pot_contributions), 1) + self.assertTrue( + all( + contribution["amount"] == sum(contribution["contributors"].values()) + for contribution in summary.pot_contributions + ) + ) + self.assertEqual(set(payload["hole_cards"]), {"p1", "p2", "p3"}) + self.assertEqual(payload["starting_stacks"], {"p1": 100, "p2": 100, "p3": 100}) + self.assertIn("ending_stacks", payload) + self.assertIn("pot_contributions", payload) + def test_short_all_in_does_not_reopen_raising_to_prior_actor(self) -> None: seen: list[tuple[str, str, list[str]]] = [] players = [ diff --git a/tests/test_server.py b/tests/test_server.py new file mode 100644 index 0000000..367a898 --- /dev/null +++ b/tests/test_server.py @@ -0,0 +1,76 @@ +import json +import unittest +from threading import Thread +from urllib.request import Request, urlopen + +from texas_holdem import server as poker_server +from texas_holdem.service import GameManager + + +class ServerTests(unittest.TestCase): + def setUp(self) -> None: + self.previous_manager = poker_server.MANAGER + poker_server.MANAGER = GameManager() + self.server = poker_server.create_server("127.0.0.1", 0) + self.thread = Thread(target=self.server.serve_forever, daemon=True) + self.thread.start() + host, port = self.server.server_address + self.base_url = f"http://{host}:{port}" + + def tearDown(self) -> None: + self.server.shutdown() + self.server.server_close() + self.thread.join(timeout=2) + poker_server.MANAGER = self.previous_manager + + def request_json( + self, + method: str, + path: str, + payload: dict[str, object] | None = None, + ) -> dict[str, object]: + data = None + headers = {} + if payload is not None: + data = json.dumps(payload).encode("utf-8") + headers["Content-Type"] = "application/json" + request = Request( + f"{self.base_url}{path}", + data=data, + headers=headers, + method=method, + ) + with urlopen(request, timeout=5) as response: + return json.loads(response.read().decode("utf-8")) + + def test_get_hand_route_returns_expanded_hand_summary(self) -> None: + self.request_json( + "POST", + "/game", + { + "game_id": "route-demo", + "seed": 17, + "starting_stack": 200, + "small_blind": 5, + "big_blind": 10, + "players": [ + {"id": "a", "type": "calling"}, + {"id": "b", "type": "calling"}, + ], + }, + ) + self.request_json("POST", "/game/route-demo/hands", {"count": 1}) + + hand = self.request_json("GET", "/game/route-demo/hands/1") + game = self.request_json("GET", "/game/route-demo") + + self.assertEqual(hand["hand_number"], 1) + self.assertEqual(set(hand["hole_cards"]), {"a", "b"}) + self.assertEqual(hand["starting_stacks"], {"a": 200, "b": 200}) + self.assertIn("ending_stacks", hand) + self.assertIn("pot_contributions", hand) + self.assertEqual(game["hands"][0], hand) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_service.py b/tests/test_service.py index 1b5f60a..b632cf7 100644 --- a/tests/test_service.py +++ b/tests/test_service.py @@ -40,8 +40,16 @@ class ServiceTests(unittest.TestCase): hands = manager.run_hands(game.game_id, count=1) + state = manager.get_game_state("demo") + hand = manager.get_hand_state("demo", 1) + self.assertEqual(len(hands), 1) - self.assertEqual(manager.get_game_state("demo")["hand_number"], 1) + self.assertEqual(state["hand_number"], 1) + self.assertEqual(hand, state["hands"][0]) + self.assertIn("hole_cards", hand) + self.assertIn("starting_stacks", hand) + self.assertIn("ending_stacks", hand) + self.assertIn("pot_contributions", hand) def test_get_game_state_does_not_block_during_run(self) -> None: manager = GameManager() diff --git a/texas_holdem/engine.py b/texas_holdem/engine.py index ba4a16a..4db4143 100644 --- a/texas_holdem/engine.py +++ b/texas_holdem/engine.py @@ -62,6 +62,7 @@ class TableGame: self.board = [] self.action_history: list[ActionRecord] = [] self.hand_summaries: list[HandSummary] = [] + self._last_pot_contributions: list[dict[str, object]] = [] # ``blind_history`` is an append-only log of every blind level change # (including the initial one). Each entry's ``hand_number`` is the # first hand that played under those stakes, which makes it trivial @@ -116,6 +117,11 @@ class TableGame: for player in self.players: player.reset_for_hand() + starting_stacks = { + player.player_id: player.stack + for player in self.players + if player.in_hand + } self._advance_button() assert self.button_index is not None @@ -127,6 +133,11 @@ class TableGame: self._broadcast_game_update() self._deal_hole_cards(deck) + hole_cards = { + player.player_id: list(player.hole_cards) + for player in self.players + if player.in_hand + } small_blind_index, big_blind_index = self._blind_indexes() self._post_blind(small_blind_index, "small_blind", self.small_blind) self._post_blind(big_blind_index, "big_blind", self.big_blind) @@ -150,6 +161,11 @@ class TableGame: self._betting_round(street, start_index, self.big_blind) awards = self._award_pots() + ending_stacks = { + player.player_id: player.stack + for player in self.players + if player.player_id in starting_stacks + } summary = HandSummary( game_id=self.game_id, hand_number=self.hand_number, @@ -158,6 +174,10 @@ class TableGame: actions=list(self.action_history), awards=awards, blinds=active_blinds, + hole_cards=hole_cards, + starting_stacks=starting_stacks, + ending_stacks=ending_stacks, + pot_contributions=deepcopy(self._last_pot_contributions), showdown_hands=self._collect_showdown_hands(), started_at=started_at, finished_at=time(), @@ -557,21 +577,43 @@ class TableGame: return current_bet, min_raise, full_raise def _award_pots(self) -> list[PotAward]: + self._last_pot_contributions = [] total_pot = sum(player.total_bet for player in self.players) live_players = [player for player in self.players if self._is_live(player)] if not live_players or total_pot <= 0: return [] - if len(live_players) == 1: - live_players[0].stack += total_pot - return [PotAward(total_pot, [live_players[0].player_id], None)] - levels = sorted({player.total_bet for player in self.players if player.total_bet > 0}) + if len(live_players) == 1: + winner = live_players[0] + winner.stack += total_pot + previous_level = 0 + for level in levels: + contributors = [player for player in self.players if player.total_bet >= level] + pot_amount = (level - previous_level) * len(contributors) + self._last_pot_contributions.append( + { + "amount": pot_amount, + "contributors": { + player.player_id: level - previous_level + for player in contributors + }, + "winners": [winner.player_id], + "hand_value": None, + } + ) + previous_level = level + return [PotAward(total_pot, [winner.player_id], None)] + previous_level = 0 awards: list[PotAward] = [] for level in levels: contributors = [player for player in self.players if player.total_bet >= level] pot_amount = (level - previous_level) * len(contributors) + level_contributions = { + player.player_id: level - previous_level + for player in contributors + } previous_level = level contenders = [player for player in contributors if self._is_live(player)] if not contenders or pot_amount <= 0: @@ -593,13 +635,22 @@ class TableGame: winner.stack += share for winner in ordered_winners[:remainder]: winner.stack += 1 + winner_ids = [winner.player_id for winner in ordered_winners] awards.append( PotAward( amount=pot_amount, - winners=[winner.player_id for winner in ordered_winners], + winners=winner_ids, hand_value=best_value, ) ) + self._last_pot_contributions.append( + { + "amount": pot_amount, + "contributors": level_contributions, + "winners": winner_ids, + "hand_value": best_value, + } + ) return awards def _collect_showdown_hands(self) -> dict[str, list]: diff --git a/texas_holdem/models.py b/texas_holdem/models.py index aceaba8..1455a4d 100644 --- a/texas_holdem/models.py +++ b/texas_holdem/models.py @@ -178,6 +178,10 @@ class HandSummary: # on the summary (rather than only on the game) guarantees historical # hands remain self-describing even after the blinds are raised later. blinds: BlindLevel | None = None + hole_cards: dict[str, list[Card]] = field(default_factory=dict) + starting_stacks: dict[str, int] = field(default_factory=dict) + ending_stacks: dict[str, int] = field(default_factory=dict) + pot_contributions: list[dict[str, Any]] = field(default_factory=list) showdown_hands: dict[str, list[Card]] = field(default_factory=dict) started_at: float = field(default_factory=time) finished_at: float = field(default_factory=time) @@ -191,6 +195,16 @@ class HandSummary: "board": [str(card) for card in self.board], "actions": [record.to_dict() for record in self.actions], "awards": [award.to_dict() for award in self.awards], + "hole_cards": { + player_id: [str(card) for card in cards] + for player_id, cards in self.hole_cards.items() + }, + "starting_stacks": dict(self.starting_stacks), + "ending_stacks": dict(self.ending_stacks), + "pot_contributions": [ + self._pot_contribution_to_dict(contribution) + for contribution in self.pot_contributions + ], # ``showdown_hands`` is only populated when more than one player # remained eligible for a pot; empty dict means the hand ended # without a showdown (e.g. everyone folded but the winner). @@ -201,3 +215,26 @@ class HandSummary: "started_at": self.started_at, "finished_at": self.finished_at, } + + @staticmethod + def _pot_contribution_to_dict(contribution: dict[str, Any]) -> dict[str, object]: + hand_value = contribution.get("hand_value") + if isinstance(hand_value, HandValue): + hand_value = hand_value.to_dict() + elif isinstance(hand_value, dict): + hand_value = dict(hand_value) + + raw_contributors = contribution.get("contributors") or {} + contributors = { + str(player_id): int(amount) + for player_id, amount in dict(raw_contributors).items() + } + return { + "amount": int(contribution.get("amount") or 0), + "contributors": contributors, + "winners": [ + str(player_id) + for player_id in contribution.get("winners", []) + ], + "hand_value": hand_value, + } diff --git a/texas_holdem/server.py b/texas_holdem/server.py index 2e8c405..c025642 100644 --- a/texas_holdem/server.py +++ b/texas_holdem/server.py @@ -28,6 +28,14 @@ class PokerRequestHandler(BaseHTTPRequestHandler): if len(path) == 2 and path[0] in {"game", "games"}: self._json(MANAGER.get_game_state(path[1])) return + if len(path) == 4 and path[0] in {"game", "games"} and path[2] == "hands": + try: + hand_number = int(path[3]) + except ValueError: + self._json({"error": "not found"}, HTTPStatus.NOT_FOUND) + return + self._json(MANAGER.get_hand_state(path[1], hand_number)) + return self._json({"error": "not found"}, HTTPStatus.NOT_FOUND) except KeyError as exc: self._json({"error": str(exc)}, HTTPStatus.NOT_FOUND) diff --git a/texas_holdem/service.py b/texas_holdem/service.py index e8c2acd..bcb2b29 100644 --- a/texas_holdem/service.py +++ b/texas_holdem/service.py @@ -78,6 +78,13 @@ class GameManager: def get_game_state(self, game_id: str) -> dict[str, object]: return self.get_game(game_id).snapshot_completed() + def get_hand_state(self, game_id: str, hand_number: int) -> dict[str, object]: + state = self.get_game_state(game_id) + for hand in state.get("hands", []): + if hand.get("hand_number") == hand_number: + return hand + raise KeyError(f"hand not found: {game_id} #{hand_number}") + def list_games(self) -> list[dict[str, object]]: with self._lock: games = list(self._games.values())