woggioni
e98d43e59a
CI / Build and push docker image (push) Successful in 6m2s
kaya-openapi 0.0.8 supports named component schemas, so the routes' description-only responses become $refs: schemas for the core payloads (auth providers, lobby, game view, ongoing games, match history, leaderboard, ratings) live in tavolo.platform.openapi and are registered through the mixin's components. Engine-defined parts (options echo, game views, match results) stay open objects on purpose. Error responses now carry the error envelope schema, and a new suite sweeps the document to keep every /api response typed.
364 lines
12 KiB
Python
364 lines
12 KiB
Python
"""Shared fixtures for the tavolo-platform test suite.
|
|
|
|
The centerpiece is :class:`DummyEngine`: a tiny two-player game
|
|
implementing the platform's
|
|
:class:`~tavolo.platform.engine.GameEngine` contract, so every platform
|
|
behaviour (lobby, store, websockets, deadlines, stats) is exercised
|
|
without importing any real game. Its rules: the first player to reach
|
|
``target`` plays wins the match.
|
|
|
|
:func:`make_platform` builds a throwaway :class:`~kaya.core.KayaApp`
|
|
wired with in-memory stores, dummy OIDC providers (patched per test) and
|
|
a sqlite :class:`~tavolo.platform.tortoise_mixin.TortoiseMixin`, so
|
|
tests construct their own app instead of importing a global one.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import contextlib
|
|
import unittest.mock as _mock
|
|
from datetime import datetime, timedelta, timezone
|
|
from functools import wraps
|
|
from typing import Any, Callable, Coroutine, Dict, Iterator, List, Mapping, Optional, Sequence, Tuple
|
|
|
|
from kaya.core import KayaApp
|
|
from kaya.oidc import OIDCConfig, OIDCProviders, OIDCUser
|
|
from kaya.openapi import OpenAPIMixin
|
|
from kaya.session import InMemorySessionStore, SessionMixin
|
|
|
|
from tavolo.platform import (
|
|
AlreadyJoined,
|
|
Deadline,
|
|
GameEngine,
|
|
GameError,
|
|
GameFinished,
|
|
GameNotStarted,
|
|
GameSession,
|
|
IllegalMove,
|
|
LobbyFull,
|
|
MatchResult,
|
|
NotYourTurn,
|
|
Platform,
|
|
PlatformMixin,
|
|
PlayerResult,
|
|
Seat,
|
|
)
|
|
from tavolo.platform.auth import get_ws_user # noqa: F401 (re-exported for patching)
|
|
from tavolo.platform.deadlines import DeadlineScheduler
|
|
from tavolo.platform.openapi import SCHEMAS
|
|
from tavolo.platform.registry import GameRegistry
|
|
from tavolo.platform.store import InMemoryGameStore
|
|
from tavolo.platform.tortoise_mixin import TortoiseMixin
|
|
|
|
|
|
class DummyEngine(GameEngine):
|
|
"""A two-player toy game: first to ``target`` plays wins.
|
|
|
|
Team labels are "A"/"B" (one player per team) so Elo paths are
|
|
exercised too. A ``deadline_in_seconds`` creation option arms a
|
|
``tick`` deadline that plays for the first player when it fires.
|
|
"""
|
|
|
|
id = "dummy"
|
|
name = "Dummy game"
|
|
description = "A two-player toy game for testing the platform."
|
|
min_players = 2
|
|
max_players = 2
|
|
options_schema = {
|
|
"type": "object",
|
|
"properties": {
|
|
"target": {
|
|
"type": "integer",
|
|
"minimum": 1,
|
|
"default": 3,
|
|
"description": "Plays needed to win the match.",
|
|
},
|
|
"deadline_in_seconds": {
|
|
"type": "number",
|
|
"description": "Arm a tick deadline this far in the future.",
|
|
},
|
|
},
|
|
}
|
|
|
|
def create(self, session: GameSession, options: Mapping[str, Any]) -> None:
|
|
target = options.get("target", 3)
|
|
if isinstance(target, bool) or not isinstance(target, int) or target < 1:
|
|
raise IllegalMove("target must be a positive integer")
|
|
# The creator takes team A.
|
|
creator = session.players[0]
|
|
session.players[0] = Seat(
|
|
user_sub=creator.user_sub,
|
|
display_name=creator.display_name,
|
|
team="A",
|
|
)
|
|
session.state = {
|
|
"target": target,
|
|
"plays": [],
|
|
"started": False,
|
|
"finished": False,
|
|
"winner": None,
|
|
"deadline_in_seconds": options.get("deadline_in_seconds"),
|
|
}
|
|
|
|
def join(self, session: GameSession, user_sub: str, display_name: str) -> None:
|
|
state = session.state
|
|
if state["started"]:
|
|
raise GameNotStarted("game has already started")
|
|
if session.seated(user_sub):
|
|
raise AlreadyJoined("already joined this game")
|
|
if len(session.players) >= 2:
|
|
raise LobbyFull("game is full")
|
|
session.players.append(
|
|
Seat(
|
|
user_sub=user_sub,
|
|
display_name=display_name,
|
|
team="A" if not session.players else "B",
|
|
)
|
|
)
|
|
if len(session.players) == 2:
|
|
state["started"] = True
|
|
|
|
def handle_action(
|
|
self, session: GameSession, user_sub: str, action: str, payload: Mapping[str, Any]
|
|
) -> None:
|
|
state = session.state
|
|
if not state["started"]:
|
|
raise GameNotStarted("the game has not started yet")
|
|
if state["finished"]:
|
|
raise GameFinished("the match is over")
|
|
if not session.seated(user_sub):
|
|
raise NotYourTurn("you are not seated in this game")
|
|
if action != "play":
|
|
raise IllegalMove(f"unknown action: {action!r}")
|
|
self._play(session, user_sub)
|
|
|
|
def _play(self, session: GameSession, user_sub: str) -> None:
|
|
state = session.state
|
|
state["plays"].append(user_sub)
|
|
if len(state["plays"]) >= state["target"]:
|
|
state["finished"] = True
|
|
state["winner"] = user_sub
|
|
|
|
def view_for(self, session: GameSession, user_sub: Optional[str]) -> Dict[str, Any]:
|
|
state = session.state
|
|
return {
|
|
"started": state["started"],
|
|
"finished": state["finished"],
|
|
"winner": state["winner"],
|
|
"plays": len(state["plays"]),
|
|
"target": state["target"],
|
|
"viewer_seated": user_sub is not None and session.seated(user_sub),
|
|
}
|
|
|
|
def lobby_view(self, session: GameSession) -> Dict[str, Any]:
|
|
return {
|
|
"phase": "playing" if session.state["started"] else "lobby",
|
|
"target": session.state["target"],
|
|
}
|
|
|
|
def in_lobby(self, session: GameSession) -> bool:
|
|
return not session.state["started"]
|
|
|
|
def is_finished(self, session: GameSession) -> bool:
|
|
return bool(session.state["finished"])
|
|
|
|
def result(self, session: GameSession) -> MatchResult:
|
|
state = session.state
|
|
winner = state["winner"]
|
|
if winner is None:
|
|
raise GameError("no result: the match is not finished")
|
|
subs = [seat.user_sub for seat in session.players]
|
|
plays: List[str] = state["plays"]
|
|
return MatchResult(
|
|
teams=[[subs[0]], [subs[1]]],
|
|
winner_team=subs.index(winner),
|
|
players=[
|
|
PlayerResult(
|
|
user_sub=seat.user_sub,
|
|
seat=index,
|
|
won=seat.user_sub == winner,
|
|
team=seat.team,
|
|
score=float(plays.count(seat.user_sub)),
|
|
details={"plays": plays.count(seat.user_sub)},
|
|
)
|
|
for index, seat in enumerate(session.players)
|
|
],
|
|
summary={
|
|
"target": state["target"],
|
|
"plays": len(plays),
|
|
"winner": winner,
|
|
},
|
|
)
|
|
|
|
def state_to_json(self, state: Any) -> Dict[str, Any]:
|
|
return dict(state)
|
|
|
|
def state_from_json(self, data: Mapping[str, Any]) -> Any:
|
|
return dict(data)
|
|
|
|
def next_deadline(self, session: GameSession) -> Optional[Deadline]:
|
|
seconds = session.state.get("deadline_in_seconds")
|
|
if (
|
|
seconds is None
|
|
or not session.state["started"]
|
|
or session.state["finished"]
|
|
):
|
|
return None
|
|
due_at = datetime.now(timezone.utc) + timedelta(seconds=float(seconds))
|
|
return Deadline(
|
|
kind="tick",
|
|
due_at=due_at,
|
|
token=f"tick:{len(session.state['plays'])}",
|
|
)
|
|
|
|
def fire_deadline(self, session: GameSession, kind: str, token: str) -> None:
|
|
current = self.next_deadline(session)
|
|
if current is None or current.kind != kind or current.token != token:
|
|
raise GameError("stale deadline")
|
|
# The house plays for the first player.
|
|
self._play(session, session.players[0].user_sub)
|
|
|
|
|
|
def make_platform(
|
|
engines: Sequence[GameEngine] = (DummyEngine(),),
|
|
heartbeat_ms: int = 50,
|
|
oidc_providers: Sequence[str] = ("default",),
|
|
) -> Tuple[KayaApp, Platform, TortoiseMixin]:
|
|
"""Build a throwaway app + platform wired with in-memory stores."""
|
|
registry = GameRegistry(engines)
|
|
game_store = InMemoryGameStore(registry)
|
|
session_mixin = SessionMixin(InMemorySessionStore())
|
|
oidc = OIDCProviders(
|
|
[
|
|
OIDCConfig(
|
|
provider=name,
|
|
issuer=f"http://localhost:8180/{name}",
|
|
client_id="tavolo",
|
|
client_secret=None,
|
|
redirect_uri=f"http://localhost:8080/auth/{name}/callback",
|
|
post_login_redirect="/",
|
|
post_logout_redirect="/",
|
|
)
|
|
for name in oidc_providers
|
|
],
|
|
session=session_mixin,
|
|
)
|
|
tortoise_mixin = TortoiseMixin(
|
|
database_url="sqlite://:memory:",
|
|
models_modules=["tavolo.platform.models"],
|
|
skip_paths=frozenset({"/api/health", "/api/docs", "/api/openapi.json"}),
|
|
)
|
|
openapi_mixin = OpenAPIMixin(
|
|
title="tavolo-platform-tests",
|
|
version="0.1.0",
|
|
description="test app",
|
|
spec_path="/api/openapi.json",
|
|
docs_path="/api/docs",
|
|
components={"schemas": SCHEMAS},
|
|
)
|
|
scheduler = DeadlineScheduler(
|
|
game_store,
|
|
registry,
|
|
heartbeat_ms=heartbeat_ms,
|
|
context_binder=tortoise_mixin.ensure_context,
|
|
)
|
|
platform = Platform(
|
|
registry=registry,
|
|
game_store=game_store,
|
|
scheduler=scheduler,
|
|
oidc=oidc,
|
|
)
|
|
app = KayaApp(
|
|
mixins=[
|
|
session_mixin,
|
|
oidc,
|
|
tortoise_mixin,
|
|
openapi_mixin,
|
|
PlatformMixin(platform),
|
|
]
|
|
)
|
|
_tortoise_mixins.append(tortoise_mixin)
|
|
_schedulers.append(scheduler)
|
|
return app, platform, tortoise_mixin
|
|
|
|
|
|
_tortoise_mixins: List[TortoiseMixin] = []
|
|
_schedulers: List[DeadlineScheduler] = []
|
|
|
|
|
|
def async_test(coro: Callable[..., Coroutine[Any, Any, None]]) -> Callable[..., None]:
|
|
"""Like ``pwo.async_test`` (fresh loop per test), but tear down the
|
|
platform pieces afterwards: close Tortoise contexts (otherwise
|
|
orphaned aiosqlite threads block interpreter shutdown) and stop
|
|
deadline consumers."""
|
|
|
|
@wraps(coro)
|
|
def wrapper(*args: Any, **kwargs: Any) -> None:
|
|
async def run() -> None:
|
|
loop = asyncio.get_running_loop()
|
|
try:
|
|
await coro(*args, **kwargs)
|
|
finally:
|
|
for scheduler in _schedulers:
|
|
scheduler.stop_consumer(loop)
|
|
_schedulers.clear()
|
|
for mixin in _tortoise_mixins:
|
|
await mixin.aclose()
|
|
_tortoise_mixins.clear()
|
|
|
|
with asyncio.Runner() as runner:
|
|
runner.run(run())
|
|
|
|
return wrapper
|
|
|
|
|
|
async def use_db(tortoise_mixin: TortoiseMixin):
|
|
"""Bind the app's Tortoise context for this loop, for seeding rows."""
|
|
from tortoise.context import TortoiseContext
|
|
|
|
await tortoise_mixin._bind()
|
|
ctx: Optional[TortoiseContext] = tortoise_mixin._ctx
|
|
assert ctx is not None
|
|
return ctx
|
|
|
|
|
|
def make_user(sub: str, name: Optional[str] = None) -> OIDCUser:
|
|
return OIDCUser({"sub": sub, "preferred_username": name or sub})
|
|
|
|
|
|
@contextlib.contextmanager
|
|
def oidc_user(oidc: OIDCProviders, sub: str, name: Optional[str] = None) -> Iterator[OIDCUser]:
|
|
"""Context manager: make ``oidc.get_users`` return this user for the
|
|
first configured provider."""
|
|
user = make_user(sub, name)
|
|
provider = next(iter(oidc.providers))
|
|
patcher = _mock.patch.object(oidc, "get_users", return_value={provider: user})
|
|
patcher.start()
|
|
try:
|
|
yield user
|
|
finally:
|
|
patcher.stop()
|
|
|
|
|
|
@contextlib.contextmanager
|
|
def ws_users(users: Sequence[OIDCUser]) -> Iterator[None]:
|
|
"""Context manager: patch ``auth.get_ws_user`` to hand out ``users``
|
|
one per websocket connection, in order. Once exhausted it keeps
|
|
returning the last user."""
|
|
from tavolo.platform import auth
|
|
|
|
remaining = list(users)
|
|
last = remaining[-1] if remaining else None
|
|
|
|
def _next(_ws):
|
|
if remaining:
|
|
return remaining.pop(0)
|
|
return last
|
|
|
|
patcher = _mock.patch.object(auth, "get_ws_user", side_effect=_next)
|
|
patcher.start()
|
|
try:
|
|
yield
|
|
finally:
|
|
patcher.stop()
|