"""Deadline-queue timeout tests. Timeouts must be driven by the persisted deadlines and the shared queue, not by connected sockets: these tests seed games, queue their deadlines and let the background consumer fire them without a single websocket. """ from __future__ import annotations import asyncio import unittest from datetime import datetime, timedelta, timezone from typing import Optional from pwo import async_test from tavolo import deadlines from tavolo.app import game_store from tavolo.game import engine from tavolo.game.state import GameState, PlayerState PLAYERS = ("alice", "bob", "carol", "dave") def _ms(iso: str) -> int: """Epoch milliseconds for an ISO-8601 timestamp (the queue-entry form).""" return int(datetime.fromisoformat(iso).timestamp() * 1000) def _hand_end_state(game_id: str, deadline: str) -> GameState: """A game paused on the hand-end summary, waiting for acks.""" state = GameState( id=game_id, join_code="DLhend", creator_sub="alice", target_score=11, phase="hand_end", # Long turn timeout: the next hand's auto-play must not interfere # with later tests sharing this store. turn_timeout=3600, ) state.players = [ PlayerState(sub=name, name=name.capitalize(), seat=i) for i, name in enumerate(PLAYERS) ] state.hand_end_deadline = deadline return state async def _wait_for(predicate, timeout: float = 5.0) -> Optional[GameState]: """Poll the store until ``predicate`` holds for the loaded state.""" deadline = asyncio.get_running_loop().time() + timeout while asyncio.get_running_loop().time() < deadline: state = await predicate() if state is not None: return state await asyncio.sleep(0.05) return None class ConnectionIndependenceTest(unittest.TestCase): @async_test async def test_turn_timeout_fires_with_no_connections(self) -> None: state = engine.create_game( "dl-turn-1", "DLT001", "alice", "Alice", target_score=11, turn_timeout=1, ) for name in PLAYERS[1:]: engine.join_game(state, name, name.capitalize()) assert state.turn_deadline is not None await game_store.save(state) await deadlines.sync_deadline(game_store, state) # Nobody ever connects: the consumer must still auto-play for Bob # (seat 1, first to act). result = await _wait_for( lambda: _turn_is(state.id, 2), ) self.assertIsNotNone(result, "turn deadline never fired") assert result is not None self.assertEqual(1, result.last_move.seat if result.last_move else None) # Defuse the follow-on turn deadlines so this game cannot keep # auto-playing while later tests run. result.turn_timeout = 3600 await game_store.save(result) @async_test async def test_hand_end_timeout_fires_with_no_connections(self) -> None: deadline = (datetime.now(timezone.utc) + timedelta(seconds=1)).isoformat() state = _hand_end_state("dl-handend-1", deadline) await game_store.save(state) await deadlines.sync_deadline(game_store, state) # Nobody acks (nobody is even connected): the deadline must deal # the next hand. result = await _wait_for( lambda: _phase_is("dl-handend-1", "playing"), ) self.assertIsNotNone(result, "hand-end deadline never fired") assert result is not None self.assertEqual(2, result.hand_number) self.assertEqual([], result.acked) async def _turn_is(game_id: str, turn: int) -> Optional[GameState]: state = await game_store.load(game_id) return state if state is not None and state.turn == turn else None async def _phase_is(game_id: str, phase: str) -> Optional[GameState]: state = await game_store.load(game_id) return state if state is not None and state.phase == phase else None class ProcessDueTest(unittest.TestCase): """Direct ``process_due`` behaviour: revalidation and idempotency.""" @async_test async def test_processing_twice_is_a_no_op(self) -> None: # Simulates a worker dying after firing but before removing the # entry: another worker re-delivers the same entry. deadline = (datetime.now(timezone.utc) - timedelta(seconds=1)).isoformat() state = _hand_end_state("dl-idem-1", deadline) await game_store.save(state) member = deadlines.encode({ "game_id": state.id, "kind": deadlines.KIND_HAND_END, "hand": state.hand_number, "deadline": _ms(deadline), }) await deadlines.process_due(game_store, member) await deadlines.process_due(game_store, member) result = await game_store.load(state.id) assert result is not None # Advanced exactly once: hand 2, not hand 3. self.assertEqual("playing", result.phase) self.assertEqual(2, result.hand_number) @async_test async def test_stale_entry_is_discarded(self) -> None: # A turn entry enqueued before a play landed in time: the state's # deadline has moved, so the entry must not fire. state = engine.create_game( "dl-stale-1", "DLS001", "alice", "Alice", target_score=11, turn_timeout=3600, ) for name in PLAYERS[1:]: engine.join_game(state, name, name.capitalize()) await game_store.save(state) member = deadlines.encode({ "game_id": state.id, "kind": deadlines.KIND_TURN, "hand": state.hand_number, "turn": state.turn, # Not the live deadline (epoch milliseconds). "deadline": 946684800000, }) await game_store.add_deadline(member, due_at=0.0) await deadlines.process_due(game_store, member) result = await game_store.load(state.id) assert result is not None self.assertEqual(state.turn, result.turn) # The entry was removed after processing. self.assertNotIn(member, await game_store.due_deadlines(float("inf"))) @async_test async def test_entry_for_expired_game_is_dropped(self) -> None: member = deadlines.encode({ "game_id": "dl-gone", "kind": deadlines.KIND_TURN, "hand": 1, "turn": 0, "deadline": 946684800000, }) await game_store.add_deadline(member, due_at=0.0) await deadlines.process_due(game_store, member) self.assertNotIn(member, await game_store.due_deadlines(float("inf"))) @async_test async def test_malformed_entry_is_dropped(self) -> None: await game_store.add_deadline("not json", due_at=0.0) await deadlines.process_due(game_store, "not json") self.assertNotIn("not json", await game_store.due_deadlines(float("inf"))) if __name__ == "__main__": unittest.main()