mamamiyear
c0bc5384f4
- 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()
112 lines
3.5 KiB
Python
112 lines
3.5 KiB
Python
import unittest
|
|
from threading import Event, Thread
|
|
|
|
from texas_holdem.agents import PokerAgent
|
|
from texas_holdem.models import Observation, PlayerAction
|
|
from texas_holdem.service import GameManager
|
|
|
|
|
|
class BlockingAgent(PokerAgent):
|
|
def __init__(self, entered: Event, release: Event) -> None:
|
|
self.entered = entered
|
|
self.release = release
|
|
|
|
def decide(self, observation: Observation) -> PlayerAction:
|
|
self.entered.set()
|
|
if not self.release.wait(timeout=5):
|
|
raise RuntimeError("test timed out waiting to release blocking agent")
|
|
for action in observation.legal_actions:
|
|
if action["action"] == "check":
|
|
return PlayerAction("check")
|
|
return PlayerAction("call")
|
|
|
|
|
|
class ServiceTests(unittest.TestCase):
|
|
def test_create_and_run_game(self) -> None:
|
|
manager = GameManager()
|
|
game = manager.create_game(
|
|
{
|
|
"game_id": "demo",
|
|
"seed": 11,
|
|
"starting_stack": 200,
|
|
"small_blind": 5,
|
|
"big_blind": 10,
|
|
"players": [
|
|
{"id": "a", "type": "calling"},
|
|
{"id": "b", "type": "calling"},
|
|
],
|
|
}
|
|
)
|
|
|
|
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(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()
|
|
entered = Event()
|
|
release = Event()
|
|
game = manager.create_game(
|
|
{
|
|
"game_id": "blocking",
|
|
"seed": 13,
|
|
"starting_stack": 200,
|
|
"small_blind": 5,
|
|
"big_blind": 10,
|
|
"players": [
|
|
{"id": "a", "type": "calling"},
|
|
{"id": "b", "type": "calling"},
|
|
],
|
|
}
|
|
)
|
|
manager.run_hands("blocking", count=1)
|
|
game.agents["a"] = BlockingAgent(entered, release)
|
|
|
|
thread = Thread(target=lambda: manager.run_hands("blocking", count=1))
|
|
thread.start()
|
|
self.assertTrue(entered.wait(timeout=2))
|
|
|
|
state = manager.get_game_state("blocking")
|
|
|
|
release.set()
|
|
thread.join(timeout=2)
|
|
self.assertFalse(thread.is_alive())
|
|
self.assertEqual(state["hand_number"], 1)
|
|
self.assertEqual(len(state["hands"]), 1)
|
|
|
|
def test_duplicate_http_agent_endpoint_is_rejected_across_active_games(self) -> None:
|
|
manager = GameManager()
|
|
payload = {
|
|
"starting_stack": 200,
|
|
"small_blind": 5,
|
|
"big_blind": 10,
|
|
"players": [
|
|
{
|
|
"id": "ai",
|
|
"agent": {
|
|
"type": "http",
|
|
"endpoint": "http://127.0.0.1:9101/act",
|
|
},
|
|
},
|
|
{"id": "b", "type": "calling"},
|
|
],
|
|
}
|
|
|
|
manager.create_game({"game_id": "g1", **payload})
|
|
|
|
with self.assertRaisesRegex(ValueError, "already belongs to game g1"):
|
|
manager.create_game({"game_id": "g2", **payload})
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|