324 lines
11 KiB
Python
324 lines
11 KiB
Python
"""In-memory game store behaviour (the Redis store shares this interface)."""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import time
|
|
import unittest
|
|
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))
|
|
|
|
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 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]] = "payload"
|
|
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_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))
|
|
|
|
|
|
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_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()
|