woggioni
71493f720b
- Add canonical_principal(provider, sub) => 'oidc:v1:<provider>:<sub>' - Preserve provider identity across HTTP sessions and WebSockets - Apply principals throughout: game creation/joining/seating, WebSocket views/actions, Redis engine state, match history, leaderboard focus, ratings, /api/me, OpenAPI, and telemetry - Widen identity columns to 1024 chars with schema-only aerich migration (5_20261024221522_canonical_principals.py) - Document provider names as stable security identifiers - Add regression tests: same subject at different providers cannot share games, private hands, join codes, moves, history, or ratings; distinct seat and rating ownership verified - All platform, integration, game, and frontend tests pass; mypy clean
96 lines
2.8 KiB
Python
96 lines
2.8 KiB
Python
"""Helpers for faking the OIDC authenticated user during tests.
|
|
|
|
HTTP handlers funnel through ``OIDCProviders.get_users`` (wrapped by
|
|
``tavolo.platform.auth.current_user``); websocket handlers through
|
|
:func:`tavolo.platform.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
|
|
# ``tavolo.platform.auth`` themselves.
|
|
from tavolo.app import oidc
|
|
from tavolo.platform import auth
|
|
from tavolo.platform.auth import AuthenticatedUser, canonical_principal
|
|
|
|
# The app's first configured provider (the test environment configures one
|
|
# named "default").
|
|
_PROVIDER = next(iter(oidc.providers))
|
|
|
|
|
|
def make_user(
|
|
sub: str,
|
|
name: Optional[str] = None,
|
|
email: Optional[str] = None,
|
|
picture: Optional[str] = None,
|
|
) -> OIDCUser:
|
|
claims = {"sub": sub, "preferred_username": name or sub}
|
|
if email:
|
|
claims["email"] = email
|
|
if picture:
|
|
claims["picture"] = picture
|
|
return OIDCUser(claims)
|
|
|
|
|
|
def make_principal(sub: str, provider: str = _PROVIDER) -> str:
|
|
return canonical_principal(provider, sub)
|
|
|
|
|
|
@contextlib.contextmanager
|
|
def oidc_user(
|
|
sub: str,
|
|
name: Optional[str] = None,
|
|
email: Optional[str] = None,
|
|
picture: Optional[str] = None,
|
|
provider: Optional[str] = None,
|
|
) -> Iterator[OIDCUser]:
|
|
"""Context manager: make ``oidc.get_users`` return this user.
|
|
|
|
``email`` is optional so tests can simulate an IdP that still returns
|
|
the claim: the app requests only the ``openid`` and ``profile`` scopes
|
|
and ignores email where present. ``picture`` simulates the profile
|
|
scope's avatar claim.
|
|
"""
|
|
user = make_user(sub, name, email, picture)
|
|
selected = provider or _PROVIDER
|
|
patcher = _mock.patch.object(oidc, "get_users", return_value={selected: 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 = [
|
|
AuthenticatedUser(
|
|
provider=_PROVIDER,
|
|
claims=user,
|
|
principal=canonical_principal(_PROVIDER, user.sub),
|
|
)
|
|
for user in 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()
|