woggioni
e9ddb82e9a
Multiplayer scopone scientifico backend on the kaya framework:
- OIDC login (kaya-oidc), session-backed WebSocket auth
- Pure rules engine (forced captures, scopa, primiera scoring) with
full-match simulation tests
- Live game state in Redis (JSON + TTL, join codes, per-game locks,
pub/sub state push); in-memory fallback for tests
- WebSocket /ws/games/{id} for real-time play; REST lobby endpoints
(create/join/snapshot) with hidden-hand views
- Finished matches persisted to Postgres (Tortoise + aerich) for match
history and leaderboard endpoints
- Docker Compose stack: postgres, redis, mock-oauth2-server, db-migrate, app
- 45 tests passing; mypy clean
57 lines
1.8 KiB
Python
57 lines
1.8 KiB
Python
"""Helpers for faking the OIDC authenticated user during tests.
|
|
|
|
HTTP handlers funnel through ``oidc_mixin.get_user``; websocket handlers
|
|
through :func:`scopa.auth.get_ws_user`. Patching those two entry points
|
|
lets route and websocket tests run entirely in-process with no IdP.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import contextlib
|
|
import unittest.mock as _mock
|
|
from typing import Iterator, Optional, Sequence
|
|
|
|
from kaya.oidc import OIDCUser
|
|
|
|
# Import the app first: it pulls in the route modules, which import
|
|
# ``scopa.auth`` themselves. Importing ``auth`` before ``app`` would hit a
|
|
# partially initialized module (same constraint as reimpasto).
|
|
from scopa.app import oidc_mixin
|
|
from scopa import auth
|
|
|
|
|
|
def make_user(sub: str, name: Optional[str] = None) -> OIDCUser:
|
|
return OIDCUser({"sub": sub, "preferred_username": name or sub})
|
|
|
|
|
|
@contextlib.contextmanager
|
|
def oidc_user(sub: str, name: Optional[str] = None) -> Iterator[OIDCUser]:
|
|
"""Context manager: patch ``oidc_mixin.get_user`` to return this user."""
|
|
user = make_user(sub, name)
|
|
patcher = _mock.patch.object(oidc_mixin, "get_user", return_value=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."""
|
|
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()
|