"""In-memory game store behaviour (the Redis store shares this interface).""" from __future__ import annotations import asyncio import time import unittest from datetime import datetime, timezone from typing import Any, Optional, cast from tavolo.platform import GameSession, Seat from tavolo.platform.registry import GameRegistry from tavolo.platform.store import ( DEADLINES_KEY, ONGOING_GAMES_KEY, InMemoryGameStore, RedisGameStore, ) from helpers import DummyEngine, async_test def _registry() -> GameRegistry: return GameRegistry([DummyEngine()]) def _session(game_id: str = "g1", code: str = "CODE01") -> GameSession: engine = DummyEngine() session = GameSession( id=game_id, game_type=engine.id, join_code=code, creator_sub="alice", players=[Seat(user_sub="alice", display_name="alice", team="A")], ) engine.create(session, {"target": 5}) return session class _FakePipeline: """Records commands and applies them to the fake Redis on execute.""" def __init__(self, redis: "_FakeRedis") -> None: self._redis = redis self._ops: list = [] async def __aenter__(self) -> "_FakePipeline": return self async def __aexit__(self, *exc: object) -> bool: return False def set(self, key: str, value: str, ex: object = None) -> None: self._ops.append(("set", key, value)) def zadd(self, key: str, mapping: dict) -> None: self._ops.append(("zadd", key, mapping)) def zrem(self, key: str, member: str) -> None: self._ops.append(("zrem", key, member)) def expire(self, key: str, ttl: int) -> None: self._ops.append(("expire", key)) def zremrangebyscore(self, key: str, minimum: object, maximum: object) -> None: self._ops.append(("zremrangebyscore", key)) def zcard(self, key: str) -> None: self._ops.append(("zcard", key)) async def execute(self) -> list: results = [self._redis.apply(op) for op in self._ops] self._ops = [] return results class _FakeRedis: """The slice of the Redis client the game store uses.""" def __init__(self) -> None: self.keys: dict = {} self.zsets: dict = {} self.zrange_calls: list = [] def pipeline(self, transaction: bool = True) -> _FakePipeline: return _FakePipeline(self) async def zadd(self, key: str, mapping: dict) -> int: return cast(int, self.apply(("zadd", key, mapping))) async def zrem(self, key: str, member: str) -> int: return cast(int, self.apply(("zrem", key, member))) async def get(self, key: str) -> object: return self.keys.get(key) async def zremrangebyscore( self, key: str, minimum: object, maximum: object ) -> int: return cast(int, self.apply(("zremrangebyscore", key))) async def zrange( self, name: str, start: Any, end: Any, byscore: bool = False, offset: Optional[int] = None, num: Optional[int] = None, **kwargs: object, ) -> list: self.zrange_calls.append((name, start, end, byscore, offset, num)) members = sorted( self.zsets.get(name, {}).items(), key=lambda item: item[1] ) if byscore: low = float("-inf") if start == "-inf" else float(start) high = float("inf") if end == "+inf" else float(end) members = [item for item in members if low <= item[1] <= high] if offset is not None and num is not None: members = members[offset : offset + num] return [member for member, _ in members] def apply(self, op: tuple) -> object: name, *args = op if name == "set": self.keys[args[0]] = args[1] return True if name == "zadd": self.zsets.setdefault(args[0], {}).update(args[1]) return 1 if name == "zrem": self.zsets.get(args[0], {}).pop(args[1], None) return 1 if name == "expire": return True if name == "zremrangebyscore": zset = self.zsets.get(args[0], {}) for member in [ member for member, score in zset.items() if score <= time.time() ]: del zset[member] return 0 if name == "zcard": return len(self.zsets.get(args[0], {})) raise AssertionError(f"unexpected command: {name}") class InMemoryGameStoreTest(unittest.TestCase): @async_test async def test_save_load_roundtrip(self) -> None: store = InMemoryGameStore(_registry()) session = _session() await store.save(session) loaded = await store.load("g1") self.assertIsNotNone(loaded) assert loaded is not None self.assertEqual("CODE01", loaded.join_code) self.assertEqual("dummy", loaded.game_type) self.assertEqual(5, loaded.state["target"]) self.assertEqual(["alice"], [p.user_sub for p in loaded.players]) # The loaded state is a deserialized copy, not the same object. self.assertIsNot(loaded.state, session.state) @async_test async def test_unknown_game_type_rejected(self) -> None: store = InMemoryGameStore(_registry()) session = _session() session.game_type = "nope" with self.assertRaises(Exception): await store.save(session) @async_test async def test_load_missing_returns_none(self) -> None: store = InMemoryGameStore(_registry()) self.assertIsNone(await store.load("nope")) self.assertIsNone(await store.find_by_code("NOPE01")) @async_test async def test_find_by_code(self) -> None: store = InMemoryGameStore(_registry()) await store.save(_session()) found = await store.find_by_code("code01") # case-insensitive self.assertIsNotNone(found) assert found is not None self.assertEqual("g1", found.id) @async_test async def test_load_returns_a_copy(self) -> None: store = InMemoryGameStore(_registry()) await store.save(_session()) first = await store.load("g1") assert first is not None first.state["target"] = 999 second = await store.load("g1") assert second is not None self.assertEqual(5, second.state["target"]) @async_test async def test_publish_reaches_subscriber(self) -> None: store = InMemoryGameStore(_registry()) await store.save(_session()) received = [] async with store.subscribe("g1") as events: await store.publish("g1") async for _ in events: received.append(True) break self.assertEqual([True], received) @async_test async def test_lock_serializes_concurrent_mutations(self) -> None: store = InMemoryGameStore(_registry()) order = [] async def holder() -> None: async with store.lock("g5"): order.append("holder-enter") await asyncio.sleep(0.05) order.append("holder-exit") async def contender() -> None: await asyncio.sleep(0.01) async with store.lock("g5"): order.append("contender") await asyncio.gather(holder(), contender()) self.assertEqual( ["holder-enter", "holder-exit", "contender"], order ) @async_test async def test_count_ongoing_tracks_unfinished_games(self) -> None: store = InMemoryGameStore(_registry()) self.assertEqual(0, await store.count_ongoing()) session = _session() await store.save(session) self.assertEqual(1, await store.count_ongoing()) other = _session(game_id="g2", code="CODE02") await store.save(other) self.assertEqual(2, await store.count_ongoing()) # A finished match is retained for the final board but no longer # counts as ongoing. session.state["finished"] = True await store.save(session) self.assertEqual(1, await store.count_ongoing()) other.state["finished"] = True await store.save(other) self.assertEqual(0, await store.count_ongoing()) @async_test async def test_list_ongoing_newest_first_and_unfinished_only(self) -> None: store = InMemoryGameStore(_registry()) old = _session("old", "CODE01") old.created_at = datetime(2026, 1, 1, tzinfo=timezone.utc) new = _session("new", "CODE02") new.created_at = datetime(2026, 1, 2, tzinfo=timezone.utc) await store.save(old) await store.save(new) self.assertEqual( ["new", "old"], [s.id for s in await store.list_ongoing()] ) self.assertEqual( ["new"], [s.id for s in await store.list_ongoing(limit=1)] ) # A finished match retained for the final board is not listed. new.state["finished"] = True await store.save(new) self.assertEqual(["old"], [s.id for s in await store.list_ongoing()]) @async_test async def test_deadline_queue(self) -> None: store = InMemoryGameStore(_registry()) self.assertIsNone(await store.next_deadline()) self.assertEqual([], await store.due_deadlines(now=100.0)) await store.add_deadline("b", due_at=50.0) await store.add_deadline("a", due_at=10.0) await store.add_deadline("c", due_at=200.0) # Re-adding an existing member only updates its due time. await store.add_deadline("b", due_at=60.0) self.assertEqual(10.0, await store.next_deadline()) self.assertEqual(["a"], await store.due_deadlines(now=10.0)) self.assertEqual(["a", "b"], await store.due_deadlines(now=100.0)) # Due entries come out in due-time order and stay queued until removed. self.assertEqual(["a", "b"], await store.due_deadlines(now=100.0)) await store.remove_deadline("a") await store.remove_deadline("a") # removing twice is a no-op self.assertEqual(60.0, await store.next_deadline()) self.assertEqual(["b"], await store.due_deadlines(now=100.0)) @async_test async def test_deadline_notifications_reach_subscribers(self) -> None: store = InMemoryGameStore(_registry()) async with store.subscribe_deadlines() as events: await store.notify_deadline() signal = await asyncio.wait_for(anext(events), timeout=1.0) self.assertIsNone(signal) # The subscription is gone after the context exits: notifying # again is a no-op rather than an error. await store.notify_deadline() class RedisGameStoreTest(unittest.TestCase): @async_test async def test_save_marks_games_ongoing_and_finished_ones_stop_counting(self) -> None: redis = _FakeRedis() store = RedisGameStore(redis, _registry(), ttl_seconds=3600) # type: ignore[arg-type] session = _session() await store.save(session) self.assertEqual(1, await store.count_ongoing()) score = redis.zsets[ONGOING_GAMES_KEY][session.id] self.assertGreater(score, time.time()) self.assertLessEqual(score, time.time() + 3600) session.state["finished"] = True await store.save(session) self.assertNotIn(session.id, redis.zsets[ONGOING_GAMES_KEY]) self.assertEqual(0, await store.count_ongoing()) @async_test async def test_count_ongoing_prunes_expired_entries(self) -> None: redis = _FakeRedis() redis.zsets[ONGOING_GAMES_KEY] = { "stale": time.time() - 10, "live": time.time() + 100, } store = RedisGameStore(redis, _registry()) # type: ignore[arg-type] self.assertEqual(1, await store.count_ongoing()) self.assertNotIn("stale", redis.zsets[ONGOING_GAMES_KEY]) @async_test async def test_list_ongoing_loads_unfinished_sessions(self) -> None: redis = _FakeRedis() store = RedisGameStore(redis, _registry(), ttl_seconds=3600) # type: ignore[arg-type] old = _session("old", "CODE01") old.created_at = datetime(2026, 1, 1, tzinfo=timezone.utc) new = _session("new", "CODE02") new.created_at = datetime(2026, 1, 2, tzinfo=timezone.utc) await store.save(old) await store.save(new) self.assertEqual( ["new", "old"], [s.id for s in await store.list_ongoing()] ) self.assertEqual( ["new"], [s.id for s in await store.list_ongoing(limit=1)] ) # A finished match retained for the final board is not listed. new.state["finished"] = True await store.save(new) self.assertEqual(["old"], [s.id for s in await store.list_ongoing()]) # An ongoing entry whose game key expired is skipped. redis.zsets[ONGOING_GAMES_KEY]["gone"] = time.time() + 100 self.assertEqual(["old"], [s.id for s in await store.list_ongoing()]) @async_test async def test_due_deadlines_uses_zrange_byscore(self) -> None: redis = _FakeRedis() store = RedisGameStore(redis, _registry()) # type: ignore[arg-type] await store.add_deadline("b", due_at=50.0) await store.add_deadline("a", due_at=10.0) await store.add_deadline("c", due_at=200.0) # Re-adding an existing member only updates its due time. await store.add_deadline("b", due_at=60.0) self.assertEqual(["a"], await store.due_deadlines(now=20.0)) self.assertEqual( [(DEADLINES_KEY, "-inf", 20.0, True, 0, 32)], redis.zrange_calls ) # Matches come back in due-time order and honour the limit. self.assertEqual(["a"], await store.due_deadlines(now=100.0, limit=1)) self.assertEqual(["a", "b"], await store.due_deadlines(now=100.0)) if __name__ == "__main__": unittest.main()