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/<game_id>/hands/<hand_number> route - Service: add get_hand_state() method - Tests: add ServerTests for new endpoint, update existing tests - Existing GET /game/<game_id> auto-inherits new fields via shared to_dict()
This commit is contained in:
@@ -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 = [
|
||||
|
||||
@@ -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()
|
||||
@@ -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()
|
||||
|
||||
+56
-5
@@ -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]:
|
||||
|
||||
@@ -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,
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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())
|
||||
|
||||
Reference in New Issue
Block a user