Auto-play a random legal card when the turn timeout expires

server/tests/test_engine.py

@@ -1,12 +1,14 @@
 """Rule engine tests: captures, scope, scoring and full-match simulation."""
 from __future__ import annotations
 
+import random
 import unittest
 
 from scopa.game import engine
 from scopa.game.errors import (
     CardNotInHand,
     GameFinished,
+    GameNotStarted,
     IllegalMove,
     NotYourTurn,
 )
@@ -390,5 +392,59 @@ class HandEndAckTest(unittest.TestCase):
         self.assertIn("award", view["last_hand"])
 
 
+class AutoPlayTest(unittest.TestCase):
+    def test_auto_play_plays_a_card_and_advances_turn(self) -> None:
+        state = make_state([["02D", "03C"], ["04D"], ["05D"], ["06D"]],
+                           table=["09B"])
+        state.turn_deadline = "2000-01-01T00:00:00+00:00"
+        engine.auto_play(state, random.Random(7))
+        self.assertEqual(1, state.turn)
+        self.assertEqual(1, len(state.players[0].hand))
+        # The played card could not capture the nine, so the table grew.
+        self.assertEqual(2, len(state.table))
+        self.assertIsNotNone(state.last_move)
+        assert state.last_move is not None
+        self.assertEqual(0, state.last_move.seat)
+        self.assertNotEqual("2000-01-01T00:00:00+00:00", state.turn_deadline)
+
+    def test_auto_play_takes_a_mandatory_capture(self) -> None:
+        # p0 holds only the five of denari, which must capture the equal
+        # five of coppe instead of the unrelated nine on the table.
+        state = make_state([["05D"], ["04D"], ["05D"], ["06D"]],
+                           table=["05C", "09B"])
+        engine.auto_play(state)
+        self.assertIsNotNone(state.last_move)
+        assert state.last_move is not None
+        self.assertEqual("05D", state.last_move.card)
+        self.assertEqual(["05C"], state.last_move.captured)
+        self.assertEqual(["09B"], [c.code for c in state.table])
+        self.assertEqual(["05C", "05D"],
+                         [c.code for c in state.players[0].captured])
+
+    def test_auto_play_can_end_the_hand_and_clears_deadline(self) -> None:
+        state = make_state([["02D"], [], [], []], table=["02C"])
+        state.turn_deadline = "2000-01-01T00:00:00+00:00"
+        engine.auto_play(state)
+        self.assertEqual("hand_end", state.phase)
+        self.assertIsNone(state.turn_deadline)
+        self.assertTrue(state.hand_end_deadline)
+
+    def test_auto_play_requires_playing_phase(self) -> None:
+        state = make_state([["02D"], ["04D"], ["05D"], ["06D"]], table=[])
+        state.phase = "hand_end"
+        with self.assertRaises(GameNotStarted):
+            engine.auto_play(state)
+
+    def test_create_game_copies_turn_timeout_and_arms_deadline(self) -> None:
+        state = engine.create_game("g", "CODE98", "p0", "p0", turn_timeout=7)
+        self.assertEqual(7, state.turn_timeout)
+        for i in range(1, 4):
+            engine.join_game(state, f"p{i}", f"p{i}")
+        self.assertEqual(PHASE_PLAYING, state.phase)
+        self.assertTrue(state.turn_deadline)
+        view = engine.state_for_player(state, "p0")
+        self.assertTrue(view["turn_deadline"])
+
+
 if __name__ == "__main__":
     unittest.main()

server/tests/test_websocket.py

@@ -238,5 +238,50 @@ class HandEndWebSocketTest(unittest.TestCase):
                     self.assertEqual(2, update["game"]["hand_number"])
 
 
+class TurnTimeoutWebSocketTest(unittest.TestCase):
+    @async_test
+    async def test_turn_timeout_auto_plays_a_card(self) -> None:
+        state = engine.create_game(
+            "turn-timeout-1", "TT0001", "alice", "Alice",
+            target_score=11, turn_timeout=1,
+        )
+        for name in PLAYERS[1:]:
+            engine.join_game(state, name, name.capitalize())
+        await game_store.save(state)
+
+        ws_transport = ASGIWebSocketTransport(app=app)
+        async with AsyncClient(transport=ws_transport, base_url="http://testserver") as ws_client:
+            with ws_users([make_user("alice")]):
+                async with aconnect_ws(f"/ws/games/{state.id}", ws_client) as ws:
+                    first = await ws.receive_json()
+                    # Bob (seat 1) is first to act and never connects.
+                    self.assertEqual(1, first["game"]["turn"])
+                    deadline = first["game"]["turn_deadline"]
+                    self.assertIsNotNone(deadline)
+
+                    # Nobody plays: the timer must play a random card for Bob.
+                    update = None
+                    for _ in range(20):
+                        try:
+                            update = await asyncio.wait_for(
+                                ws.receive_json(), timeout=2
+                            )
+                        except asyncio.TimeoutError:
+                            break
+                        if (
+                            update.get("type") == "state"
+                            and update["game"]["turn"] == 2
+                        ):
+                            break
+                    self.assertIsNotNone(update)
+                    assert update is not None
+                    self.assertEqual(2, update["game"]["turn"])
+                    self.assertEqual(1, update["game"]["last_move"]["seat"])
+                    self.assertEqual(
+                        9, update["game"]["players"][1]["cards_left"]
+                    )
+                    self.assertNotEqual(deadline, update["game"]["turn_deadline"])
+
+
 if __name__ == "__main__":
     unittest.main()