Compare commits

...
Author SHA1 Message Date
woggioni f59064e580 Upgrade kaya to 0.0.7
CI / Build and push docker image (push) Successful in 1m56s
2026-10-01 23:43:55 +08:00
woggioni a89bd06010 Support multiple OIDC providers
CI / Build and push docker image (push) Successful in 2m27s
kaya-oidc 0.0.6 can run several providers in one app (OIDCProviders),
each with its own login route and namespaced session state. Adopt it:
providers are configured in indexed, contiguous OIDC_<n>_* blocks
(0-based) with OIDC_<n>_PROVIDER naming the provider in its login URL
and session keys. The old unindexed OIDC_ISSUER/OIDC_CLIENT_ID/... names
are no longer read; startup fails with a rename hint when they are set.
With no OIDC variables at all, the single dev mock provider named
"default" is used, as before.

The platform now authenticates through OIDCProviders: auth.current_user
returns the session's user (sessions stay single-identity, so at most one
provider is logged in) and require_auth/route handlers use it, while the
websocket path keeps scanning for any provider's user. New routes expose
the configuration to the SPA and keep old links working: GET
/api/auth/providers lists the login URLs, GET /auth/login redirects to
the first provider and GET /auth/logout to whichever provider the
session is logged in with. The lobby renders one login button per
provider from that endpoint.

Update the env examples, compose and k8s manifests to the indexed scheme,
document the multiple-provider setup and the migration, and cover it with
config, auth and route tests.
2026-10-01 20:38:51 +08:00
woggioni 5eccaafadc Upgrade kaya to 0.0.6
kaya-oidc 0.0.6 namespaces its session state per provider as
oidc:{provider}:user (the default provider is "default", so the key is
oidc:default:user). The unprefixed oidc_user key is no longer written:
left unadapted, every websocket upgrade would be rejected with 401 and
traced requests would silently lose their user.id attribute. Adopt it
with a session_user helper on tavolo.platform.auth that finds any
provider's user, used by both get_ws_user and telemetry's
session_tag_hook. Websocket and route handlers keep their signatures.

kaya-otel 0.0.6 only enriches websocket spans with standard URL, network
and peer attributes (plus error.type), which needs no application change.

Regenerate the lock: the eight kaya pins move to 0.0.6 and, the wheels'
declared metadata being unchanged, nothing else moves. tavolo-platform
now floors kaya-core>=0.0.6 (it reads the new session layout) and
tavolo-app floors kaya-otel>=0.0.6.

As kaya documents, the session-key move signs existing logged-in users
out once; they simply log in again.
2026-10-01 20:38:21 +08:00
woggioni 6e9d1a4652 Wake deadline consumers on enqueue instead of polling every second
CI / Build and push docker image (push) Successful in 1m15s
2026-09-30 07:40:04 +08:00
woggioni 723b3b11cf Upgrade kaya to 0.0.5
CI / Build and push docker image (push) Successful in 1m20s
kaya 0.0.5 surfaces a vanished websocket peer as
kaya.core.WebSocketDisconnect instead of leaking granian's
RSGIProtocolClosed: receive() reports a close message so iteration ends
normally, sends raise the new exception, and KayaApp.handle_websocket
swallows it as normal termination. Adopt it: drop the granian-specific
workaround from the platform websocket handler, catch
WebSocketDisconnect in the send helper, and model the new semantics in
the abrupt-disconnect test (with coverage that a send on a vanished
peer is ignored).

Regenerate the lock: the eight kaya pins move to 0.0.5 (declared
metadata is unchanged) and pip-compile also bumps PyJWT to 2.15.1, a
security release with no breaking changes for kaya-oidc's jwt.decode
usage. tavolo-platform now floors kaya-core>=0.0.5 (it imports
WebSocketDisconnect) and tavolo-app floors kaya-otel>=0.0.5.
2026-09-30 06:38:02 +08:00
woggioni ed850aa73c Sort the player's hand by rank then suit 2026-09-30 00:58:31 +08:00
woggioni 2910fc4600 Refuse websocket upgrades with HTTP statuses, not close codes
A close() before accept() carries the HTTP status of the refused
upgrade, not a WebSocket close code: RSGI (granian) uses the value
verbatim as the response status, so 4401 panicked on the invalid status
(InvalidStatusCode) and — granian being built with panic = "abort" —
aborted the worker and took the whole app down. ASGI servers answer 403
regardless, and browsers report a refused handshake as an abnormal close
(1006), so the 4401/4403/4404 codes never reached clients anyway.

Reject 401/404/403 to mirror GET /api/games/{id}, and classify a failed
first connection in the game page through the REST API so players still
see "session expired" / "game no longer available" instead of a retry
loop. Add a regression test asserting every pre-accept rejection uses a
status in the HTTP range.
2026-09-29 22:31:41 +08:00
woggioni 21911caf9e Poll deadlines with ZRANGE BYSCORE instead of the deprecated ZRANGEBYSCORE 2026-09-28 05:50:54 +08:00
woggioni 08f158ef28 Fetch match-history participants in one query per page
CI / Build and push docker image (push) Successful in 53s
2026-09-28 05:21:34 +08:00
woggioni 834f1f26a9 Tag traced requests with the session id and OIDC sub 2026-09-28 05:11:22 +08:00
woggioni bcd1371019 Cap urllib3.connectionpool at INFO in the default logging config
The root logger runs at DEBUG, so urllib3.connectionpool emitted a
record per pooled HTTP request. Keep its INFO+ output while dropping
the per-connection chatter.
2026-09-28 04:38:28 +08:00
woggioni 3cf7e2e257 Show match finish times in the browser timezone
CI / Build and push docker image (push) Successful in 2m18s
The history table truncated finished_at, which the API serializes in
UTC, so the Finished column always showed UTC wall time. Parse the
timestamp with the browser's Date and render its local components,
keeping the existing YYYY-MM-DD HH:MM layout.
2026-09-28 04:26:07 +08:00
woggioni 2bdf247cb3 Restore cursor value types before keyset filtering
Cursors are JSON, so a datetime sort key reached the next request as a
string. Tortoise's DatetimeField.to_db_value does not parse strings, so
asyncpg rejected the string where a timestamptz parameter was expected;
sqlite tolerated it, hiding the bug in tests. Coerce decoded cursor
values through their model field's to_python_value before building the
keyset filter.
2026-09-28 04:26:00 +08:00
woggioni a3e8ff3206 Trace Postgres and Redis queries with the contrib instrumentations
CI / Build and push docker image (push) Successful in 2m42s
Builds the TracerProvider in tavolo.telemetry (kaya-otel's is private) and
injects it into kaya-otel, then instruments asyncpg and redis through
opentelemetry-instrumentation-asyncpg/redis. Redis command values are
redacted and asyncpg spans carry the parameterized SQL only.
2026-09-28 03:38:59 +08:00
woggioni 98d4485375 Export app, process and log telemetry when OTEL_ENABLED is set
Adds tavolo.active_users and tavolo.ongoing_games gauges, process/CPython
metrics via opentelemetry-instrumentation-system-metrics and OTLP log
export. The game store tracks ongoing matches (Redis sorted set, in-memory
set) and the session stores expose live counts for the gauges; a poller
refreshes them since metric callbacks cannot await the stores.
2026-09-28 03:38:59 +08:00
woggioni 5be4926885 Bundle kaya-otel in the Docker image
CI / Build and push docker image (push) Successful in 1m57s
The tavolo-app otel extra was never installed: requirements.txt was
locked without it and the Dockerfile installs the local distributions
with --no-deps, so any container with OTEL_ENABLED set died at import
with "kaya-otel is not installed".

Add the extra's full closure (kaya-otel plus its OpenTelemetry/protobuf
dependencies) to the pinned lock and refresh the comments that claimed
the extra had to be added separately.
2026-09-26 15:47:16 +08:00
47 changed files with 2841 additions and 238 deletions
+6 -2
View File
@@ -78,8 +78,12 @@ Trunk proxies `/api`, `/auth` and `/ws` to `127.0.0.1:8080` (see
run the backend with:
```sh
OIDC_POST_LOGIN_REDIRECT=http://localhost:8000/ \
OIDC_POST_LOGOUT_REDIRECT=http://localhost:8000/ \
OIDC_0_PROVIDER=default \
OIDC_0_ISSUER=http://localhost:8180/tavolo \
OIDC_0_CLIENT_ID=tavolo \
OIDC_0_REDIRECT_URI=http://localhost:8080/auth/default/callback \
OIDC_0_POST_LOGIN_REDIRECT=http://localhost:8000/ \
OIDC_0_POST_LOGOUT_REDIRECT=http://localhost:8000/ \
.venv/bin/granian --host 127.0.0.1 --port 8080 tavolo.app:app
```
+22 -12
View File
@@ -21,7 +21,7 @@
# - This file deploys a ClusterIP Service only. Until you add an Ingress,
# reach the app with:
# kubectl port-forward -n tavolo svc/tavolo 8080:80
# and set OIDC_REDIRECT_URI to match whatever URL the browser uses.
# and set OIDC_0_REDIRECT_URI to match whatever URL the browser uses.
---
apiVersion: v1
@@ -71,22 +71,32 @@ data:
# CORS_ALLOW_CREDENTIALS: "false"
# CORS_EXPOSE_HEADERS: ""
# CORS_MAX_AGE: "600"
# OpenTelemetry (kaya-otel): traces + metrics via OTLP/HTTP, disabled
# unless OTEL_ENABLED is set. Requires the otel extra in the image.
# OpenTelemetry (kaya-otel + contrib instrumentations): traces
# (HTTP/WebSocket, Postgres, Redis), metrics and logs via OTLP/HTTP,
# disabled unless OTEL_ENABLED is set. The image bundles the otel extra.
# OTEL_ENABLED: "true"
# OTEL_SERVICE_NAME: "tavolo"
# OTEL_EXPORTER_OTLP_ENDPOINT: "http://otel-collector.observability:4318"
# OTEL_EXPORTER_OTLP_HEADERS: "Authorization=Bearer ..."
# OTEL_EXCLUDED_PATHS: "/api/health" # default; paths skipped by tracing
# OIDC (provider lives in another namespace).
OIDC_CLIENT_ID: tavolo
OIDC_POST_LOGIN_REDIRECT: /
OIDC_POST_LOGOUT_REDIRECT: /
# OTEL_METRIC_EXPORT_INTERVAL_MILLIS: "60000"
# OTEL_METRICS_POLL_INTERVAL_SECONDS: "15"
# OIDC (the provider lives in another namespace). Each provider is an
# indexed block: OIDC_<n>_PROVIDER names it in the login URL
# (/auth/<provider>/login) and indices must be contiguous from 0. Add
# OIDC_1_*, OIDC_2_*, ... blocks for more providers; add their secrets
# to tavolo-secrets as well.
OIDC_0_PROVIDER: default
OIDC_0_CLIENT_ID: tavolo
OIDC_0_POST_LOGIN_REDIRECT: /
OIDC_0_POST_LOGOUT_REDIRECT: /
# Issuer URL as seen by the browser AND the pods (see notes above).
OIDC_ISSUER: REPLACE_ME
OIDC_0_ISSUER: REPLACE_ME
# Public callback URL of this deployment, e.g.
# http://<host>:<port>/auth/callback (must be allowed at the provider).
OIDC_REDIRECT_URI: REPLACE_ME
# http://<host>:<port>/auth/default/callback (must be allowed at the
# provider). When unset it defaults to
# http://localhost:8080/auth/<provider>/callback.
OIDC_0_REDIRECT_URI: REPLACE_ME
---
apiVersion: v1
@@ -103,8 +113,8 @@ stringData:
# The only Postgres secret: the password for DATABASE_USER at
# DATABASE_HOST (both configured in the tavolo-config ConfigMap).
DATABASE_PASSWORD: REPLACE_ME
# Client secret for OIDC_CLIENT_ID at the provider.
OIDC_CLIENT_SECRET: REPLACE_ME
# Client secret for OIDC_0_CLIENT_ID at the provider.
OIDC_0_CLIENT_SECRET: REPLACE_ME
---
# Redis: sessions + live game state. Ephemeral by design (emptyDir): worst
+13 -8
View File
@@ -103,13 +103,15 @@ services:
DATABASE_USER: tavolo
DATABASE_PASSWORD: ${DATABASE_PASSWORD:-password}
# By default the app and browsers reach the mock IdP under the same
# name (see README /etc/hosts note); override OIDC_ISSUER and
# OIDC_REDIRECT_URI to use a real provider or a different host port.
OIDC_ISSUER: ${OIDC_ISSUER:-http://mockoauth:8180/tavolo}
# name (see README /etc/hosts note); override the OIDC_0_* values to
# use a real provider or a different host port. Additional providers
# take OIDC_1_*, OIDC_2_*, ... blocks (indices are contiguous from 0).
OIDC_0_PROVIDER: ${OIDC_0_PROVIDER:-default}
OIDC_0_ISSUER: ${OIDC_0_ISSUER:-http://mockoauth:8180/tavolo}
# The mock OIDC server does not validate clients: any id/secret works.
OIDC_CLIENT_ID: tavolo
OIDC_CLIENT_SECRET: ${OIDC_CLIENT_SECRET:-dev-secret}
OIDC_REDIRECT_URI: ${OIDC_REDIRECT_URI:-http://localhost:8080/auth/callback}
OIDC_0_CLIENT_ID: tavolo
OIDC_0_CLIENT_SECRET: ${OIDC_0_CLIENT_SECRET:-dev-secret}
OIDC_0_REDIRECT_URI: ${OIDC_0_REDIRECT_URI:-http://localhost:8080/auth/default/callback}
REDIS_URL: redis://redis:6379/0
HAND_ACK_TIMEOUT_SECONDS: ${HAND_ACK_TIMEOUT_SECONDS:-30}
TURN_TIMEOUT_SECONDS: ${TURN_TIMEOUT_SECONDS:-30}
@@ -122,8 +124,11 @@ services:
CORS_ALLOW_CREDENTIALS: ${CORS_ALLOW_CREDENTIALS:-}
CORS_EXPOSE_HEADERS: ${CORS_EXPOSE_HEADERS:-}
CORS_MAX_AGE: ${CORS_MAX_AGE:-}
# OpenTelemetry (kaya-otel): disabled unless OTEL_ENABLED is set.
# Requires the otel extra in the image (see server/pyproject.toml).
# OpenTelemetry (kaya-otel + contrib instrumentations): traces
# (HTTP/WebSocket, Postgres, Redis), metrics and logs via OTLP/HTTP,
# disabled unless OTEL_ENABLED is set.
# The image bundles the otel extra (see packages/tavolo-app/pyproject.toml
# and server/requirements.txt).
OTEL_ENABLED: ${OTEL_ENABLED:-}
OTEL_SERVICE_NAME: ${OTEL_SERVICE_NAME:-}
OTEL_EXPORTER_OTLP_ENDPOINT: ${OTEL_EXPORTER_OTLP_ENDPOINT:-}
+40 -6
View File
@@ -13,12 +13,29 @@ DATABASE_OPTIONS=
# test suite (sqlite://:memory:) and handy for managed-DB URLs.
#DATABASE_URL=postgres://tavolo:password@localhost:5432/tavolo
# OIDC (mock-oauth2-server in dev; it does not validate clients, so any
# client id/secret works. For a real IdP like Keycloak, use its values here.)
OIDC_ISSUER=http://localhost:8180/tavolo
OIDC_CLIENT_ID=tavolo
OIDC_CLIENT_SECRET=dev-secret
OIDC_REDIRECT_URI=http://localhost:8080/auth/callback
# OIDC providers. Each provider is an indexed block: OIDC_<n>_PROVIDER
# (required; names the provider in the login URL /auth/<provider>/login and
# in its session state), OIDC_<n>_ISSUER, OIDC_<n>_CLIENT_ID,
# OIDC_<n>_CLIENT_SECRET (optional: leave unset for public clients),
# OIDC_<n>_REDIRECT_URI (defaults to
# http://localhost:8080/auth/<provider>/callback), OIDC_<n>_POST_LOGIN_REDIRECT
# and OIDC_<n>_POST_LOGOUT_REDIRECT (default /). Indices must be contiguous
# from 0; add a second provider with an OIDC_1_* block (and so on).
#
# The values below target the dev mock-oauth2-server (it does not validate
# clients, so any client id/secret works). For a real IdP like Keycloak,
# use its values and register the redirect URI there.
OIDC_0_PROVIDER=default
OIDC_0_ISSUER=http://localhost:8180/tavolo
OIDC_0_CLIENT_ID=tavolo
OIDC_0_CLIENT_SECRET=dev-secret
OIDC_0_REDIRECT_URI=http://localhost:8080/auth/default/callback
# A second provider, e.g. Google:
#OIDC_1_PROVIDER=google
#OIDC_1_ISSUER=https://accounts.google.com
#OIDC_1_CLIENT_ID=...
#OIDC_1_CLIENT_SECRET=...
#OIDC_1_REDIRECT_URI=http://localhost:8080/auth/google/callback
# Redis (session and live-game storage). Unset to fall back to in-memory
# stores (only sensible for local development with a single worker).
@@ -61,6 +78,23 @@ TURN_TIMEOUT_SECONDS=30
# Seconds browsers may cache the preflight response (default 600):
#CORS_MAX_AGE=600
# OpenTelemetry (via kaya-otel + the contrib instrumentation packages).
# Off unless OTEL_ENABLED is truthy; exports traces (HTTP/WebSocket,
# Postgres via asyncpg, Redis), metrics and logs over OTLP/HTTP (requires
# the tavolo-app "otel" extra, bundled in the Docker image). The endpoint
# defaults to the standard OTEL_EXPORTER_OTLP_ENDPOINT (localhost:4318)
# when unset.
#OTEL_ENABLED=true
#OTEL_SERVICE_NAME=tavolo
#OTEL_EXPORTER_OTLP_ENDPOINT=http://otel-collector:4318
#OTEL_EXPORTER_OTLP_HEADERS=Authorization=Bearer token,X-Tenant=one
# Paths excluded from tracing and metrics (default /api/health).
#OTEL_EXCLUDED_PATHS=/api/health
# Metric export interval (ms, default 60000) and session/game count poll
# interval (seconds, default 15).
#OTEL_METRIC_EXPORT_INTERVAL_MILLIS=60000
#OTEL_METRICS_POLL_INTERVAL_SECONDS=15
# App server
APP_HOST=0.0.0.0
APP_PORT=8080
+2 -1
View File
@@ -60,7 +60,8 @@ RUN --mount=type=cache,target=/root/.cache/pip \
# The three sibling distributions, installed from the local tree (PEP 420
# namespace packages sharing the `tavolo` namespace). --no-deps: every
# third-party dependency is already pinned in requirements.txt.
# third-party dependency — including the tavolo-app `otel` extra — is
# already pinned in requirements.txt.
COPY server/packages/ ./packages/
RUN --mount=type=cache,target=/root/.cache/pip \
/opt/venv/bin/pip install --no-deps ./packages/tavolo-platform ./packages/tavolo-scopone ./packages/tavolo-app
+91 -9
View File
@@ -43,7 +43,8 @@ Because the browser and the app both talk to the OIDC issuer at
echo "127.0.0.1 mockoauth" | sudo tee -a /etc/hosts
```
Then log in: `GET http://localhost:8080/auth/login`, type any username
Then log in: `GET http://localhost:8080/auth/login` (a provider-agnostic
alias that redirects to the first configured provider), type any username
(e.g. `alice`) in the interactive login, and you are redirected back with a
session cookie. That cookie authenticates both the REST API and the
WebSocket endpoint.
@@ -65,14 +66,17 @@ All configuration comes from environment variables (see `.env.example`):
| `DATABASE_OPTIONS` | unset | Extra DSN query parameters, e.g. `ssl=require` |
| `DATABASE_URL` | unset | Full-DSN override; when set, the `DATABASE_*` parts above are ignored (used for sqlite in tests and for managed-DB URLs) |
| `REDIS_URL` | unset | Redis DSN for sessions + live games. Unset falls back to in-memory stores |
| `OIDC_ISSUER` | `http://localhost:8180/tavolo` | OIDC issuer URL |
| `OIDC_CLIENT_ID` | `tavolo` | OIDC client id |
| `OIDC_CLIENT_SECRET` | unset | OIDC client secret |
| `OIDC_REDIRECT_URI` | `http://localhost:8080/auth/callback` | Login callback URL |
| `OIDC_<n>_PROVIDER` | `default` ¹ | Provider name (required when OIDC is configured): used in the login URL `/auth/<provider>/login` and to namespace its session keys. Unique, `[A-Za-z0-9_-]+` |
| `OIDC_<n>_ISSUER` | `http://localhost:8180/tavolo` ¹ | OIDC issuer URL |
| `OIDC_<n>_CLIENT_ID` | `tavolo` ¹ | OIDC client id |
| `OIDC_<n>_CLIENT_SECRET` | unset | OIDC client secret; leave unset for public clients |
| `OIDC_<n>_REDIRECT_URI` | `http://localhost:8080/auth/<provider>/callback` | Login callback URL; must be registered at the provider |
| `OIDC_<n>_POST_LOGIN_REDIRECT` | `/` | Where the browser is sent after login |
| `OIDC_<n>_POST_LOGOUT_REDIRECT` | `/` | Where the browser is sent after logout |
| `GAME_TTL_SECONDS` | `86400` | Sliding TTL of a live game in Redis |
| `HAND_ACK_TIMEOUT_SECONDS` | `30` | Seconds the scopone between-hands scoring summary waits for acknowledgements (wired into the `ScoponeEngine`) |
| `TURN_TIMEOUT_SECONDS` | `30` | Seconds a scopone player has to play before the server plays a random legal card for them (wired into the `ScoponeEngine`) |
| `DEADLINE_HEARTBEAT_MS` | `1000` | Upper bound on the deadline consumer's poll interval (locally enqueued deadlines fire on time regardless) |
| `DEADLINE_HEARTBEAT_MS` | `30000` | Backstop poll interval of the deadline consumer; enqueued deadlines are announced over pub/sub and wake consumers immediately, so this only bounds recovery from lost notifications or a dead worker |
| `LOGGING_CONFIG` | unset | Path to a YAML logging configuration file (see below). Unset logs DEBUG to the console |
| `CORS_ALLOW_ORIGINS` | unset | Comma-separated origins allowed for cross-origin requests, or `*` for any. CORS is disabled unless this or `CORS_ALLOW_ORIGIN_REGEX` is set |
| `CORS_ALLOW_ORIGIN_REGEX` | unset | Regex (fullmatch) additionally matched against request origins, e.g. `https://tavolo-[a-z0-9-]+\.vercel\.app` |
@@ -81,13 +85,53 @@ All configuration comes from environment variables (see `.env.example`):
| `CORS_ALLOW_CREDENTIALS` | `false` | `1`/`true`/`yes`/`on` allow cookies/credentials on cross-origin requests |
| `CORS_EXPOSE_HEADERS` | unset | Comma-separated response headers exposed to the browser |
| `CORS_MAX_AGE` | `600` | Seconds browsers may cache the preflight response |
| `OTEL_ENABLED` | `false` | `1`/`true`/`yes`/`on` enable OpenTelemetry traces and metrics (requires the `otel` extra, i.e. `pip install tavolo-app[otel]`) |
| `OTEL_ENABLED` | `false` | `1`/`true`/`yes`/`on` enable OpenTelemetry traces, metrics and logs (the Docker image bundles the `otel` extra; non-image installs need `pip install tavolo-app[otel]`) |
| `OTEL_SERVICE_NAME` | `tavolo` | `service.name` resource attribute of the exported telemetry |
| `OTEL_EXPORTER_OTLP_ENDPOINT` | unset | Base URL of an OTLP/HTTP collector (e.g. `http://localhost:4318`); unset uses the exporter default |
| `OTEL_EXPORTER_OTLP_HEADERS` | unset | Comma-separated `key=value` headers sent to the collector (e.g. authentication) |
| `OTEL_EXCLUDED_PATHS` | `/api/health` | Comma-separated paths excluded from tracing and metrics (exact matches) |
| `OTEL_METRIC_EXPORT_INTERVAL_MILLIS` | `60000` | How often the OTLP metric reader exports |
| `OTEL_METRICS_POLL_INTERVAL_SECONDS` | `15` | How often the session/game count gauges are refreshed |
| `APP_HOST` / `APP_PORT` | `0.0.0.0` / `8080` | Bind address |
¹ When no `OIDC_<n>_*` variable is set at all, a single provider named
`default` pointing at the dev mock IdP is used. Indices must be contiguous
from 0 (`OIDC_0_*`, `OIDC_1_*`, ...).
### Multiple OIDC providers
Providers are configured in indexed blocks; each becomes its own login
route `/auth/<provider>/login` with its own callback URL and session
namespace:
```sh
OIDC_0_PROVIDER=google
OIDC_0_ISSUER=https://accounts.google.com
OIDC_0_CLIENT_ID=...
OIDC_0_CLIENT_SECRET=...
OIDC_0_REDIRECT_URI=https://tavolo.example.com/auth/google/callback
OIDC_1_PROVIDER=github
OIDC_1_ISSUER=https://github.com/login/oauth
OIDC_1_CLIENT_ID=...
OIDC_1_CLIENT_SECRET=...
OIDC_1_REDIRECT_URI=https://tavolo.example.com/auth/github/callback
```
The login page reads `GET /api/auth/providers` and shows one button per
provider. `GET /auth/login` redirects to the first provider (useful for
bookmarks and scripts), and `GET /auth/logout` signs out whichever
provider the session is logged in with.
Sessions hold a single identity: logging in with another provider
replaces the previous one. Each provider's `sub` is a separate account,
so match history and Elo ratings are per provider identity (someone using
two providers has two accounts). Migrating from the pre-0.0.6 unindexed
variables: rename `OIDC_ISSUER`/`OIDC_CLIENT_ID`/... to `OIDC_0_...`, add
`OIDC_0_PROVIDER`, and register the new callback path
(`/auth/<provider>/callback`) at the provider. Sessions survive only when
the provider is named `default`; otherwise users log in once more.
## Logging
The application logs through the Python stdlib `logging` module, one
@@ -123,6 +167,37 @@ loggers:
level: INFO
```
## OpenTelemetry
With `OTEL_ENABLED` set, the app exports over OTLP/HTTP (kaya-otel plus
`tavolo.telemetry`):
- **Traces**: HTTP server spans, one span per WebSocket connection, plus
database spans for the Postgres (asyncpg, through Tortoise) and Redis
clients. Redis command values are redacted by the instrumentation
(`GET ? ?`) and asyncpg spans carry the parameterized SQL only — no
parameter values. Each HTTP/WebSocket span is tagged with the request's
session and user, so all traces of a session can be found together:
`session.id` is a truncated SHA-256 of the session id (the cookie is a
credential, so it is never exported verbatim) and `user.id` is the OIDC
`sub`, present once logged in.
- **Metrics**: the HTTP server metrics from kaya-otel, plus
`tavolo.active_users` (sessions in the session store),
`tavolo.ongoing_games` (live matches not finished yet, maintained by the
game store), and process/CPython metrics from
`opentelemetry-instrumentation-system-metrics`: `process.cpu.time`,
`process.cpu.utilization`, `process.memory.usage`,
`process.memory.virtual`, `cpython.gc.collections`,
`cpython.gc.collected_objects` and `cpython.gc.uncollectable_objects`.
- **Logs**: forwarded from the root logger at the configured console level
(OpenTelemetry's own records are filtered out to avoid export loops).
The session/game gauges are polled every
`OTEL_METRICS_POLL_INTERVAL_SECONDS` (Redis-backed stores report
cluster-wide counts, in-memory stores the process's own). Note that with
multiple workers or replicas each process reports the same global counts:
aggregate with `max`, not `sum`.
## Data model
### Redis (live games)
@@ -143,6 +218,9 @@ loggers:
on any player being connected and survive the death of any worker
(delivery is at-least-once; engines revalidate the entry's token
against the live state before firing).
- `tavolo:deadlines:events` — a pub/sub channel announcing newly enqueued
deadlines; it wakes every worker's consumer immediately, while the
heartbeat poll remains as a backstop for lost signals.
### Postgres (statistics, via Tortoise ORM + aerich migrations)
@@ -200,8 +278,11 @@ All endpoints except `/api/health`, `/api/docs`, `/api/openapi.json`,
## WebSocket protocol
Connect to `/ws/games/{game_id}` with your session cookie. Only seated
players are accepted; unauthenticated, unknown-game and non-player
connections are closed with codes `4401`, `4404` and `4403` respectively.
players are accepted; the upgrade is refused before the handshake
completes with an HTTP status mirroring `GET /api/games/{id}`: `401`
unauthenticated, `404` unknown game, `403` not seated. (Browsers report
a refused handshake as an abnormal close — there is no close code to
read.)
Server → client messages are JSON objects with a `type`:
@@ -356,6 +437,7 @@ packages/tavolo-app/ # deployable app: composition root (own pyproject)
src/tavolo/
├── app.py # session/OIDC/Tortoise/OpenAPI/Platform/DeadlineScheduler mixins
├── config.py # env -> frozen Settings
├── telemetry.py # OTLP metric/log providers, app gauges, count poller
├── aerich_config.py # aerich CLI configuration
├── logging_config.py # logging setup
└── static.py # SPA shell hosting
+4 -3
View File
@@ -10,7 +10,8 @@ from __future__ import annotations
import os
os.environ.setdefault("DATABASE_URL", "sqlite://:memory:")
os.environ.setdefault("OIDC_ISSUER", "http://localhost:8180/tavolo")
os.environ.setdefault("OIDC_CLIENT_ID", "tavolo")
os.environ.setdefault("OIDC_REDIRECT_URI", "http://localhost:8080/auth/callback")
os.environ.setdefault("OIDC_0_PROVIDER", "default")
os.environ.setdefault("OIDC_0_ISSUER", "http://localhost:8180/tavolo")
os.environ.setdefault("OIDC_0_CLIENT_ID", "tavolo")
os.environ.setdefault("OIDC_0_REDIRECT_URI", "http://localhost:8080/auth/default/callback")
os.environ.pop("REDIS_URL", None)
+7
View File
@@ -11,6 +11,12 @@ The deployable tavolo application: the composition root wiring
`ScoponeEngine` with its configured timeouts, and exposes
`tavolo.app:app` for granian.
- `config.py` — environment-driven frozen `Settings`.
- `telemetry.py` — the optional OpenTelemetry wiring beyond kaya-otel:
meter/tracer/logger providers, the
`tavolo.active_users`/`tavolo.ongoing_games` gauges (polled from the
stores), process/CPython metrics via
`opentelemetry-instrumentation-system-metrics`, Postgres/Redis query
spans via the asyncpg/redis instrumentations, and OTLP log export.
- `aerich_config.py` — Tortoise ORM configuration consumed by the aerich
CLI (migrations live in `server/migrations/`; aerich itself is
configured via the `[tool.aerich]` section of `server/pyproject.toml`).
@@ -23,6 +29,7 @@ The deployable tavolo application: the composition root wiring
.venv/bin/python -m mypy --namespace-packages --explicit-package-bases \
src/tavolo/app.py \
src/tavolo/config.py \
src/tavolo/telemetry.py \
src/tavolo/static.py \
src/tavolo/aerich_config.py \
src/tavolo/logging_config.py
+4 -1
View File
@@ -34,7 +34,10 @@ dev = [
"httpx-ws",
]
otel = [
"kaya-otel>=0.0.4",
"kaya-otel>=0.0.7",
"opentelemetry-instrumentation-asyncpg>=0.66b0",
"opentelemetry-instrumentation-redis>=0.66b0",
"opentelemetry-instrumentation-system-metrics>=0.66b0",
]
[tool.setuptools.packages.find]
+75 -37
View File
@@ -6,7 +6,7 @@ platform:
- :class:`~kaya.session.SessionMixin` (sessions persisted in Redis via
:class:`~kaya.session.redis.RedisSessionStore` when ``REDIS_URL`` is set,
otherwise an in-memory store — e.g. for tests)
- :class:`~kaya.oidc.OIDCMixin` (OIDC login)
- :class:`~kaya.oidc.OIDCProviders` (the configured OIDC providers)
- :class:`~tavolo.platform.tortoise_mixin.TortoiseMixin` (Postgres match
statistics; skipped for ``/api/health`` and the OpenAPI documentation
endpoints)
@@ -31,14 +31,13 @@ from __future__ import annotations
from importlib.metadata import version as _pkg_version
from logging import getLogger
from typing import Optional
from typing import Any, Dict, Optional
from kaya.core import KayaApp, KayaMixin
from kaya.cors import CorsMixin
from kaya.oidc import OIDCConfig, OIDCMixin
from kaya.oidc import OIDCConfig, OIDCProviders
from kaya.openapi import OpenAPIMixin
from kaya.session import InMemorySessionStore, SessionMixin, SessionStore
from kaya.session.redis import RedisSessionStore
from kaya.session import SessionMixin
from redis.asyncio import Redis
from tavolo.platform import GameRegistry, Platform, PlatformMixin
from tavolo.platform.deadlines import DeadlineScheduler, DeadlineSchedulerMixin
@@ -48,6 +47,14 @@ from tavolo.scopone import ScoponeEngine
from .config import Settings, settings
from .logging_config import configure_logging
from .telemetry import (
CountableSessionStore,
CountingInMemorySessionStore,
CountingRedisSessionStore,
otel_headers,
session_tag_hook,
telemetry_from_settings,
)
configure_logging(settings.logging_config)
log = getLogger(__name__)
@@ -73,13 +80,23 @@ def cors_mixin_from_settings(settings: Settings) -> Optional[CorsMixin]:
)
def otel_mixin_from_settings(settings: Settings) -> Optional[KayaMixin]:
def otel_mixin_from_settings(
settings: Settings,
meter_provider: Optional[Any] = None,
tracer_provider: Optional[Any] = None,
) -> Optional[KayaMixin]:
"""Build a :class:`~kaya.otel.OTelMixin` from the OTEL_* settings.
Returns ``None`` — telemetry disabled — unless ``OTEL_ENABLED`` is
truthy. kaya-otel is an optional dependency (the ``otel`` extra), so it
is imported lazily here: default installs and the test suite never need
the OpenTelemetry packages.
``meter_provider``/``tracer_provider`` are the providers built by
:func:`tavolo.telemetry.telemetry_from_settings`; injecting them lets
the custom instruments and the database instrumentations share the same
OTLP pipelines (kaya-otel then leaves their shutdown to the telemetry
mixin).
"""
if not settings.otel_enabled:
return None
@@ -90,17 +107,21 @@ def otel_mixin_from_settings(settings: Settings) -> Optional[KayaMixin]:
"OTEL_ENABLED is set but kaya-otel is not installed; "
"install tavolo-app with the 'otel' extra"
) from exc
headers = dict(
pair.split("=", 1)
for pair in (settings.otel_exporter_headers or ())
if "=" in pair
)
return OTelMixin(
service_name=settings.otel_service_name,
endpoint=settings.otel_exporter_endpoint,
headers=headers or None,
excluded_paths=settings.otel_excluded_paths,
)
kwargs: Dict[str, Any] = {
"service_name": settings.otel_service_name,
"endpoint": settings.otel_exporter_endpoint,
"headers": otel_headers(settings),
"excluded_paths": settings.otel_excluded_paths,
# Runs after the session mixin's after hook, so the session id has
# been persisted/minted and the OIDC claims are on the session.
"server_response_hook": session_tag_hook,
"websocket_close_hook": session_tag_hook,
}
if meter_provider is not None:
kwargs["meter_provider"] = meter_provider
if tracer_provider is not None:
kwargs["tracer_provider"] = tracer_provider
return OTelMixin(**kwargs)
registry = GameRegistry()
@@ -117,12 +138,12 @@ log.debug(
settings.turn_timeout_seconds,
)
session_store: SessionStore
session_store: CountableSessionStore
game_store: GameStore
if settings.redis_url is not None:
# Lazy client: no connection is opened until a session is actually
# loaded/saved, so importing this module never requires a live Redis.
session_store = RedisSessionStore(Redis.from_url(settings.redis_url))
session_store = CountingRedisSessionStore(Redis.from_url(settings.redis_url))
game_store = RedisGameStore(
Redis.from_url(settings.redis_url, decode_responses=False),
registry,
@@ -130,23 +151,31 @@ if settings.redis_url is not None:
)
log.info("using Redis stores (sessions + live games, game TTL %ds)", settings.game_ttl_seconds)
else:
session_store = InMemorySessionStore()
session_store = CountingInMemorySessionStore()
game_store = InMemoryGameStore(registry)
log.info("REDIS_URL unset: using in-memory stores (sessions + live games)")
session_mixin = SessionMixin(session_store)
oidc_mixin = OIDCMixin(
OIDCConfig(
issuer=settings.oidc_issuer,
client_id=settings.oidc_client_id,
client_secret=settings.oidc_client_secret,
redirect_uri=settings.oidc_redirect_uri,
post_login_redirect=settings.oidc_post_login_redirect,
post_logout_redirect=settings.oidc_post_logout_redirect,
fetch_userinfo=True,
),
oidc = OIDCProviders(
[
OIDCConfig(
provider=provider.provider,
issuer=provider.issuer,
client_id=provider.client_id,
client_secret=provider.client_secret,
redirect_uri=provider.redirect_uri,
post_login_redirect=provider.post_login_redirect,
post_logout_redirect=provider.post_logout_redirect,
fetch_userinfo=True,
)
for provider in settings.oidc_providers
],
session=session_mixin,
)
log.info(
"OIDC providers: %s",
", ".join(provider.provider for provider in settings.oidc_providers),
)
openapi_mixin = OpenAPIMixin(
title="tavolo",
version=_pkg_version("tavolo-app"),
@@ -171,20 +200,29 @@ platform = Platform(
registry=registry,
game_store=game_store,
scheduler=scheduler,
oidc=oidc_mixin,
oidc=oidc,
)
mixins: list[KayaMixin] = [session_mixin, oidc_mixin, tortoise_mixin, openapi_mixin,
mixins: list[KayaMixin] = [session_mixin, oidc, tortoise_mixin, openapi_mixin,
PlatformMixin(platform), DeadlineSchedulerMixin(scheduler)]
otel_mixin = otel_mixin_from_settings(settings)
if otel_mixin is not None:
# First in the list: before hooks run in registration order (after hooks
telemetry = telemetry_from_settings(settings, session_store, game_store)
if telemetry is not None:
otel_mixin = otel_mixin_from_settings(
settings,
meter_provider=telemetry.meter_provider,
tracer_provider=telemetry.tracer_provider,
)
assert otel_mixin is not None # OTEL_ENABLED was checked above
# kaya-otel first: before hooks run in registration order (after hooks
# in reverse), so the span covers session loading, OIDC handling and the
# handler itself. CORS, when enabled, is still prepended before it so
# preflight short-circuits stay untraced.
# preflight short-circuits stay untraced. The telemetry mixin carries no
# hooks but owns the metric/logger providers' polling and shutdown, so
# it must be in the list too.
mixins.insert(0, telemetry.mixin)
mixins.insert(0, otel_mixin)
log.info(
"OpenTelemetry enabled: service=%s endpoint=%s",
"OpenTelemetry enabled: service=%s endpoint=%s (traces, metrics, logs)",
settings.otel_service_name,
settings.otel_exporter_endpoint or "(OTLP default)",
)
+152 -22
View File
@@ -6,6 +6,7 @@ dataclass. No pydantic-settings, no settings module.
from __future__ import annotations
import os
import re
from dataclasses import dataclass
from typing import Optional, Tuple
from urllib.parse import quote
@@ -39,6 +40,135 @@ def _env_bool(name: str, default: bool = False) -> bool:
return value.strip().lower() in ("1", "true", "yes", "on")
# Per-provider OIDC settings are indexed with a 0-based suffix:
# OIDC_<n>_PROVIDER, OIDC_<n>_ISSUER, ... Indices must be contiguous from 0.
_OIDC_VARIABLES = (
"PROVIDER",
"ISSUER",
"CLIENT_ID",
"CLIENT_SECRET",
"REDIRECT_URI",
"POST_LOGIN_REDIRECT",
"POST_LOGOUT_REDIRECT",
)
_OIDC_ENV = re.compile(r"^OIDC_(\d+)_([A-Z0-9_]+)$")
_PROVIDER_NAME = re.compile(r"^[A-Za-z0-9_-]+$")
# The pre-0.0.6 unindexed names, detected only to point the migration out.
_LEGACY_OIDC_VARIABLES = (
"OIDC_ISSUER",
"OIDC_CLIENT_ID",
"OIDC_CLIENT_SECRET",
"OIDC_REDIRECT_URI",
"OIDC_POST_LOGIN_REDIRECT",
"OIDC_POST_LOGOUT_REDIRECT",
)
_DEFAULT_OIDC_ISSUER = "http://localhost:8180/tavolo"
_DEFAULT_OIDC_CLIENT_ID = "tavolo"
@dataclass(frozen=True)
class OidcProviderSettings:
"""One OIDC provider, parsed from the ``OIDC_<n>_*`` variables.
``provider`` names the provider in URLs (``/auth/<provider>/login``)
and namespaces its session state (``oidc:<provider>:...``); it must
match kaya-oidc's ``[A-Za-z0-9_-]+`` pattern.
"""
provider: str
issuer: str
client_id: str
client_secret: Optional[str]
redirect_uri: str
post_login_redirect: str
post_logout_redirect: str
def _oidc_providers_from_env() -> Tuple[OidcProviderSettings, ...]:
"""Parse the indexed ``OIDC_<n>_*`` provider configuration.
With no indexed variables at all, fall back to a single provider
named ``default`` pointing at the dev mock IdP, mirroring the
pre-0.0.6 defaults. The old unindexed variables are rejected with a
migration hint: they would otherwise be silently ignored.
"""
legacy = [name for name in _LEGACY_OIDC_VARIABLES if os.environ.get(name)]
if legacy:
renamed = ", ".join(
f"{name} -> OIDC_0_{name[len('OIDC_'):]}" for name in legacy
)
raise RuntimeError(
"the OIDC variables are now indexed per provider "
f"({renamed}; add OIDC_0_PROVIDER)"
)
indices: set = set()
for name in os.environ:
match = _OIDC_ENV.match(name)
if match is None:
continue
index, variable = match.groups()
if variable not in _OIDC_VARIABLES:
raise RuntimeError(
f"unknown OIDC environment variable {name!r}; expected one of "
+ ", ".join(f"OIDC_<n>_{v}" for v in _OIDC_VARIABLES)
)
indices.add(int(index))
if not indices:
return (
OidcProviderSettings(
provider="default",
issuer=_DEFAULT_OIDC_ISSUER,
client_id=_DEFAULT_OIDC_CLIENT_ID,
client_secret=None,
redirect_uri="http://localhost:8080/auth/default/callback",
post_login_redirect="/",
post_logout_redirect="/",
),
)
missing = sorted(set(range(max(indices) + 1)) - indices)
if missing:
raise RuntimeError(
"OIDC provider indices must be contiguous from 0; missing "
+ ", ".join(f"OIDC_{index}_*" for index in missing)
)
providers = []
seen = set()
for index in sorted(indices):
prefix = f"OIDC_{index}_"
provider = _env(prefix + "PROVIDER")
if _PROVIDER_NAME.match(provider) is None:
raise RuntimeError(
f"{prefix}PROVIDER must be a non-empty string of letters, "
f"digits, underscores or hyphens (got {provider!r})"
)
if provider in seen:
raise RuntimeError(f"duplicate OIDC provider name {provider!r}")
seen.add(provider)
providers.append(
OidcProviderSettings(
provider=provider,
issuer=_env(prefix + "ISSUER"),
client_id=_env(prefix + "CLIENT_ID"),
client_secret=os.environ.get(prefix + "CLIENT_SECRET") or None,
# kaya-oidc's OIDCProviders derives the callback route from
# the provider name; keep the default in sync with it.
redirect_uri=_env(
prefix + "REDIRECT_URI",
f"http://localhost:8080/auth/{provider}/callback",
),
post_login_redirect=_env(prefix + "POST_LOGIN_REDIRECT", "/"),
post_logout_redirect=_env(prefix + "POST_LOGOUT_REDIRECT", "/"),
)
)
return tuple(providers)
def _database_url_from_parts(engine: str,
user: str,
password: Optional[str],
@@ -70,15 +200,10 @@ def _database_url_from_parts(engine: str,
@dataclass(frozen=True)
class Settings:
database_url: str
oidc_issuer: str
oidc_client_id: str
oidc_client_secret: Optional[str]
oidc_redirect_uri: str
# Where the browser is sent after login/logout. In production the SPA is
# served by this app ("/"); in development point these at the trunk dev
# server (e.g. "http://localhost:8000/").
oidc_post_login_redirect: str
oidc_post_logout_redirect: str
# The configured OIDC providers, in index order. Each provider
# authenticates users independently and namespaces its session state;
# see OidcProviderSettings above.
oidc_providers: Tuple[OidcProviderSettings, ...]
app_host: str
app_port: int
redis_url: Optional[str]
@@ -97,10 +222,10 @@ class Settings:
# card for them, covering disconnects and idle players (wired into the
# ScoponeEngine; see tavolo.app).
turn_timeout_seconds: int
# Upper bound on how long the deadline consumer sleeps between polls.
# Locally enqueued deadlines wake the consumer immediately; the
# heartbeat only bounds the discovery delay for deadlines enqueued by
# other workers.
# Upper bound on how long the deadline consumer sleeps between polls
# when nothing else wakes it. Enqueued deadlines are announced over a
# pub/sub channel, so this is only the backstop recovering lost
# notifications and deadlines orphaned by a dead worker.
deadline_heartbeat_ms: int
# Path to a YAML logging configuration file (logging.config.dictConfig
# schema). Unset uses the built-in default: DEBUG to the console.
@@ -125,6 +250,10 @@ class Settings:
# Paths excluded from tracing and metrics (exact matches). Defaults to
# the health endpoint, which k8s probes would otherwise spam.
otel_excluded_paths: Tuple[str, ...]
# How often the OTLP metric reader exports (milliseconds) and how often
# the telemetry gauges refresh the session/game counts (seconds).
otel_metric_export_interval_millis: int
otel_metrics_poll_interval_seconds: int
@staticmethod
def from_env() -> "Settings":
@@ -143,12 +272,7 @@ class Settings:
# Raw DSN query string (e.g. "ssl=require"); empty = none.
options=os.environ.get("DATABASE_OPTIONS", ""),
),
oidc_issuer=_env("OIDC_ISSUER", "http://localhost:8180/tavolo"),
oidc_client_id=_env("OIDC_CLIENT_ID", "tavolo"),
oidc_client_secret=os.environ.get("OIDC_CLIENT_SECRET"),
oidc_redirect_uri=_env("OIDC_REDIRECT_URI", "http://localhost:8080/auth/callback"),
oidc_post_login_redirect=_env("OIDC_POST_LOGIN_REDIRECT", "/"),
oidc_post_logout_redirect=_env("OIDC_POST_LOGOUT_REDIRECT", "/"),
oidc_providers=_oidc_providers_from_env(),
app_host=_env("APP_HOST", "0.0.0.0"),
app_port=int(_env("APP_PORT", "8080")),
# When unset, sessions and live games fall back to in-memory
@@ -159,7 +283,7 @@ class Settings:
static_dir=_env("STATIC_DIR", "web/dist"),
hand_ack_timeout_seconds=int(_env("HAND_ACK_TIMEOUT_SECONDS", "30")),
turn_timeout_seconds=int(_env("TURN_TIMEOUT_SECONDS", "30")),
deadline_heartbeat_ms=int(_env("DEADLINE_HEARTBEAT_MS", "1000")),
deadline_heartbeat_ms=int(_env("DEADLINE_HEARTBEAT_MS", "30000")),
logging_config=os.environ.get("LOGGING_CONFIG"),
# CORS is disabled unless CORS_ALLOW_ORIGINS (a comma-separated
# list of origins, or "*" for any) or CORS_ALLOW_ORIGIN_REGEX
@@ -171,14 +295,20 @@ class Settings:
cors_allow_credentials=_env_bool("CORS_ALLOW_CREDENTIALS"),
cors_expose_headers=_env_list("CORS_EXPOSE_HEADERS"),
cors_max_age=int(_env("CORS_MAX_AGE", "600")),
# OpenTelemetry is opt-in: set OTEL_ENABLED=1 to export traces
# and metrics via OTLP/HTTP (requires the ``otel`` extra).
# OpenTelemetry is opt-in: set OTEL_ENABLED=1 to export traces,
# metrics and logs via OTLP/HTTP (requires the ``otel`` extra).
otel_enabled=_env_bool("OTEL_ENABLED"),
otel_service_name=_env("OTEL_SERVICE_NAME", "tavolo"),
otel_exporter_endpoint=os.environ.get("OTEL_EXPORTER_OTLP_ENDPOINT") or None,
# Comma-separated key=value pairs, e.g. "Authorization=Bearer x".
otel_exporter_headers=_env_list("OTEL_EXPORTER_OTLP_HEADERS"),
otel_excluded_paths=_env_list("OTEL_EXCLUDED_PATHS") or ("/api/health",),
otel_metric_export_interval_millis=int(
_env("OTEL_METRIC_EXPORT_INTERVAL_MILLIS", "60000")
),
otel_metrics_poll_interval_seconds=int(
_env("OTEL_METRICS_POLL_INTERVAL_SECONDS", "15")
),
)
@@ -2,7 +2,9 @@
Configured once at app import time (see :mod:`tavolo.app`). By default a
single console handler at DEBUG level is installed, formatting records as
``{asctime} [{levelname}] ({processName}/{threadName}) - {name} - {message}``.
``{asctime} [{levelname}] ({processName}/{threadName}) - {name} - {message}``;
``urllib3.connectionpool`` is capped at INFO so its per-request DEBUG
chatter does not drown the log.
Point the ``LOGGING_CONFIG`` environment variable at a YAML file to take
over the configuration entirely; the file follows the
:data:`logging.config.dictConfig` schema, e.g.::
@@ -55,6 +57,13 @@ DEFAULT_CONFIG = {
"level": "DEBUG",
"handlers": ["console"],
},
"loggers": {
# Every pooled HTTP request is a DEBUG record; keep the logger's
# INFO+ output, drop the per-connection noise.
"urllib3.connectionpool": {
"level": "INFO",
},
},
}
@@ -0,0 +1,375 @@
"""OpenTelemetry metrics and log export beyond kaya-otel.
kaya-otel instruments HTTP requests and websockets with traces and HTTP
metrics; this module adds what it does not cover, all behind the optional
``otel`` extra (OpenTelemetry imports are lazy, so default installs and
the test suite never need them):
* application gauges: ``tavolo.active_users`` (sessions held in the
session store) and ``tavolo.ongoing_games`` (live matches that have not
finished yet);
* process metrics through ``opentelemetry-instrumentation-system-metrics``:
CPU time/utilization, memory usage and CPython garbage-collection
counters;
* OTLP log export through a stdlib :class:`logging.Handler` attached to
the root logger, mirroring what the console sees;
* session/user tagging of request and websocket spans through
:func:`session_tag_hook` (kaya-otel's response/close hooks), so traces
can be correlated with a session (hashed id) and an OIDC user.
The store counts are asynchronous while OpenTelemetry instrument
callbacks are synchronous, so :class:`TelemetryMixin` polls the stores on
a background task and its callbacks report the cached values.
"""
from __future__ import annotations
import asyncio
import hashlib
import logging
from dataclasses import dataclass
from logging import getLogger
from typing import TYPE_CHECKING, Any, Dict, Optional, Union
from kaya.core import KayaApp, KayaMixin
from kaya.session import InMemorySessionStore, SessionStore
from kaya.session.redis import RedisSessionStore
from redis.asyncio import Redis
from tavolo.platform.auth import session_user
from tavolo.platform.store import GameStore
from .config import Settings
if TYPE_CHECKING:
from opentelemetry.sdk._logs import LoggerProvider
from opentelemetry.sdk.metrics import MeterProvider
from opentelemetry.sdk.trace import TracerProvider
log = getLogger(__name__)
# Session keys live under kaya's default prefix. Keeping it explicit here
# makes the Redis count scan exact and preserves existing sessions.
SESSION_KEY_PREFIX = "kaya:session:"
# The interpreter metrics to export. The instrumentation package can also
# emit host-level ``system.*`` metrics; those are deliberately left off:
# the application cares about its own process, and per-CPU host metrics
# add cardinality without context.
#
# ``process.cpu.utilization`` ignores its configuration value upstream
# (the callback always emits one aggregate observation), so its ``None``
# is only for readability.
PROCESS_METRICS_CONFIG: Dict[str, Optional[list]] = {
"process.cpu.time": ["user", "system"],
"process.cpu.utilization": None,
"process.memory.usage": None,
"process.memory.virtual": None,
"cpython.gc.collections": None,
"cpython.gc.collected_objects": None,
"cpython.gc.uncollectable_objects": None,
}
def otel_headers(settings: Settings) -> Optional[Dict[str, str]]:
"""Parse ``OTEL_EXPORTER_OTLP_HEADERS`` into a mapping (or ``None``)."""
headers = dict(
pair.split("=", 1)
for pair in (settings.otel_exporter_headers or ())
if "=" in pair
)
return headers or None
def _signal_endpoint(base: Optional[str], signal: str) -> Optional[str]:
"""``{base}/v1/<signal>``, or ``None`` for the exporter's own default."""
return f"{base}/{signal}" if base else None
def session_tag_hook(span: Any, carrier: Any) -> None:
"""Tag an HTTP request/websocket span with the request's session/user.
Wired into kaya-otel as its ``server_response_hook`` and
``websocket_close_hook``: those run after the session mixin's after
hook has persisted (and, at login, minted) the session, so the final
session id is visible here.
The session id is the session cookie, i.e. a credential, so only a
truncated SHA-256 of it is exported; session ids are
``secrets.token_urlsafe(32)`` (256 bits), which makes the digest
irreversible in practice. The OIDC ``sub`` is a pseudonymous
identifier already persisted in Postgres and served by the API, so it
is exported verbatim as ``user.id`` (OpenTelemetry's current end-user
attribute; ``enduser.id`` is PII-flagged and being superseded).
"""
session = getattr(carrier, "session", None)
if session is None:
return
session_id = getattr(session, "id", None)
if session_id:
span.set_attribute(
"session.id",
hashlib.sha256(session_id.encode()).hexdigest()[:16],
)
user = session_user(session)
if user is not None:
sub = user.get("sub")
if sub:
span.set_attribute("user.id", str(sub))
class _NoOtelLogs(logging.Filter):
"""Drop OpenTelemetry's own log records.
Export failures are logged by the SDK; forwarding those through the
same exporter would feed an endless retry loop.
"""
def filter(self, record: logging.LogRecord) -> bool:
return not record.name.startswith("opentelemetry")
class CountingInMemorySessionStore(InMemorySessionStore):
"""In-memory session store that can report its live session count."""
async def count(self) -> int:
now = self._clock()
return sum(
1
for session_id in self._data
if session_id not in self._expires
or self._expires[session_id] > now
)
class CountingRedisSessionStore(RedisSessionStore):
"""Redis session store that can report its live session count."""
def __init__(self, client: Redis, prefix: str = SESSION_KEY_PREFIX) -> None:
super().__init__(client, prefix=prefix)
async def count(self) -> int:
total = 0
async for _ in self._client.scan_iter(match=f"{self._prefix}*", count=500):
total += 1
return total
# The session stores the application builds: both expose ``count()`` on top
# of kaya's ``SessionStore`` interface. Game stores need no alias — the
# platform's ``GameStore`` declares ``count_ongoing()`` itself.
CountableSessionStore = Union[CountingInMemorySessionStore, CountingRedisSessionStore]
class TelemetryMixin(KayaMixin):
"""Application gauges, DB instrumentation and log export.
The meter, tracer and logger providers are built by
:func:`telemetry_from_settings` and injected: kaya-otel only shuts
down providers it created itself, so this mixin owns their shutdown
(and uninstruments the database clients it instrumented).
"""
def __init__(
self,
*,
meter_provider: "MeterProvider",
tracer_provider: "TracerProvider",
logger_provider: Optional["LoggerProvider"],
log_handler: Optional[logging.Handler],
session_store: CountableSessionStore,
game_store: GameStore,
poll_interval_seconds: float,
) -> None:
self._meter_provider = meter_provider
self._tracer_provider = tracer_provider
self._logger_provider = logger_provider
self._log_handler = log_handler
self._session_store = session_store
self._game_store = game_store
self._poll_interval = poll_interval_seconds
self._active_users: Optional[int] = None
self._ongoing_games: Optional[int] = None
self._poll_task: Optional[asyncio.Task] = None
self._instrumentor: Optional[Any] = None
self._db_instrumentors: list = []
def apply(self, app: KayaApp) -> None:
from opentelemetry.instrumentation.asyncpg import AsyncPGInstrumentor
from opentelemetry.instrumentation.redis import RedisInstrumentor
from opentelemetry.instrumentation.system_metrics import (
SystemMetricsInstrumentor,
)
from opentelemetry.metrics import CallbackOptions, Observation
instrumentor = SystemMetricsInstrumentor(config=PROCESS_METRICS_CONFIG)
self._instrumentor = instrumentor
instrumentor.instrument(meter_provider=self._meter_provider)
# Query spans for the Postgres (asyncpg, used by Tortoise) and Redis
# clients. Redis command values are redacted by the instrumentation
# (``SET ? ?``); asyncpg spans carry the parameterized SQL only.
self._db_instrumentors = [
AsyncPGInstrumentor(capture_parameters=False),
RedisInstrumentor(),
]
for db_instrumentor in self._db_instrumentors:
db_instrumentor.instrument(tracer_provider=self._tracer_provider)
meter = self._meter_provider.get_meter(__name__)
def active_users(options: CallbackOptions):
if self._active_users is not None:
yield Observation(self._active_users)
def ongoing_games(options: CallbackOptions):
if self._ongoing_games is not None:
yield Observation(self._ongoing_games)
meter.create_observable_gauge(
"tavolo.active_users",
callbacks=[active_users],
unit="{user}",
description="Sessions currently held in the session store.",
)
meter.create_observable_gauge(
"tavolo.ongoing_games",
callbacks=[ongoing_games],
unit="{game}",
description="Live games that have not finished yet.",
)
# -- lifecycle ----------------------------------------------------------
def setup(self, loop: asyncio.AbstractEventLoop) -> None:
self._poll_task = loop.create_task(self._poll())
def shutdown(self, loop: asyncio.AbstractEventLoop) -> None:
if self._poll_task is not None:
self._poll_task.cancel()
self._poll_task = None
for db_instrumentor in self._db_instrumentors:
db_instrumentor.uninstrument()
self._db_instrumentors = []
if self._log_handler is not None:
logging.getLogger().removeHandler(self._log_handler)
self._log_handler.close()
self._log_handler = None
if self._logger_provider is not None:
self._logger_provider.shutdown()
self._tracer_provider.shutdown()
self._meter_provider.shutdown()
async def _poll(self) -> None:
while True:
await self.refresh_counts()
await asyncio.sleep(self._poll_interval)
async def refresh_counts(self) -> None:
"""Refresh the cached store counts (the poll loop calls this)."""
try:
self._active_users = await self._session_store.count()
except Exception:
log.warning("active_users refresh failed", exc_info=True)
try:
self._ongoing_games = await self._game_store.count_ongoing()
except Exception:
log.warning("ongoing_games refresh failed", exc_info=True)
@dataclass
class Telemetry:
"""The providers and mixin built from the ``OTEL_*`` settings."""
meter_provider: "MeterProvider"
tracer_provider: "TracerProvider"
logger_provider: Optional["LoggerProvider"]
log_handler: Optional[logging.Handler]
mixin: TelemetryMixin
def telemetry_from_settings(
settings: Settings,
session_store: CountableSessionStore,
game_store: GameStore,
) -> Optional[Telemetry]:
"""Build the metric/trace/log providers, or ``None`` when telemetry is off.
Only called when ``OTEL_ENABLED`` is set; the OpenTelemetry imports
are lazy so the default install does not need the ``otel`` extra.
"""
if not settings.otel_enabled:
return None
try:
from opentelemetry.exporter.otlp.proto.http._log_exporter import (
OTLPLogExporter,
)
from opentelemetry.exporter.otlp.proto.http.metric_exporter import (
OTLPMetricExporter,
)
from opentelemetry.exporter.otlp.proto.http.trace_exporter import (
OTLPSpanExporter,
)
from opentelemetry.sdk._logs import LoggerProvider, LoggingHandler
from opentelemetry.sdk._logs.export import BatchLogRecordProcessor
from opentelemetry.sdk.metrics import MeterProvider
from opentelemetry.sdk.metrics.export import PeriodicExportingMetricReader
from opentelemetry.sdk.resources import Resource
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
except ImportError as exc:
raise RuntimeError(
"OTEL_ENABLED is set but the OpenTelemetry dependencies are not "
"installed; install tavolo-app with the 'otel' extra"
) from exc
endpoint = settings.otel_exporter_endpoint
headers = otel_headers(settings)
resource = Resource.create({"service.name": settings.otel_service_name})
tracer_provider = TracerProvider(resource=resource)
tracer_provider.add_span_processor(
BatchSpanProcessor(
OTLPSpanExporter(
endpoint=_signal_endpoint(endpoint, "v1/traces"),
headers=headers,
)
)
)
meter_provider = MeterProvider(
resource=resource,
metric_readers=[
PeriodicExportingMetricReader(
OTLPMetricExporter(
endpoint=_signal_endpoint(endpoint, "v1/metrics"),
headers=headers,
),
export_interval_millis=settings.otel_metric_export_interval_millis,
)
],
)
logger_provider = LoggerProvider(resource=resource)
logger_provider.add_log_record_processor(
BatchLogRecordProcessor(
OTLPLogExporter(
endpoint=_signal_endpoint(endpoint, "v1/logs"),
headers=headers,
)
)
)
handler = LoggingHandler(level=logging.NOTSET, logger_provider=logger_provider)
handler.addFilter(_NoOtelLogs())
logging.getLogger().addHandler(handler)
mixin = TelemetryMixin(
meter_provider=meter_provider,
tracer_provider=tracer_provider,
logger_provider=logger_provider,
log_handler=handler,
session_store=session_store,
game_store=game_store,
poll_interval_seconds=settings.otel_metrics_poll_interval_seconds,
)
return Telemetry(
meter_provider=meter_provider,
tracer_provider=tracer_provider,
logger_provider=logger_provider,
log_handler=handler,
mixin=mixin,
)
@@ -9,9 +9,9 @@ description = "Game-independent multiplayer game platform (lobby, live play, mat
readme = "README.md"
requires-python = ">=3.10"
dependencies = [
"kaya-core",
"kaya-core>=0.0.7",
"kaya-session",
"kaya-oidc",
"kaya-oidc>=0.0.7",
"kaya-openapi",
"tortoise-orm",
"redis",
@@ -9,7 +9,29 @@ from __future__ import annotations
from typing import Any, Callable, Mapping, Optional
from kaya.core import HttpContext, WebSocket
from kaya.oidc import OIDCMixin, OIDCUser
from kaya.oidc import OIDCProviders, OIDCUser
from kaya.session import Session
# kaya-oidc namespaces its session state per provider as
# ``oidc:{provider}:user`` (since 0.0.6), so the lookup cannot assume a
# fixed key or provider name.
_OIDC_PREFIX = "oidc:"
_USER_SUFFIX = ":user"
def session_user(session: Session) -> Optional[OIDCUser]:
"""Return the authenticated user stored in a session, if any.
The kaya-oidc mixin writes the userinfo at login under a per-provider
key, so any provider's user is accepted (the configured provider is
``default`` unless the app says otherwise).
"""
for key in session:
if key.startswith(_OIDC_PREFIX) and key.endswith(_USER_SUFFIX):
claims = session.get(key)
if isinstance(claims, Mapping):
return OIDCUser(claims)
return None
def get_ws_user(ws: WebSocket) -> Optional[OIDCUser]:
@@ -21,10 +43,7 @@ def get_ws_user(ws: WebSocket) -> Optional[OIDCUser]:
session = getattr(ws, "session", None)
if session is None:
return None
claims = session.get("oidc_user")
if not isinstance(claims, Mapping):
return None
return OIDCUser(claims)
return session_user(session)
def display_name(user: OIDCUser) -> str:
@@ -36,7 +55,16 @@ def display_name(user: OIDCUser) -> str:
return user.sub
def require_auth(oidc: OIDCMixin) -> Callable[[Callable[..., Any]], Callable[..., Any]]:
def current_user(oidc: OIDCProviders, ctx: HttpContext) -> Optional[OIDCUser]:
"""Return the authenticated user of an HTTP request, if any.
Sessions are single-identity by default, so at most one provider has
a user; if several do, the first configured provider wins.
"""
return next(iter(oidc.get_users(ctx).values()), None)
def require_auth(oidc: OIDCProviders) -> Callable[[Callable[..., Any]], Callable[..., Any]]:
"""Build a decorator gating a handler on being authenticated.
Responds ``401`` with a JSON error envelope when unauthenticated —
@@ -46,7 +74,7 @@ def require_auth(oidc: OIDCMixin) -> Callable[[Callable[..., Any]], Callable[...
def decorator(handler: Callable[..., Any]) -> Callable[..., Any]:
async def guarded(ctx: HttpContext, *args: Any, **kwargs: Any) -> None:
if oidc.get_user(ctx) is None:
if current_user(oidc, ctx) is None:
await ctx.send_bytes(
401,
b'{"error":"unauthenticated"}',
@@ -11,8 +11,11 @@ machinery; the *meaning* of each deadline belongs to the game engine:
token), and the scheduler enqueues it in the store's shared deadline
queue (a Redis sorted set in production, see
:mod:`tavolo.platform.store`); enqueueing is idempotent;
* a background consumer running on **every** worker polls the queue for
due entries and hands them to the engine's
* enqueueing a deadline announces it on a shared pub/sub channel, so
every worker's consumer wakes immediately instead of waiting for its
next poll; the consumer also wakes on a slow heartbeat, which is the
backstop recovering notifications lost to a reconnect or a worker
crash. Due entries are handed to the engine's
:meth:`~tavolo.platform.engine.GameEngine.fire_deadline` under the
per-game lock; the engine revalidates the token against the live state
and raises :class:`~tavolo.platform.errors.GameError` when the entry
@@ -65,8 +68,8 @@ class DeadlineScheduler:
"""Enqueue and fire the deadlines game engines ask for.
Holds the store and the registry so a single instance serves every
game type. One consumer task (and its wake-up event) is kept per
event loop — tests run each test on a fresh loop.
game type. One consumer task, its listener task and its wake-up
event are kept per event loop — tests run each test on a fresh loop.
``context_binder`` is an optional async callable the consumer awaits
before persisting a finished match. The consumer is a long-lived task
@@ -80,14 +83,19 @@ class DeadlineScheduler:
store: GameStore,
registry: GameRegistry,
heartbeat_ms: int = 1000,
retry_ms: int = 1000,
context_binder: Optional[Callable[[], Awaitable[None]]] = None,
) -> None:
self._store = store
self._registry = registry
self._heartbeat = heartbeat_ms / 1000
# Backoff for entries that failed to process and for listener
# reconnects; deliberately independent of the (slow) heartbeat.
self._retry_delay = retry_ms / 1000
self._context_binder = context_binder
self._consumers: Dict[asyncio.AbstractEventLoop, asyncio.Task] = {}
self._wake_events: Dict[asyncio.AbstractEventLoop, asyncio.Event] = {}
self._listeners: Dict[asyncio.AbstractEventLoop, asyncio.Task] = {}
async def sync_deadline(self, session: GameSession) -> None:
"""Enqueue the deadline the current state implies, if any.
@@ -109,6 +117,16 @@ class DeadlineScheduler:
self.ensure_consumer()
due_at = deadline.due_at.timestamp()
await self._store.add_deadline(encode(entry), due_at)
# Announce the enqueue so consumers on other workers wake up too.
# The entry is already queued, so a lost notification only delays
# their discovery to the next poll backstop.
try:
await self._store.notify_deadline()
except Exception:
log.warning(
"deadline enqueue notification failed; relying on the poll backstop",
exc_info=True,
)
wake = self._wake_events.get(asyncio.get_running_loop())
if wake is not None:
wake.set()
@@ -190,15 +208,22 @@ class DeadlineScheduler:
if old.is_closed():
self._consumers.pop(old, None)
self._wake_events.pop(old, None)
self._listeners.pop(old, None)
task = self._consumers.get(loop)
if task is None or task.done():
self._wake_events[loop] = asyncio.Event()
self._consumers[loop] = loop.create_task(self._run(loop))
listener = self._listeners.get(loop)
if listener is None or listener.done():
self._listeners[loop] = loop.create_task(self._listen(loop))
log.debug("deadline consumer started")
def stop_consumer(self, loop: asyncio.AbstractEventLoop) -> None:
task = self._consumers.pop(loop, None)
self._wake_events.pop(loop, None)
listener = self._listeners.pop(loop, None)
if listener is not None:
listener.cancel()
if task is not None:
task.cancel()
@@ -228,21 +253,48 @@ class DeadlineScheduler:
log.exception("deadline consumer: poll failed; retrying")
failed = True
if failed:
# A failed entry is still due: back off to the heartbeat
# instead of polling it again immediately.
delay = self._heartbeat
# A failed entry is still due: back off briefly instead of
# polling it again immediately. Kept independent of the
# heartbeat, which is only a slow backstop.
delay = self._retry_delay
try:
await asyncio.wait_for(wake.wait(), timeout=delay)
except asyncio.TimeoutError:
pass
async def _listen(self, loop: asyncio.AbstractEventLoop) -> None:
"""Wake the consumer on deadline-enqueue notifications.
The subscription is best effort: a dropped connection or a lost
message is recovered by the consumer's heartbeat backstop, so
this task only has to reconnect and keep forwarding signals.
"""
while True:
try:
async with self._store.subscribe_deadlines() as events:
async for _ in events:
wake = self._wake_events.get(loop)
if wake is None:
# Shut down: the consumer is gone.
return
wake.set()
except asyncio.CancelledError:
raise
except Exception:
log.exception("deadline listener failed; reconnecting")
try:
await asyncio.sleep(self._retry_delay)
except asyncio.CancelledError:
raise
class DeadlineSchedulerMixin(KayaMixin):
"""Run the deadline consumer for the whole app lifetime.
Every worker (and every pod) runs the same consumer; coordination
happens exclusively through the shared deadline queue and the
per-game locks, so any worker may fire any game's deadline.
Every worker (and every pod) runs the same consumer and listener;
coordination happens exclusively through the shared deadline queue,
its notification channel and the per-game locks, so any worker may
fire any game's deadline.
"""
def __init__(self, scheduler: DeadlineScheduler) -> None:
@@ -2,7 +2,7 @@
``Platform`` bundles the collaborators every platform endpoint needs —
the game registry, the live-game store, the deadline scheduler and the
OIDC mixin used for authentication — so route and websocket modules
OIDC providers used for authentication — so route and websocket modules
never reach for module-global state. :class:`PlatformMixin` is a plain
:class:`~kaya.core.KayaMixin`: applying it registers the lobby, stats
and health HTTP routes plus the live-play websocket endpoint, and the
@@ -14,7 +14,7 @@ from __future__ import annotations
from dataclasses import dataclass
from kaya.core import KayaApp, KayaMixin
from kaya.oidc import OIDCMixin
from kaya.oidc import OIDCProviders
from .deadlines import DeadlineScheduler
from .registry import GameRegistry
@@ -28,7 +28,7 @@ class Platform:
registry: GameRegistry
game_store: GameStore
scheduler: DeadlineScheduler
oidc: OIDCMixin
oidc: OIDCProviders
class PlatformMixin(KayaMixin):
@@ -41,9 +41,10 @@ class PlatformMixin(KayaMixin):
# Imported here so module import order stays acyclic: the route
# modules reference Platform only for typing.
from . import ws
from .routes import games, health, me, stats
from .routes import auth, games, health, me, stats
health.register(app, self.platform)
auth.register(app, self.platform)
me.register(app, self.platform)
games.register(app, self.platform)
stats.register(app, self.platform)
@@ -15,12 +15,15 @@ from __future__ import annotations
import base64
import json
from dataclasses import dataclass
from typing import Any, Dict, List, Optional, Tuple
from typing import TYPE_CHECKING, Any, Dict, List, Optional, Tuple
from tortoise.queryset import QuerySet
from .http import extract_query_params
if TYPE_CHECKING:
from tortoise.models import Model
DEFAULT_LIMIT = 20
MAX_LIMIT = 100
MIN_LIMIT = 1
@@ -81,6 +84,28 @@ def parse_cursor_params(query_string: str) -> Cursor:
Sort = List[Tuple[str, str]]
def _coerce_sort_values(
model: "type[Model]", sort: Sort, after: Dict[str, Any]
) -> Dict[str, Any]:
"""Restore the Python type of every decoded cursor value.
The cursor is JSON, so field values travel as strings (datetimes via
``default=str``). asyncpg rejects strings where a ``timestamp``
parameter is expected, and sqlite only tolerates them by accident, so
each value is run through its model field's ``to_python_value`` before
it reaches the keyset filter.
"""
fields = model._meta.fields_map
coerced: Dict[str, Any] = {}
for field, _ in sort:
value = after.get(field)
field_obj = fields.get(field)
if value is not None and field_obj is not None:
value = field_obj.to_python_value(value)
coerced[field] = value
return coerced
def _keyset_where(sort: Sort, after: Dict[str, Any]):
"""Build a Tortoise ``Q`` filter from a cursor."""
from tortoise.queryset import Q # local to keep import edge narrow
@@ -116,7 +141,8 @@ async def paginate(
order_by.append(field if direction == "ASC" else f"-{field}")
qs = queryset.order_by(*order_by)
if cursor.after:
qs = qs.filter(_keyset_where(sort, cursor.after))
after = _coerce_sort_values(queryset.model, sort, cursor.after)
qs = qs.filter(_keyset_where(sort, after))
rows = await qs.limit(cursor.limit + 1)
if len(rows) <= cursor.limit:
return rows, None
@@ -0,0 +1,62 @@
"""Authentication plumbing for the single-page app.
kaya-oidc registers one login/callback/logout route per provider
(``/auth/<provider>/...``); this module adds the provider-agnostic
pieces:
* ``GET /api/auth/providers`` lists the configured providers with their
login URLs, so the SPA can offer a choice without hardcoding names;
* ``GET /auth/login`` redirects to the first configured provider;
* ``GET /auth/logout`` redirects to the provider this session is logged
in with (or the SPA root when not logged in), so the UI can keep a
single logout link.
"""
from __future__ import annotations
from typing import TYPE_CHECKING
from kaya.core import HttpContext, KayaApp
from kaya.openapi import operation
from ..http import send_json
if TYPE_CHECKING:
from ..mixin import Platform
def _login_url(platform: "Platform", provider: str) -> str:
return f"{platform.oidc.base_path}/{provider}/login"
def register(app: KayaApp, platform: "Platform") -> None:
providers = list(platform.oidc.providers)
@app.GET("/api/auth/providers")
@operation(summary="Configured login providers",
description="Names and login URLs of the configured OIDC "
"providers, in configuration order.",
tags=["auth"],
responses={200: {"description": "The configured providers"}})
async def auth_providers(ctx: HttpContext) -> None:
await send_json(
ctx,
200,
{
"providers": [
{"name": name, "login_url": _login_url(platform, name)}
for name in providers
]
},
)
@app.GET("/auth/login")
async def login(ctx: HttpContext) -> None:
await ctx.send_empty(302, {"Location": _login_url(platform, providers[0])})
@app.GET("/auth/logout")
async def logout(ctx: HttpContext) -> None:
logged_in = platform.oidc.logged_in_providers(ctx)
location = (
f"{platform.oidc.base_path}/{logged_in[0]}/logout" if logged_in else "/"
)
await ctx.send_empty(302, {"Location": location})
@@ -21,7 +21,7 @@ from typing import TYPE_CHECKING, Any, Dict
from kaya.core import HttpContext, KayaApp
from kaya.openapi import operation
from ..auth import display_name, require_auth
from ..auth import current_user, display_name, require_auth
from ..engine import GameSession, Seat
from ..errors import GameError
from ..http import (
@@ -183,7 +183,7 @@ def register(app: KayaApp, platform: "Platform") -> None:
await send_error(ctx, 400, "options must be an object")
return
user = platform.oidc.get_user(ctx)
user = current_user(platform.oidc, ctx)
assert user is not None # enforced by @require_auth
session = GameSession(
id=str(uuid.uuid4()),
@@ -244,7 +244,7 @@ def register(app: KayaApp, platform: "Platform") -> None:
await send_error(ctx, 400, "code is required")
return
user = platform.oidc.get_user(ctx)
user = current_user(platform.oidc, ctx)
assert user is not None
existing = await platform.game_store.find_by_code(code)
if existing is None:
@@ -325,7 +325,7 @@ def register(app: KayaApp, platform: "Platform") -> None:
if session is None:
await send_error(ctx, 404, "game not found")
return
user = platform.oidc.get_user(ctx)
user = current_user(platform.oidc, ctx)
assert user is not None
if not session.seated(user.sub):
await send_error(ctx, 403, "forbidden")
@@ -6,7 +6,7 @@ from typing import TYPE_CHECKING
from kaya.core import HttpContext, KayaApp
from kaya.openapi import operation
from ..auth import display_name, require_auth
from ..auth import current_user, display_name, require_auth
from ..http import send_json
if TYPE_CHECKING:
@@ -25,6 +25,6 @@ def register(app: KayaApp, platform: "Platform") -> None:
})
@require_auth(platform.oidc)
async def me(ctx: HttpContext) -> None:
user = platform.oidc.get_user(ctx)
user = current_user(platform.oidc, ctx)
assert user is not None # enforced by @require_auth
await send_json(ctx, 200, {"sub": user.sub, "name": display_name(user)})
@@ -8,12 +8,14 @@ verbatim through the ``result`` JSON column.
"""
from __future__ import annotations
from collections import defaultdict
from typing import TYPE_CHECKING, Any, Dict, List, Optional, Tuple
from uuid import UUID
from kaya.core import HttpContext, KayaApp
from kaya.openapi import operation
from ..auth import require_auth
from ..auth import current_user, require_auth
from ..elo import INITIAL_RATING
from ..http import extract_query_params, send_error, send_json
from ..models import Match, MatchPlayer, PlayerRating
@@ -48,8 +50,9 @@ def _parse_game_type(
return game_type, None
async def _serialize_match(match: Match, viewer: str) -> Dict[str, Any]:
participants = await MatchPlayer.filter(match_id=match.id).order_by("seat")
def _serialize_match(
match: Match, viewer: str, participants: List[MatchPlayer]
) -> Dict[str, Any]:
return {
"id": str(match.id),
"game_type": match.game_type,
@@ -101,7 +104,7 @@ def register(app: KayaApp, platform: "Platform") -> None:
if error is not None:
await send_error(ctx, 400, error)
return
user = platform.oidc.get_user(ctx)
user = current_user(platform.oidc, ctx)
assert user is not None
queryset = Match.filter(players__user_sub=user.sub).distinct()
if game_type is not None:
@@ -109,7 +112,21 @@ def register(app: KayaApp, platform: "Platform") -> None:
matches, next_cursor = await paginate(
queryset, [("finished_at", "DESC"), ("id", "ASC")], cursor
)
results = [await _serialize_match(m, user.sub) for m in matches]
# One participants query for the whole page instead of one per
# match; the grouping keeps each match's players seat-ordered.
by_match: Dict[UUID, List[MatchPlayer]] = defaultdict(list)
if matches:
participants = (
await MatchPlayer.filter(match_id__in=[m.id for m in matches])
.order_by("match_id", "seat")
)
for player in participants:
# Tortoise sets the FK's raw id on instances but does not
# declare it on the model class, hence the getattr.
by_match[getattr(player, "match_id")].append(player)
results = [
_serialize_match(m, user.sub, by_match.get(m.id, [])) for m in matches
]
await send_json(ctx, 200, {"results": results, "next_cursor": next_cursor})
@app.GET("/api/leaderboard")
@@ -176,7 +193,7 @@ def register(app: KayaApp, platform: "Platform") -> None:
})
@require_auth(platform.oidc)
async def my_ratings(ctx: HttpContext) -> None:
user = platform.oidc.get_user(ctx)
user = current_user(platform.oidc, ctx)
assert user is not None
rows = await PlayerRating.filter(user_sub=user.sub).order_by("game_type")
await send_json(ctx, 200, {
@@ -26,8 +26,9 @@ signal (never the state) means updated state reaches connections on
every worker without leaking hidden information into the channel.
Timeouts are driven by a shared delayed-deadline queue: producers
enqueue an opaque ``member`` string with a due timestamp, and a consumer
on every worker polls for due entries (see
enqueue an opaque ``member`` string with a due timestamp and announce it
on a shared "deadline enqueued" channel, and a consumer on every worker
wakes on that signal to process due entries (see
:mod:`tavolo.platform.deadlines`). Delivery is at-least-once — entries
are removed only after they are processed — so a worker dying
mid-processing cannot lose a deadline; engines revalidate entries
@@ -38,6 +39,7 @@ from __future__ import annotations
import asyncio
import contextlib
import json
import time
from abc import ABC, abstractmethod
from datetime import datetime
from logging import getLogger
@@ -54,6 +56,13 @@ GAME_KEY_PREFIX = "tavolo:game:"
CODE_KEY_PREFIX = "tavolo:code:"
CHANNEL_PREFIX = "tavolo:game:"
DEADLINES_KEY = "tavolo:deadlines"
# Channel announcing newly enqueued deadline entries, so every worker's
# consumer can wake immediately instead of polling on its heartbeat.
DEADLINES_CHANNEL = "tavolo:deadlines:events"
# Sorted set of running games (score = expiry timestamp), maintained by the
# stores so the telemetry gauge can count ongoing matches without loading
# every session. Finished games are removed on their final save.
ONGOING_GAMES_KEY = "tavolo:games:ongoing"
# Sentinel pushed into in-memory subscriber queues to signal a change.
_BUMP = b"update"
@@ -120,6 +129,14 @@ class GameStore(ABC):
async def find_by_code(self, code: str) -> Optional[GameSession]:
"""Return the live session for a join ``code`` or ``None``."""
@abstractmethod
async def count_ongoing(self) -> int:
"""Return the number of live games that have not finished yet.
Lobbies and matches in progress count; matches that are over but
still retained for the final-board view do not.
"""
@abstractmethod
def lock(self, game_id: str) -> AsyncContextManager[None]:
"""Async context manager serializing mutations of one game."""
@@ -152,6 +169,18 @@ class GameStore(ABC):
async def remove_deadline(self, member: str) -> None:
"""Remove ``member`` from the queue; a no-op when absent."""
@abstractmethod
async def notify_deadline(self) -> None:
"""Wake every worker's deadline consumer (best effort).
Notification is advisory: consumers stay correct when a signal
is lost, because their poll backstop recovers it.
"""
@abstractmethod
def subscribe_deadlines(self) -> AsyncContextManager[AsyncIterator[None]]:
"""Async context manager yielding deadline-enqueue signals."""
def _channel(game_id: str) -> str:
return f"{CHANNEL_PREFIX}{game_id}:events"
@@ -181,9 +210,15 @@ class RedisGameStore(GameStore):
async def save(self, session: GameSession) -> None:
payload = json.dumps(session_to_json(session, self._registry))
engine = self._registry.require(session.game_type)
async with self._redis.pipeline(transaction=True) as pipe:
pipe.set(f"{GAME_KEY_PREFIX}{session.id}", payload, ex=self._ttl)
pipe.set(f"{CODE_KEY_PREFIX}{session.join_code}", session.id, ex=self._ttl)
if engine.is_finished(session):
pipe.zrem(ONGOING_GAMES_KEY, session.id)
else:
pipe.zadd(ONGOING_GAMES_KEY, {session.id: time.time() + self._ttl})
pipe.expire(ONGOING_GAMES_KEY, self._ttl)
await pipe.execute()
log.debug("redis save %s (game %s, ttl %ds)", session.id, session.game_type, self._ttl)
@@ -195,6 +230,17 @@ class RedisGameStore(GameStore):
game_id = game_id.decode("utf-8")
return await self.load(str(game_id))
async def count_ongoing(self) -> int:
# Prune entries whose game has expired since the last poll, then
# count what is left. Both commands run in one transaction so the
# count cannot see a half-pruned set.
now = time.time()
async with self._redis.pipeline(transaction=True) as pipe:
pipe.zremrangebyscore(ONGOING_GAMES_KEY, "-inf", now)
pipe.zcard(ONGOING_GAMES_KEY)
_, count = await pipe.execute()
return int(count)
@contextlib.asynccontextmanager
async def subscribe(self, game_id: str) -> AsyncIterator[AsyncIterator[None]]:
pubsub = self._redis.pubsub()
@@ -214,10 +260,17 @@ class RedisGameStore(GameStore):
await self._redis.zadd(DEADLINES_KEY, {member: due_at})
async def due_deadlines(self, now: float, limit: int = 32) -> List[str]:
# ``ZRANGE ... BYSCORE`` is the non-deprecated spelling of the
# old ``ZRANGEBYSCORE``.
members = cast(
list,
await self._redis.zrangebyscore(
DEADLINES_KEY, "-inf", now, start=0, num=limit
await self._redis.zrange(
DEADLINES_KEY,
"-inf",
now,
byscore=True,
offset=0,
num=limit,
),
)
return [m.decode("utf-8") if isinstance(m, bytes) else m for m in members]
@@ -229,6 +282,20 @@ class RedisGameStore(GameStore):
async def remove_deadline(self, member: str) -> None:
await self._redis.zrem(DEADLINES_KEY, member)
async def notify_deadline(self) -> None:
await self._redis.publish(DEADLINES_CHANNEL, "deadline")
@contextlib.asynccontextmanager
async def subscribe_deadlines(self) -> AsyncIterator[AsyncIterator[None]]:
pubsub = self._redis.pubsub()
await pubsub.subscribe(DEADLINES_CHANNEL)
try:
yield _redis_events(pubsub)
finally:
with contextlib.suppress(Exception):
await pubsub.unsubscribe(DEADLINES_CHANNEL)
await pubsub.aclose()
async def _redis_events(pubsub) -> AsyncIterator[None]:
async for message in pubsub.listen():
@@ -243,9 +310,11 @@ class InMemoryGameStore(GameStore):
self._registry = registry
self._games: Dict[str, str] = {}
self._codes: Dict[str, str] = {}
self._ongoing: Set[str] = set()
self._locks: Dict[str, asyncio.Lock] = {}
self._subscribers: Dict[str, Set[asyncio.Queue]] = {}
self._deadlines: Dict[str, float] = {}
self._deadline_subscribers: Set[asyncio.Queue] = set()
def _lock_for(self, game_id: str) -> asyncio.Lock:
lock = self._locks.get(game_id)
@@ -268,6 +337,11 @@ class InMemoryGameStore(GameStore):
async def save(self, session: GameSession) -> None:
self._games[session.id] = json.dumps(session_to_json(session, self._registry))
self._codes[session.join_code] = session.id
engine = self._registry.require(session.game_type)
if engine.is_finished(session):
self._ongoing.discard(session.id)
else:
self._ongoing.add(session.id)
async def find_by_code(self, code: str) -> Optional[GameSession]:
game_id = self._codes.get(code.upper())
@@ -275,6 +349,9 @@ class InMemoryGameStore(GameStore):
return None
return await self.load(game_id)
async def count_ongoing(self) -> int:
return len(self._ongoing)
@contextlib.asynccontextmanager
async def subscribe(self, game_id: str) -> AsyncIterator[AsyncIterator[None]]:
queue: asyncio.Queue = asyncio.Queue()
@@ -306,6 +383,19 @@ class InMemoryGameStore(GameStore):
async def remove_deadline(self, member: str) -> None:
self._deadlines.pop(member, None)
async def notify_deadline(self) -> None:
for queue in list(self._deadline_subscribers):
queue.put_nowait(_BUMP)
@contextlib.asynccontextmanager
async def subscribe_deadlines(self) -> AsyncIterator[AsyncIterator[None]]:
queue: asyncio.Queue = asyncio.Queue()
self._deadline_subscribers.add(queue)
try:
yield _queue_events(queue)
finally:
self._deadline_subscribers.discard(queue)
async def _queue_events(queue: asyncio.Queue) -> AsyncIterator[None]:
while True:
@@ -50,7 +50,7 @@ from contextlib import suppress
from logging import getLogger
from typing import TYPE_CHECKING, Any, Awaitable, Callable, Dict
from kaya.core import KayaApp, WebSocket
from kaya.core import KayaApp, WebSocket, WebSocketDisconnect
from . import auth
from .engine import GameSession
@@ -68,6 +68,21 @@ def _error(message: str, code: str = "invalid") -> Dict[str, Any]:
return {"type": "error", "code": code, "message": message}
async def _reject(ws: WebSocket, status: int) -> None:
"""Refuse a websocket upgrade with the HTTP ``status``.
Rejections happen before :meth:`WebSocket.accept`, so ``close()``
carries the HTTP status of the refused upgrade rather than a
WebSocket close code: ASGI servers answer ``403`` per the spec, and
RSGI (granian) uses the value verbatim as the response status.
Passing a 4xxx close code here would panic granian's RSGI layer on
the invalid status — and since granian is built with
``panic = "abort"``, that kills the worker and takes the whole app
down.
"""
await ws.close(status)
def _state_message(platform: "Platform", session: GameSession, sub: str) -> Dict[str, Any]:
engine = platform.registry.require(session.game_type)
return {
@@ -89,17 +104,17 @@ def register(app: KayaApp, platform: "Platform") -> None:
user = auth.get_ws_user(ws)
if user is None:
log.debug("websocket %s rejected: no authenticated user", game_id)
await ws.close(4401)
await _reject(ws, 401)
return
session = await platform.game_store.load(game_id)
if session is None:
log.debug("websocket rejected: unknown game %s", game_id)
await ws.close(4404)
await _reject(ws, 404)
return
if not session.seated(user.sub):
log.debug("websocket %s rejected: %s is not seated", game_id, user.sub)
await ws.close(4403)
await _reject(ws, 403)
return
await ws.accept()
@@ -109,7 +124,13 @@ def register(app: KayaApp, platform: "Platform") -> None:
async def send(payload: Dict[str, Any]) -> None:
async with send_lock:
await ws.send_text(json.dumps(payload))
try:
await ws.send_text(json.dumps(payload))
except WebSocketDisconnect:
# The peer went away without a close handshake: kaya
# raises this from the transport, and the receive loop
# ends momentarily.
pass
await send(_state_message(platform, session, user.sub))
# Backstop: make sure the current deadline is queued even if its
@@ -8,7 +8,7 @@ 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, a dummy OIDC mixin (patched per test) and
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.
"""
@@ -22,7 +22,7 @@ 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, OIDCMixin, OIDCUser
from kaya.oidc import OIDCConfig, OIDCProviders, OIDCUser
from kaya.openapi import OpenAPIMixin
from kaya.session import InMemorySessionStore, SessionMixin
@@ -220,20 +220,26 @@ class DummyEngine(GameEngine):
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_mixin = OIDCMixin(
OIDCConfig(
issuer="http://localhost:8180/tavolo",
client_id="tavolo",
client_secret=None,
redirect_uri="http://localhost:8080/auth/callback",
post_login_redirect="/",
post_logout_redirect="/",
),
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(
@@ -251,19 +257,19 @@ def make_platform(
scheduler = DeadlineScheduler(
game_store,
registry,
heartbeat_ms=50,
heartbeat_ms=heartbeat_ms,
context_binder=tortoise_mixin.ensure_context,
)
platform = Platform(
registry=registry,
game_store=game_store,
scheduler=scheduler,
oidc=oidc_mixin,
oidc=oidc,
)
app = KayaApp(
mixins=[
session_mixin,
oidc_mixin,
oidc,
tortoise_mixin,
openapi_mixin,
PlatformMixin(platform),
@@ -319,10 +325,12 @@ def make_user(sub: str, name: Optional[str] = None) -> OIDCUser:
@contextlib.contextmanager
def oidc_user(oidc: OIDCMixin, sub: str, name: Optional[str] = None) -> Iterator[OIDCUser]:
"""Context manager: patch ``oidc.get_user`` to return this user."""
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)
patcher = _mock.patch.object(oidc, "get_user", return_value=user)
provider = next(iter(oidc.providers))
patcher = _mock.patch.object(oidc, "get_users", return_value={provider: user})
patcher.start()
try:
yield user
@@ -0,0 +1,152 @@
"""Tests for the auth helpers' OIDC user lookup and gating.
kaya-oidc (0.0.6) namespaces session state per provider as
``oidc:{provider}:user`` and tavolo wires an :class:`OIDCProviders`; these
tests pin both contracts so a kaya upgrade or a provider change cannot
silently break HTTP or websocket authentication.
"""
from __future__ import annotations
import unittest
from types import SimpleNamespace
from typing import Any
from kaya.oidc import OIDCConfig, OIDCProviders
from kaya.session import InMemorySessionStore, Session, SessionMixin
from helpers import async_test
from tavolo.platform.auth import current_user, get_ws_user, require_auth, session_user
def _providers(*names: str) -> OIDCProviders:
session = SessionMixin(InMemorySessionStore())
return OIDCProviders(
[
OIDCConfig(
provider=name,
issuer=f"http://localhost:8180/{name}",
client_id="tavolo",
redirect_uri=f"http://localhost:8080/auth/{name}/callback",
)
for name in names
],
session=session,
)
def _ctx(data: Any) -> Any:
return SimpleNamespace(session=Session("sid", data))
def _ws(session: Any) -> Any:
return SimpleNamespace(session=session)
class SessionUserTest(unittest.TestCase):
def test_finds_default_provider_user(self) -> None:
user = session_user(Session("sid", {"oidc:default:user": {"sub": "alice"}}))
assert user is not None
self.assertEqual("alice", user.sub)
def test_finds_user_of_any_provider(self) -> None:
user = session_user(Session("sid", {"oidc:google:user": {"sub": "alice"}}))
assert user is not None
self.assertEqual("alice", user.sub)
def test_ignores_flow_state_and_tokens(self) -> None:
session = Session(
"sid",
{
"oidc:default:state": "state",
"oidc:default:id_token": "token",
"oidc:default:access_token": "token",
},
)
self.assertIsNone(session_user(session))
def test_ignores_legacy_unprefixed_user_key(self) -> None:
session = Session("sid", {"oidc_user": {"sub": "alice"}})
self.assertIsNone(session_user(session))
def test_ignores_malformed_user_claims(self) -> None:
session = Session("sid", {"oidc:default:user": "alice"})
self.assertIsNone(session_user(session))
class GetWsUserTest(unittest.TestCase):
def test_returns_user_from_provider_namespaced_session(self) -> None:
session = Session(
"sid",
{"oidc:default:user": {"sub": "alice", "preferred_username": "Alice"}},
)
user = get_ws_user(_ws(session))
assert user is not None
self.assertEqual("alice", user.sub)
def test_missing_session_returns_none(self) -> None:
self.assertIsNone(get_ws_user(SimpleNamespace()))
class CurrentUserTest(unittest.TestCase):
def test_anonymous_session_has_no_user(self) -> None:
self.assertIsNone(current_user(_providers("default"), _ctx({})))
def test_returns_the_authenticated_provider_user(self) -> None:
user = current_user(
_providers("default", "google"),
_ctx({"oidc:google:user": {"sub": "alice"}}),
)
assert user is not None
self.assertEqual("alice", user.sub)
def test_first_configured_provider_wins(self) -> None:
# Single-identity sessions never hold two users; this only pins the
# tie-break should they ever coexist.
user = current_user(
_providers("default", "google"),
_ctx(
{
"oidc:default:user": {"sub": "alice"},
"oidc:google:user": {"sub": "bob"},
}
),
)
assert user is not None
self.assertEqual("alice", user.sub)
class RequireAuthTest(unittest.TestCase):
@async_test
async def test_rejects_anonymous_requests(self) -> None:
oidc = _providers("default")
ctx = _ctx({})
sent = {}
async def send_bytes(*args: Any) -> None:
sent["status"] = args[0]
ctx.send_bytes = send_bytes
handled = []
@require_auth(oidc)
async def handler(_ctx: Any) -> None:
handled.append(True)
await handler(ctx)
self.assertEqual(401, sent["status"])
self.assertEqual([], handled)
@async_test
async def test_allows_authenticated_requests(self) -> None:
oidc = _providers("default")
ctx = _ctx({"oidc:default:user": {"sub": "alice"}})
handled = []
@require_auth(oidc)
async def handler(request_ctx: Any) -> None:
user = current_user(oidc, request_ctx)
assert user is not None
handled.append(user.sub)
await handler(ctx)
self.assertEqual(["alice"], handled)
@@ -112,6 +112,38 @@ async def _finished(store, game_id: str) -> Optional[GameSession]:
return None
class NotificationTest(unittest.TestCase):
"""Deadline-enqueue notifications wake consumers on other workers."""
@async_test
async def test_notification_wakes_a_long_heartbeat_consumer(self) -> None:
# The consumer's heartbeat is a minute away and nothing sets its
# local wake event: only the pub/sub notification can deliver the
# deadline in time. This is the cross-worker path.
_, platform, _ = make_platform(heartbeat_ms=60_000)
store = platform.game_store
scheduler = platform.scheduler
session = _started_session(deadline_in_seconds=0.1)
await store.save(session)
scheduler.ensure_consumer()
# Let the listener task subscribe before announcing.
await asyncio.sleep(0.05)
engine = platform.registry.require(session.game_type)
deadline = engine.next_deadline(session)
assert deadline is not None
member = encode({
"game_id": session.id,
"kind": deadline.kind,
"token": deadline.token,
})
await store.add_deadline(member, deadline.due_at.timestamp())
await store.notify_deadline()
result = await _wait_for(lambda: _plays_is(store, session.id, 1))
self.assertIsNotNone(result, "notification did not wake the consumer")
class ProcessDueTest(unittest.TestCase):
"""Direct ``process_due`` behaviour: engine revalidation and idempotency."""
@@ -0,0 +1,81 @@
"""Keyset pagination tests: cursor values are restored to Python types.
Cursors are JSON, so a cursor carrying a datetime reaches the next
request as a string. Postgres/asyncpg rejects strings for timestamp
parameters (sqlite silently coerces them), so ``paginate`` must run the
decoded values through their model fields before building the WHERE
clause.
"""
from __future__ import annotations
import unittest
import uuid
from datetime import datetime, timezone
from typing import Any, List
from tavolo.platform.models import Match
from tavolo.platform.pagination import Cursor, decode_cursor, encode_cursor, paginate
from helpers import async_test
SORT = [("finished_at", "DESC"), ("id", "ASC")]
class _CapturingQuerySet:
"""Minimal stand-in for a Tortoise QuerySet, recording its filter."""
model = Match
def __init__(self) -> None:
self.queryset_filter: Any = None
def order_by(self, *orderings: str) -> "_CapturingQuerySet":
return self
def filter(self, q: Any) -> "_CapturingQuerySet":
self.queryset_filter = q
return self
async def limit(self, count: int) -> List[Any]:
return []
def _flatten_q(q: Any) -> List[Any]:
nodes = [q]
for child in q.children:
nodes.extend(_flatten_q(child))
return nodes
def _filter_values(q: Any, key: str) -> List[Any]:
return [node.filters[key] for node in _flatten_q(q) if key in node.filters]
class PaginateCursorTypesTest(unittest.TestCase):
@async_test
async def test_cursor_values_are_restored_before_filtering(self) -> None:
match_id = uuid.uuid4()
finished_at = datetime(2026, 9, 19, 5, 2, 50, 547904, tzinfo=timezone.utc)
# Round-trip through the wire format, exactly as a request does.
after = decode_cursor(
encode_cursor({"finished_at": finished_at, "id": match_id})
)
assert after is not None
qs = _CapturingQuerySet()
await paginate(qs, SORT, Cursor(limit=1, after=after))
assert qs.queryset_filter is not None
for key in ("finished_at", "finished_at__lt"):
values = _filter_values(qs.queryset_filter, key)
self.assertEqual(1, len(values))
self.assertIsInstance(values[0], datetime)
self.assertEqual(finished_at, values[0])
id_values = _filter_values(qs.queryset_filter, "id__gt")
self.assertEqual(1, len(id_values))
self.assertIsInstance(id_values[0], uuid.UUID)
self.assertEqual(match_id, id_values[0])
if __name__ == "__main__":
unittest.main()
@@ -0,0 +1,67 @@
"""Tests for the provider-agnostic auth routes.
``GET /api/auth/providers`` feeds the SPA's login buttons; ``/auth/login``
and ``/auth/logout`` are the generic aliases that redirect to the right
per-provider kaya route.
"""
from __future__ import annotations
import unittest
from unittest import mock
from httpx import ASGITransport, AsyncClient
from helpers import async_test, make_platform
class AuthProvidersRouteTest(unittest.TestCase):
@async_test
async def test_lists_providers_in_configuration_order(self) -> None:
app, _platform, _tortoise = make_platform(oidc_providers=("default", "google"))
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport, base_url="http://127.0.0.1") as client:
response = await client.get("/api/auth/providers")
self.assertEqual(200, response.status_code)
self.assertEqual(
{
"providers": [
{"name": "default", "login_url": "/auth/default/login"},
{"name": "google", "login_url": "/auth/google/login"},
]
},
response.json(),
)
class GenericAuthRedirectTest(unittest.TestCase):
@async_test
async def test_login_redirects_to_the_first_provider(self) -> None:
app, _platform, _tortoise = make_platform(oidc_providers=("google", "github"))
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport, base_url="http://127.0.0.1") as client:
response = await client.get("/auth/login", follow_redirects=False)
self.assertEqual(302, response.status_code)
self.assertEqual("/auth/google/login", response.headers["location"])
@async_test
async def test_logout_redirects_to_the_logged_in_provider(self) -> None:
app, platform, _tortoise = make_platform()
with mock.patch.object(
platform.oidc, "logged_in_providers", return_value=["default"]
):
transport = ASGITransport(app=app)
async with AsyncClient(
transport=transport, base_url="http://127.0.0.1"
) as client:
response = await client.get("/auth/logout", follow_redirects=False)
self.assertEqual(302, response.status_code)
self.assertEqual("/auth/default/logout", response.headers["location"])
@async_test
async def test_logout_without_a_session_redirects_home(self) -> None:
app, _platform, _tortoise = make_platform()
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport, base_url="http://127.0.0.1") as client:
response = await client.get("/auth/logout", follow_redirects=False)
self.assertEqual(302, response.status_code)
self.assertEqual("/", response.headers["location"])
@@ -4,6 +4,7 @@ from __future__ import annotations
import unittest
import uuid
from datetime import datetime, timezone
from unittest import mock
from httpx import ASGITransport, AsyncClient
@@ -197,6 +198,28 @@ class StatsRouteTest(unittest.TestCase):
second.json()["results"][0]["id"],
)
@async_test
async def test_my_matches_fetches_participants_in_one_query(self) -> None:
app, platform, tortoise_mixin = make_platform()
await _seed_two_matches(tortoise_mixin)
transport = ASGITransport(app=app)
with mock.patch.object(
MatchPlayer, "filter", wraps=MatchPlayer.filter
) as spy:
async with AsyncClient(
transport=transport, base_url="http://127.0.0.1"
) as client:
with oidc_user(platform.oidc, "alice"):
response = await client.get("/api/me/matches")
self.assertEqual(200, response.status_code)
# One query for all matches, not one per match row.
self.assertEqual(1, spy.call_count)
results = response.json()["results"]
self.assertEqual(2, len(results))
for result in results:
seats = [player["seat"] for player in result["players"]]
self.assertEqual([0, 1], seats)
@async_test
async def test_my_matches_requires_auth(self) -> None:
app, _, _ = make_platform()
@@ -2,11 +2,18 @@
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 InMemoryGameStore
from tavolo.platform.store import (
DEADLINES_KEY,
ONGOING_GAMES_KEY,
InMemoryGameStore,
RedisGameStore,
)
from helpers import DummyEngine, async_test
@@ -27,6 +34,110 @@ def _session(game_id: str = "g1", code: str = "CODE01") -> GameSession:
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:
@@ -114,6 +225,28 @@ class InMemoryGameStoreTest(unittest.TestCase):
["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())
@@ -137,6 +270,65 @@ class InMemoryGameStoreTest(unittest.TestCase):
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_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()
@@ -1,13 +1,91 @@
"""WebSocket live-play tests via kaya's ASGI websocket transport."""
from __future__ import annotations
import json
import unittest
from typing import Any, Dict, Mapping, Sequence
from httpx import ASGITransport, AsyncClient
from httpx_ws import WebSocketDisconnect, aconnect_ws
from httpx_ws import WebSocketDisconnect as HttpxWebSocketDisconnect, aconnect_ws
from httpx_ws.transport import ASGIWebSocketTransport
from kaya.core import WebSocket, WebSocketDisconnect, WebSocketMessage
from helpers import async_test, make_platform, make_user, oidc_user, ws_users
from tavolo.platform import GameSession, Seat
class AbruptCloseSocket(WebSocket):
"""A socket whose peer vanished without a close handshake.
kaya surfaces this from the transport: ``receive()`` reports a
``close`` message so iteration ends normally, while sends raise
``WebSocketDisconnect``.
"""
def __init__(self, path: str, gone: bool = False) -> None:
self.path = path
self.query_string = ""
self.scheme = "ws"
self.headers: Mapping[str, Sequence[str]] = {}
self.client = ("127.0.0.1", 12345)
self.server = ("127.0.0.1", 80)
self.sent: list[Dict[str, Any]] = []
self.close_codes: list[int] = []
self.gone = gone
async def accept(self, headers: Any = None) -> None:
pass
async def receive(self) -> WebSocketMessage:
self.gone = True
return WebSocketMessage(kind="close")
async def send_text(self, data: str) -> None:
if self.gone:
raise WebSocketDisconnect(code=1006)
self.sent.append(json.loads(data))
async def send_bytes(self, data: bytes) -> None:
if self.gone:
raise WebSocketDisconnect(code=1006)
async def close(self, code: int = 1000) -> None:
self.close_codes.append(code)
async def __anext__(self) -> WebSocketMessage:
await self.receive()
raise StopAsyncIteration
class RecordingSocket(WebSocket):
"""A socket that only records pre-accept rejections."""
def __init__(self, path: str) -> None:
self.path = path
self.query_string = ""
self.scheme = "ws"
self.headers: Mapping[str, Sequence[str]] = {}
self.client = ("127.0.0.1", 12345)
self.server = ("127.0.0.1", 80)
self.close_codes: list[int] = []
async def accept(self, headers: Any = None) -> None:
raise AssertionError("a rejected socket must not be accepted")
async def receive(self) -> WebSocketMessage:
raise AssertionError("a rejected socket must not receive")
async def send_text(self, data: str) -> None:
raise AssertionError("a rejected socket must not send")
async def send_bytes(self, data: bytes) -> None:
raise AssertionError("a rejected socket must not send")
async def close(self, code: int = 1000) -> None:
self.close_codes.append(code)
async def __anext__(self) -> WebSocketMessage:
raise AssertionError("a rejected socket must not receive")
class WebSocketTest(unittest.TestCase):
@@ -87,6 +165,38 @@ class WebSocketTest(unittest.TestCase):
resent = await bob_ws.receive_json()
self.assertEqual("state", resent["type"])
@async_test
async def test_abrupt_disconnect_is_not_an_error(self) -> None:
app, platform, _ = make_platform()
api_transport = ASGITransport(app=app)
async with AsyncClient(transport=api_transport, base_url="http://127.0.0.1") as client:
state = await self._started_game(client, platform.oidc)
game_id = state["id"]
ws = AbruptCloseSocket(f"/ws/games/{game_id}")
with ws_users([make_user("bob")]):
# kaya reports a peer that vanished without a close handshake as
# a close message; the handler must exit cleanly.
await app.handle_websocket(ws)
self.assertEqual("state", ws.sent[0]["type"])
self.assertEqual([], ws.close_codes)
@async_test
async def test_send_on_vanished_peer_is_ignored(self) -> None:
app, platform, _ = make_platform()
api_transport = ASGITransport(app=app)
async with AsyncClient(transport=api_transport, base_url="http://127.0.0.1") as client:
state = await self._started_game(client, platform.oidc)
game_id = state["id"]
ws = AbruptCloseSocket(f"/ws/games/{game_id}", gone=True)
with ws_users([make_user("bob")]):
# Sending raises WebSocketDisconnect; the handler swallows it
# rather than letting it escape the forward task.
await app.handle_websocket(ws)
self.assertEqual([], ws.sent)
self.assertEqual([], ws.close_codes)
@async_test
async def test_game_over_broadcast(self) -> None:
app, platform, _ = make_platform()
@@ -121,10 +231,10 @@ class WebSocketTest(unittest.TestCase):
ws_transport = ASGIWebSocketTransport(app=app)
async with AsyncClient(transport=ws_transport, base_url="http://testserver") as ws_client:
with ws_users([make_user("alice")]):
with self.assertRaises(WebSocketDisconnect) as caught:
with self.assertRaises(HttpxWebSocketDisconnect) as caught:
async with aconnect_ws("/ws/games/no-such-game", ws_client):
pass
self.assertEqual(4404, caught.exception.code)
self.assertEqual(404, caught.exception.code)
@async_test
async def test_non_player_is_closed(self) -> None:
@@ -137,10 +247,10 @@ class WebSocketTest(unittest.TestCase):
ws_transport = ASGIWebSocketTransport(app=app)
async with AsyncClient(transport=ws_transport, base_url="http://testserver") as ws_client:
with ws_users([make_user("mallory")]):
with self.assertRaises(WebSocketDisconnect) as caught:
with self.assertRaises(HttpxWebSocketDisconnect) as caught:
async with aconnect_ws(f"/ws/games/{game_id}", ws_client):
pass
self.assertEqual(4403, caught.exception.code)
self.assertEqual(403, caught.exception.code)
@async_test
async def test_unauthenticated_is_closed(self) -> None:
@@ -148,10 +258,48 @@ class WebSocketTest(unittest.TestCase):
ws_transport = ASGIWebSocketTransport(app=app)
async with AsyncClient(transport=ws_transport, base_url="http://testserver") as ws_client:
with ws_users([]):
with self.assertRaises(WebSocketDisconnect) as caught:
with self.assertRaises(HttpxWebSocketDisconnect) as caught:
async with aconnect_ws("/ws/games/whatever", ws_client):
pass
self.assertEqual(4401, caught.exception.code)
self.assertEqual(401, caught.exception.code)
class RejectionStatusTest(unittest.TestCase):
"""Pre-accept rejections must carry HTTP statuses, not close codes.
Granian's RSGI layer uses the value passed to ``close()`` before
``accept()`` as the HTTP status of the refused upgrade; a WebSocket
close code like 4401 is not a valid status and panics the worker
(``InvalidStatusCode``, and with ``panic = "abort"`` the whole app
dies). Drive the rejections through the app with a recording socket
so the invariant is checked without a real server.
"""
@async_test
async def test_pre_accept_rejections_are_http_statuses(self) -> None:
app, platform, _ = make_platform()
session = GameSession(
id="seated-game",
game_type="dummy",
join_code="SEAT01",
creator_sub="alice",
players=[Seat(user_sub="alice", display_name="Alice")],
)
engine = platform.registry.require("dummy")
engine.create(session, {})
await platform.game_store.save(session)
cases = [
("unauthenticated", [], "/ws/games/whatever", 401),
("unknown game", [make_user("alice")], "/ws/games/no-such-game", 404),
("not seated", [make_user("mallory")], f"/ws/games/{session.id}", 403),
]
for label, users, path, expected in cases:
ws = RecordingSocket(path)
with ws_users(users):
await app.handle_websocket(ws)
self.assertEqual([expected], ws.close_codes, label)
for code in ws.close_codes:
self.assertTrue(100 <= code <= 599, f"{label}: {code} is not an HTTP status")
if __name__ == "__main__":
+83 -11
View File
@@ -1,15 +1,19 @@
#
# This file is autogenerated by pip-compile with Python 3.14
# by the following command:
# by the following command (run from server/packages/tavolo-app/):
#
# pip-compile --allow-unsafe --extra-index-url=https://pypi.org/simple --index-url=https://gitea.woggioni.net/api/packages/woggioni/pypi/simple --output-file=requirements.txt pyproject.toml
# pip-compile --allow-unsafe --extra otel --extra-index-url=https://pypi.org/simple --index-url=https://gitea.woggioni.net/api/packages/woggioni/pypi/simple --output-file=../../requirements.txt pyproject.toml
#
# NOTE: the tavolo-platform, tavolo-scopone and tavolo-app sibling packages
# are NOT pinned here: they are installed from ./packages in the Dockerfile
# and in development (pip install -e ./packages/tavolo-platform
# -e ./packages/tavolo-scopone -e ./packages/tavolo-app). All of their
# third-party dependencies are direct dependencies of tavolo-app, so this
# lock covers the full closure.
# lock covers the full closure. The otel extra (kaya-otel and its
# OpenTelemetry dependencies) IS included, so OTEL_ENABLED works in the
# Docker image; pip-compile cannot resolve the unpublishable sibling
# packages, so regeneration drops them from a copy of the input file (the
# existing pins constrain the rest).
#
--index-url https://gitea.woggioni.net/api/packages/woggioni/pypi/simple
--extra-index-url https://pypi.org/simple
@@ -39,6 +43,8 @@ cryptography==50.0.1
# via pyjwt
dictdiffer==0.10.0
# via aerich
googleapis-common-protos==1.75.4
# via opentelemetry-exporter-otlp-proto-http
granian==2.8.3
# via
# kaya-rsgi
@@ -57,29 +63,85 @@ idna==3.20
# httpx
iso8601==2.1.0
# via tortoise-orm
kaya-core==0.0.4
kaya-core==0.0.7
# via
# kaya-cors
# kaya-oidc
# kaya-openapi
# kaya-otel
# kaya-rsgi
# kaya-session
# tavolo-app (pyproject.toml)
kaya-cors==0.0.4
kaya-cors==0.0.7
# via tavolo-app (pyproject.toml)
kaya-oidc==0.0.4
kaya-oidc==0.0.7
# via tavolo-app (pyproject.toml)
kaya-openapi==0.0.4
kaya-openapi==0.0.7
# via tavolo-app (pyproject.toml)
kaya-rsgi==0.0.4
kaya-otel==0.0.7
# via tavolo-app (pyproject.toml)
kaya-session==0.0.4
kaya-rsgi==0.0.7
# via tavolo-app (pyproject.toml)
kaya-session==0.0.7
# via
# kaya-oidc
# kaya-session-redis
# tavolo-app (pyproject.toml)
kaya-session-redis==0.0.4
kaya-session-redis==0.0.7
# via tavolo-app (pyproject.toml)
opentelemetry-api==1.45.0
# via
# opentelemetry-exporter-http-transport
# opentelemetry-exporter-otlp-proto-http
# opentelemetry-instrumentation
# opentelemetry-instrumentation-asyncpg
# opentelemetry-instrumentation-redis
# opentelemetry-instrumentation-system-metrics
# opentelemetry-sdk
# opentelemetry-semantic-conventions
opentelemetry-exporter-http-transport[urllib3]==0.66b0
# via opentelemetry-exporter-otlp-proto-http
opentelemetry-exporter-otlp-common==0.66b0
# via opentelemetry-exporter-otlp-proto-http
opentelemetry-exporter-otlp-proto-common==1.45.0
# via opentelemetry-exporter-otlp-proto-http
opentelemetry-exporter-otlp-proto-http==1.45.0
# via kaya-otel
opentelemetry-instrumentation==0.66b0
# via
# opentelemetry-instrumentation-asyncpg
# opentelemetry-instrumentation-redis
# opentelemetry-instrumentation-system-metrics
opentelemetry-instrumentation-asyncpg==0.66b0
# via tavolo-app (pyproject.toml)
opentelemetry-instrumentation-redis==0.66b0
# via tavolo-app (pyproject.toml)
opentelemetry-instrumentation-system-metrics==0.66b0
# via tavolo-app (pyproject.toml)
opentelemetry-proto==1.45.0
# via
# opentelemetry-exporter-otlp-proto-common
# opentelemetry-exporter-otlp-proto-http
opentelemetry-sdk==1.45.0
# via
# kaya-otel
# opentelemetry-exporter-otlp-common
# opentelemetry-exporter-otlp-proto-http
opentelemetry-semantic-conventions==0.66b0
# via
# opentelemetry-instrumentation
# opentelemetry-instrumentation-asyncpg
# opentelemetry-instrumentation-redis
# opentelemetry-instrumentation-system-metrics
# opentelemetry-sdk
packaging==26.3
# via opentelemetry-instrumentation
protobuf==7.36.2
# via
# googleapis-common-protos
# opentelemetry-proto
psutil==7.2.2
# via opentelemetry-instrumentation-system-metrics
pwo==0.1.2
# via
# kaya-core
@@ -88,7 +150,7 @@ pwo==0.1.2
# tavolo-app (pyproject.toml)
pycparser==3.0
# via cffi
pyjwt[crypto]==2.14.0
pyjwt[crypto]==2.15.1
# via
# kaya-oidc
# tavolo-app (pyproject.toml)
@@ -106,4 +168,14 @@ typing-extensions==4.16.0
# via
# anyio
# kaya-core
# opentelemetry-api
# opentelemetry-exporter-otlp-proto-http
# opentelemetry-sdk
# opentelemetry-semantic-conventions
# pwo
urllib3==2.8.0
# via opentelemetry-exporter-http-transport
wrapt==2.5.0
# via
# opentelemetry-instrumentation
# opentelemetry-instrumentation-redis
+4 -3
View File
@@ -10,8 +10,9 @@ from __future__ import annotations
import os
os.environ.setdefault("DATABASE_URL", "sqlite://:memory:")
os.environ.setdefault("OIDC_ISSUER", "http://localhost:8180/tavolo")
os.environ.setdefault("OIDC_CLIENT_ID", "tavolo")
os.environ.setdefault("OIDC_REDIRECT_URI", "http://localhost:8080/auth/callback")
os.environ.setdefault("OIDC_0_PROVIDER", "default")
os.environ.setdefault("OIDC_0_ISSUER", "http://localhost:8180/tavolo")
os.environ.setdefault("OIDC_0_CLIENT_ID", "tavolo")
os.environ.setdefault("OIDC_0_REDIRECT_URI", "http://localhost:8080/auth/default/callback")
# Unset REDIS_URL: sessions and live games use the in-memory fallbacks.
os.environ.pop("REDIS_URL", None)
+11 -6
View File
@@ -1,8 +1,9 @@
"""Helpers for faking the OIDC authenticated user during tests.
HTTP handlers funnel through ``oidc_mixin.get_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.
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
@@ -14,9 +15,13 @@ 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_mixin
from tavolo.app import oidc
from tavolo.platform import auth
# 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) -> OIDCUser:
return OIDCUser({"sub": sub, "preferred_username": name or sub})
@@ -24,9 +29,9 @@ def make_user(sub: str, name: Optional[str] = None) -> OIDCUser:
@contextlib.contextmanager
def oidc_user(sub: str, name: Optional[str] = None) -> Iterator[OIDCUser]:
"""Context manager: patch ``oidc_mixin.get_user`` to return this user."""
"""Context manager: make ``oidc.get_users`` return this user."""
user = make_user(sub, name)
patcher = _mock.patch.object(oidc_mixin, "get_user", return_value=user)
patcher = _mock.patch.object(oidc, "get_users", return_value={_PROVIDER: user})
patcher.start()
try:
yield user
+148
View File
@@ -164,6 +164,154 @@ class OTelSettingsTests(unittest.TestCase):
settings = _settings({"OTEL_EXCLUDED_PATHS": "/api/health, /metrics"})
self.assertEqual(("/api/health", "/metrics"), settings.otel_excluded_paths)
def test_otel_interval_defaults(self):
settings = _settings({})
self.assertEqual(60000, settings.otel_metric_export_interval_millis)
self.assertEqual(15, settings.otel_metrics_poll_interval_seconds)
def test_otel_intervals_parse(self):
settings = _settings({
"OTEL_METRIC_EXPORT_INTERVAL_MILLIS": "5000",
"OTEL_METRICS_POLL_INTERVAL_SECONDS": "3",
})
self.assertEqual(5000, settings.otel_metric_export_interval_millis)
self.assertEqual(3, settings.otel_metrics_poll_interval_seconds)
class OidcSettingsTests(unittest.TestCase):
def test_defaults_to_the_dev_mock_provider(self):
settings = _settings({})
self.assertEqual(1, len(settings.oidc_providers))
provider = settings.oidc_providers[0]
self.assertEqual("default", provider.provider)
self.assertEqual("http://localhost:8180/tavolo", provider.issuer)
self.assertEqual("tavolo", provider.client_id)
self.assertIsNone(provider.client_secret)
self.assertEqual(
"http://localhost:8080/auth/default/callback", provider.redirect_uri
)
self.assertEqual("/", provider.post_login_redirect)
self.assertEqual("/", provider.post_logout_redirect)
def test_single_indexed_provider_defaults(self):
settings = _settings({
"OIDC_0_PROVIDER": "google",
"OIDC_0_ISSUER": "https://accounts.google.com",
"OIDC_0_CLIENT_ID": "client-id",
"OIDC_0_CLIENT_SECRET": "client-secret",
})
self.assertEqual(1, len(settings.oidc_providers))
provider = settings.oidc_providers[0]
self.assertEqual("google", provider.provider)
self.assertEqual("https://accounts.google.com", provider.issuer)
self.assertEqual("client-id", provider.client_id)
self.assertEqual("client-secret", provider.client_secret)
# The callback default mirrors kaya's /auth/<provider>/callback.
self.assertEqual(
"http://localhost:8080/auth/google/callback", provider.redirect_uri
)
self.assertEqual("/", provider.post_login_redirect)
self.assertEqual("/", provider.post_logout_redirect)
def test_multiple_providers_are_ordered_by_index(self):
settings = _settings({
"OIDC_0_PROVIDER": "google",
"OIDC_0_ISSUER": "https://accounts.google.com",
"OIDC_0_CLIENT_ID": "g",
"OIDC_1_PROVIDER": "github",
"OIDC_1_ISSUER": "https://github.com/login/oauth",
"OIDC_1_CLIENT_ID": "h",
"OIDC_1_REDIRECT_URI": "https://tavolo.example/auth/github/callback",
"OIDC_1_POST_LOGIN_REDIRECT": "/lobby",
"OIDC_1_POST_LOGOUT_REDIRECT": "/bye",
})
self.assertEqual(
["google", "github"], [p.provider for p in settings.oidc_providers]
)
github = settings.oidc_providers[1]
self.assertEqual(
"https://tavolo.example/auth/github/callback", github.redirect_uri
)
self.assertEqual("/lobby", github.post_login_redirect)
self.assertEqual("/bye", github.post_logout_redirect)
def test_empty_optional_values_fall_back_to_defaults(self):
settings = _settings({
"OIDC_0_PROVIDER": "google",
"OIDC_0_ISSUER": "https://accounts.google.com",
"OIDC_0_CLIENT_ID": "g",
"OIDC_0_CLIENT_SECRET": "",
"OIDC_0_REDIRECT_URI": "",
})
provider = settings.oidc_providers[0]
self.assertIsNone(provider.client_secret)
self.assertEqual(
"http://localhost:8080/auth/google/callback", provider.redirect_uri
)
def test_missing_provider_name(self):
with self.assertRaisesRegex(RuntimeError, "OIDC_0_PROVIDER"):
_settings({"OIDC_0_ISSUER": "https://idp", "OIDC_0_CLIENT_ID": "c"})
def test_missing_issuer(self):
with self.assertRaisesRegex(RuntimeError, "OIDC_0_ISSUER"):
_settings({"OIDC_0_PROVIDER": "google", "OIDC_0_CLIENT_ID": "c"})
def test_missing_client_id(self):
with self.assertRaisesRegex(RuntimeError, "OIDC_0_CLIENT_ID"):
_settings({"OIDC_0_PROVIDER": "google", "OIDC_0_ISSUER": "https://idp"})
def test_indices_must_start_at_zero(self):
with self.assertRaisesRegex(RuntimeError, "contiguous"):
_settings({
"OIDC_3_PROVIDER": "google",
"OIDC_3_ISSUER": "https://idp",
"OIDC_3_CLIENT_ID": "c",
})
def test_indices_must_be_contiguous(self):
with self.assertRaisesRegex(RuntimeError, "OIDC_1_"):
_settings({
"OIDC_0_PROVIDER": "google",
"OIDC_0_ISSUER": "https://idp",
"OIDC_0_CLIENT_ID": "c",
"OIDC_2_PROVIDER": "github",
"OIDC_2_ISSUER": "https://idp2",
"OIDC_2_CLIENT_ID": "c2",
})
def test_duplicate_provider_names_are_rejected(self):
with self.assertRaisesRegex(RuntimeError, "duplicate"):
_settings({
"OIDC_0_PROVIDER": "google",
"OIDC_0_ISSUER": "https://idp",
"OIDC_0_CLIENT_ID": "c",
"OIDC_1_PROVIDER": "google",
"OIDC_1_ISSUER": "https://idp2",
"OIDC_1_CLIENT_ID": "c2",
})
def test_invalid_provider_name_is_rejected(self):
with self.assertRaisesRegex(RuntimeError, "OIDC_0_PROVIDER"):
_settings({
"OIDC_0_PROVIDER": "not a name",
"OIDC_0_ISSUER": "https://idp",
"OIDC_0_CLIENT_ID": "c",
})
def test_unknown_indexed_variable_is_rejected(self):
with self.assertRaisesRegex(RuntimeError, "OIDC_0_ISUER"):
_settings({
"OIDC_0_PROVIDER": "google",
"OIDC_0_ISSUER": "https://idp",
"OIDC_0_CLIENT_ID": "c",
"OIDC_0_ISUER": "typo",
})
def test_legacy_unindexed_variables_get_a_migration_hint(self):
with self.assertRaisesRegex(RuntimeError, "OIDC_ISSUER -> OIDC_0_ISSUER"):
_settings({"OIDC_ISSUER": "https://idp", "OIDC_CLIENT_ID": "c"})
if __name__ == "__main__":
unittest.main()
+9
View File
@@ -18,12 +18,15 @@ class LoggingConfigTest(unittest.TestCase):
self._root_level = root.level
tavolo = logging.getLogger("tavolo")
self._tavolo_level = tavolo.level
pool = logging.getLogger("urllib3.connectionpool")
self._pool_level = pool.level
def tearDown(self) -> None:
root = logging.getLogger()
root.handlers = self._root_handlers
root.level = self._root_level
logging.getLogger("tavolo").level = self._tavolo_level
logging.getLogger("urllib3.connectionpool").level = self._pool_level
def test_default_config_when_unset(self) -> None:
configure_logging(None)
@@ -33,6 +36,12 @@ class LoggingConfigTest(unittest.TestCase):
any(isinstance(h, logging.StreamHandler) for h in root.handlers),
"default config installs a console stream handler",
)
# The noisy per-request HTTP logger is capped at INFO while the
# root stays at DEBUG.
self.assertEqual(
logging.INFO,
logging.getLogger("urllib3.connectionpool").getEffectiveLevel(),
)
def test_yaml_config_is_applied(self) -> None:
with tempfile.TemporaryDirectory() as tmp:
+28
View File
@@ -16,6 +16,7 @@ from unittest.mock import patch
from tavolo.app import otel_mixin_from_settings
from tavolo.config import Settings
from tavolo.telemetry import session_tag_hook
def _settings(env: dict) -> Settings:
@@ -58,8 +59,17 @@ class OTelMixinFromSettingsTests(unittest.TestCase):
"endpoint": "http://collector:4318",
"headers": {"Authorization": "Bearer t", "X-Tenant": "one"},
"excluded_paths": ("/api/health",),
"server_response_hook": session_tag_hook,
"websocket_close_hook": session_tag_hook,
}, mixin.kwargs)
def test_session_tag_hooks_are_wired(self):
with _stub_kaya_otel():
mixin = otel_mixin_from_settings(_settings({"OTEL_ENABLED": "1"}))
assert isinstance(mixin, _StubOTelMixin)
self.assertIs(session_tag_hook, mixin.kwargs["server_response_hook"])
self.assertIs(session_tag_hook, mixin.kwargs["websocket_close_hook"])
def test_defaults_when_only_enabled(self):
with _stub_kaya_otel():
mixin = otel_mixin_from_settings(_settings({"OTEL_ENABLED": "on"}))
@@ -69,6 +79,24 @@ class OTelMixinFromSettingsTests(unittest.TestCase):
self.assertIsNone(mixin.kwargs["headers"])
self.assertEqual(("/api/health",), mixin.kwargs["excluded_paths"])
def test_meter_provider_is_passed_through(self):
with _stub_kaya_otel():
provider = object()
mixin = otel_mixin_from_settings(
_settings({"OTEL_ENABLED": "1"}), meter_provider=provider
)
assert isinstance(mixin, _StubOTelMixin)
self.assertIs(provider, mixin.kwargs["meter_provider"])
def test_tracer_provider_is_passed_through(self):
with _stub_kaya_otel():
provider = object()
mixin = otel_mixin_from_settings(
_settings({"OTEL_ENABLED": "1"}), tracer_provider=provider
)
assert isinstance(mixin, _StubOTelMixin)
self.assertIs(provider, mixin.kwargs["tracer_provider"])
def test_excluded_paths_are_passed_through(self):
with _stub_kaya_otel():
mixin = otel_mixin_from_settings(_settings({
+35
View File
@@ -0,0 +1,35 @@
"""Auth routes on the fully wired application.
The integration suite's app is built from the indexed test environment
(one provider named ``default``); these tests pin the public surface the
SPA and bookmarks rely on.
"""
from __future__ import annotations
import unittest
from httpx import ASGITransport, AsyncClient
from tavolo.app import app
from tests.helpers import async_test
class AuthRoutesTest(unittest.TestCase):
@async_test
async def test_providers_endpoint_reports_the_configured_provider(self) -> None:
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport, base_url="http://127.0.0.1") as client:
response = await client.get("/api/auth/providers")
self.assertEqual(200, response.status_code)
self.assertEqual(
{"providers": [{"name": "default", "login_url": "/auth/default/login"}]},
response.json(),
)
@async_test
async def test_generic_login_redirects_to_the_provider(self) -> None:
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport, base_url="http://127.0.0.1") as client:
response = await client.get("/auth/login", follow_redirects=False)
self.assertEqual(302, response.status_code)
self.assertEqual("/auth/default/login", response.headers["location"])
+427
View File
@@ -0,0 +1,427 @@
"""Telemetry unit tests: app gauges, process metrics, log export and counts.
The OpenTelemetry packages are an optional extra (``tavolo-app[otel]``),
so the provider-dependent tests skip when they are not installed.
"""
from __future__ import annotations
import hashlib
import logging
import os
import sys
import unittest
from types import SimpleNamespace
from typing import Optional
from unittest.mock import patch
from kaya.session import Session
from tavolo.telemetry import (
CountingInMemorySessionStore,
CountingRedisSessionStore,
TelemetryMixin,
_NoOtelLogs,
session_tag_hook,
telemetry_from_settings,
)
from tavolo.config import Settings
from tests.helpers import async_test
try:
from opentelemetry.sdk._logs import LoggingHandler, LoggerProvider
from opentelemetry.sdk._logs.export import (
InMemoryLogRecordExporter,
SimpleLogRecordProcessor,
)
from opentelemetry.sdk.metrics import MeterProvider
from opentelemetry.sdk.metrics.export import InMemoryMetricReader
from opentelemetry.sdk.resources import Resource
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import SimpleSpanProcessor
from opentelemetry.sdk.trace.export.in_memory_span_exporter import (
InMemorySpanExporter,
)
HAVE_OTEL = True
except ImportError: # pragma: no cover - installs without the otel extra
HAVE_OTEL = False
def _settings(env: dict) -> Settings:
with patch.dict(os.environ, env, clear=True):
return Settings.from_env()
class _FakeSessions:
def __init__(self, count: int = 0) -> None:
self.value: Optional[int] = count
self.error: Optional[Exception] = None
async def count(self) -> int:
if self.error is not None:
raise self.error
assert self.value is not None
return self.value
class _FakeGames:
def __init__(self, count: int = 0) -> None:
self.value: Optional[int] = count
async def count_ongoing(self) -> int:
assert self.value is not None
return self.value
class _FakeRedis:
"""Minimal async Redis double covering the session-count scan."""
def __init__(self, keys: list) -> None:
self.keys = keys
async def scan_iter(self, match: str = "", count: int = 0):
prefix = match.rstrip("*")
for key in self.keys:
if key.startswith(prefix):
yield key
class TelemetrySetupTest(unittest.TestCase):
def test_disabled_returns_none(self) -> None:
self.assertIsNone(
telemetry_from_settings(
_settings({}), _FakeSessions(), _FakeGames()
)
)
def test_missing_otel_dependencies_raise(self) -> None:
settings = _settings({"OTEL_ENABLED": "true"})
missing = "opentelemetry.exporter.otlp.proto.http._log_exporter"
with patch.dict(sys.modules, {missing: None}):
with self.assertRaises(RuntimeError):
telemetry_from_settings(settings, _FakeSessions(), _FakeGames())
class SessionCountTest(unittest.TestCase):
@async_test
async def test_in_memory_counts_live_sessions(self) -> None:
now = [1000.0]
store = CountingInMemorySessionStore(clock=lambda: now[0])
await store.save("a", Session("a", {"user": "alice"}), max_age=10)
await store.save("b", Session("b", {"user": "bob"}), max_age=10)
await store.save("c", Session("c", {"user": "carol"})) # no expiry
self.assertEqual(3, await store.count())
now[0] += 11 # a and b have expired
self.assertEqual(1, await store.count())
await store.delete("c")
self.assertEqual(0, await store.count())
@async_test
async def test_redis_counts_only_session_keys(self) -> None:
store = CountingRedisSessionStore(
_FakeRedis(["kaya:session:a", "kaya:session:b", "tavolo:code:X"])
)
self.assertEqual(2, await store.count())
class NoOtelLogsFilterTest(unittest.TestCase):
def test_filters_opentelemetry_records_only(self) -> None:
filter_ = _NoOtelLogs()
self.assertFalse(
filter_.filter(logging.makeLogRecord({"name": "opentelemetry.sdk"}))
)
self.assertTrue(
filter_.filter(logging.makeLogRecord({"name": "tavolo.app"}))
)
self.assertTrue(
filter_.filter(logging.makeLogRecord({"name": "granian.server"}))
)
class _RecordedSpan:
def __init__(self) -> None:
self.attributes: dict = {}
def set_attribute(self, key: str, value) -> None:
self.attributes[key] = value
class SessionTagHookTest(unittest.TestCase):
def test_tags_authenticated_session(self) -> None:
span = _RecordedSpan()
carrier = SimpleNamespace(
session=Session("sid", {"oidc:default:user": {"sub": "alice"}})
)
session_tag_hook(span, carrier)
self.assertEqual(
hashlib.sha256(b"sid").hexdigest()[:16],
span.attributes["session.id"],
)
self.assertEqual("alice", span.attributes["user.id"])
def test_tags_user_of_any_provider(self) -> None:
span = _RecordedSpan()
carrier = SimpleNamespace(
session=Session("sid", {"oidc:google:user": {"sub": "alice"}})
)
session_tag_hook(span, carrier)
self.assertEqual("alice", span.attributes["user.id"])
def test_ignores_legacy_unprefixed_user_key(self) -> None:
span = _RecordedSpan()
carrier = SimpleNamespace(
session=Session("sid", {"oidc_user": {"sub": "alice"}})
)
session_tag_hook(span, carrier)
self.assertIn("session.id", span.attributes)
self.assertNotIn("user.id", span.attributes)
def test_tags_session_without_user(self) -> None:
span = _RecordedSpan()
session_tag_hook(span, SimpleNamespace(session=Session("sid", {})))
self.assertIn("session.id", span.attributes)
self.assertNotIn("user.id", span.attributes)
def test_tags_user_when_session_id_not_yet_minted(self) -> None:
span = _RecordedSpan()
carrier = SimpleNamespace(
session=Session(None, {"oidc:default:user": {"sub": "alice"}})
)
session_tag_hook(span, carrier)
self.assertNotIn("session.id", span.attributes)
self.assertEqual("alice", span.attributes["user.id"])
def test_missing_session_is_a_noop(self) -> None:
span = _RecordedSpan()
session_tag_hook(span, SimpleNamespace())
self.assertEqual({}, span.attributes)
def test_malformed_user_is_ignored(self) -> None:
span = _RecordedSpan()
carrier = SimpleNamespace(session=Session("sid", {"oidc:default:user": "alice"}))
session_tag_hook(span, carrier)
self.assertIn("session.id", span.attributes)
self.assertNotIn("user.id", span.attributes)
@unittest.skipUnless(HAVE_OTEL, "tavolo-app[otel] not installed")
class SessionTagIntegrationTest(unittest.TestCase):
"""The hook is wired as kaya-otel's response hook, so it sees the
session after the session mixin persisted and minted it."""
@async_test
async def test_spans_carry_hashed_session_and_oidc_sub(self) -> None:
from httpx import ASGITransport, AsyncClient
from kaya.core import HttpContext, KayaApp
from kaya.otel import OTelMixin
from kaya.session import InMemorySessionStore, SessionMixin
span_exporter = InMemorySpanExporter()
tracer_provider = TracerProvider(
resource=Resource.create({"service.name": "test"})
)
tracer_provider.add_span_processor(SimpleSpanProcessor(span_exporter))
app = KayaApp(
mixins=[
OTelMixin(
service_name="test",
tracer_provider=tracer_provider,
meter_provider=MeterProvider(
metric_readers=[InMemoryMetricReader()]
),
server_response_hook=session_tag_hook,
websocket_close_hook=session_tag_hook,
),
SessionMixin(InMemorySessionStore()),
]
)
@app.GET("/login")
async def login(ctx: HttpContext) -> None:
session = ctx.session
assert session is not None
session["oidc:default:user"] = {"sub": "alice"}
await ctx.send_str(200, "ok")
@app.GET("/me")
async def me(ctx: HttpContext) -> None:
await ctx.send_str(200, "ok")
transport = ASGITransport(app=app) # type: ignore[arg-type]
async with AsyncClient(
transport=transport, base_url="http://127.0.0.1"
) as client:
first = await client.get("/login")
second = await client.get("/me")
self.assertEqual(200, first.status_code)
self.assertEqual(200, second.status_code)
# The cookie value is the session id; only its digest is exported.
cookie = first.cookies["session_id"]
digest = hashlib.sha256(cookie.encode()).hexdigest()[:16]
spans = {span.name: span for span in span_exporter.get_finished_spans()}
self.assertEqual({"GET /login", "GET /me"}, set(spans))
for name, span in spans.items():
attributes = span.attributes
assert attributes is not None
self.assertEqual(digest, attributes["session.id"], name)
self.assertEqual("alice", attributes["user.id"], name)
def _metric_points(reader) -> dict:
"""Map metric name -> list of (attributes, value) data points."""
data = reader.get_metrics_data()
assert data is not None
points = {}
for resource_metrics in data.resource_metrics:
for scope_metrics in resource_metrics.scope_metrics:
for metric in scope_metrics.metrics:
points[metric.name] = [
(dict(point.attributes), point.value)
for point in metric.data.data_points
]
return points
@unittest.skipUnless(HAVE_OTEL, "tavolo-app[otel] not installed")
class MetricExportTest(unittest.TestCase):
# The instrumentation packages are process-wide singletons whose
# callbacks cannot be re-registered, so the providers and mixin are
# built once for the whole class (as the application does).
sessions = _FakeSessions(0)
games = _FakeGames(0)
@classmethod
def setUpClass(cls) -> None:
resource = Resource.create({"service.name": "test"})
cls.reader = InMemoryMetricReader()
cls.provider = MeterProvider(resource=resource, metric_readers=[cls.reader])
cls.span_exporter = InMemorySpanExporter()
cls.tracer_provider = TracerProvider(resource=resource)
cls.tracer_provider.add_span_processor(
SimpleSpanProcessor(cls.span_exporter)
)
cls.mixin = TelemetryMixin(
meter_provider=cls.provider,
tracer_provider=cls.tracer_provider,
logger_provider=None,
log_handler=None,
session_store=cls.sessions,
game_store=cls.games,
poll_interval_seconds=1,
)
cls.mixin.apply(None) # type: ignore[arg-type]
@classmethod
def tearDownClass(cls) -> None:
cls.mixin.shutdown(None) # type: ignore[arg-type]
@async_test
async def test_app_gauges_report_polled_counts(self) -> None:
self.sessions.value = 3
self.sessions.error = None
self.games.value = 2
await self.mixin.refresh_counts()
points = _metric_points(self.reader)
self.assertEqual([({}, 3)], points["tavolo.active_users"])
self.assertEqual([({}, 2)], points["tavolo.ongoing_games"])
@async_test
async def test_failed_refresh_keeps_last_value(self) -> None:
self.sessions.value = 5
self.sessions.error = None
await self.mixin.refresh_counts()
self.sessions.error = RuntimeError("redis down")
await self.mixin.refresh_counts()
points = _metric_points(self.reader)
self.assertEqual([({}, 5)], points["tavolo.active_users"])
@async_test
async def test_process_and_gc_metrics_are_exported(self) -> None:
await self.mixin.refresh_counts()
points = _metric_points(self.reader)
for name in (
"process.cpu.time",
"process.cpu.utilization",
"process.memory.usage",
"process.memory.virtual",
"cpython.gc.collections",
"cpython.gc.collected_objects",
"cpython.gc.uncollectable_objects",
):
self.assertIn(name, points)
self.assertGreater(points["process.memory.usage"][0][1], 0)
modes = {attrs["type"] for attrs, _ in points["process.cpu.time"]}
self.assertEqual({"user", "system"}, modes)
generations = {
attrs["cpython.gc.generation"]
for attrs, _ in points["cpython.gc.collections"]
}
self.assertEqual({0, 1, 2}, generations)
# Host-level metrics are deliberately not configured.
self.assertNotIn("system.cpu.time", points)
@async_test
async def test_database_clients_are_instrumented(self) -> None:
import asyncpg.connection
import redis.asyncio
self.assertTrue(hasattr(redis.asyncio.Redis.execute_command, "__wrapped__"))
self.assertTrue(hasattr(asyncpg.connection.Connection.fetch, "__wrapped__"))
# Exercise each wrapper through stubs: the downstream call fails (no
# server), but the span is recorded before it does.
client = object.__new__(redis.asyncio.Redis)
client.connection_pool = SimpleNamespace(
connection_kwargs={"host": "127.0.0.1", "port": 6379, "db": 0}
)
with self.assertRaises(AttributeError):
await client.execute_command("GET", "key")
connection = object.__new__(asyncpg.connection.Connection)
connection._aborted = True # keep __del__ quiet on the stub
with self.assertRaises(Exception):
await connection.fetch("SELECT 1")
spans = {
span.name: span for span in self.span_exporter.get_finished_spans()
}
self.assertEqual("GET ?", spans["GET"].attributes["db.statement"])
self.assertEqual("redis", spans["GET"].attributes["db.system"])
self.assertEqual("SELECT 1", spans["SELECT"].attributes["db.statement"])
self.assertEqual("postgresql", spans["SELECT"].attributes["db.system"])
@unittest.skipUnless(HAVE_OTEL, "tavolo-app[otel] not installed")
class LogExportTest(unittest.TestCase):
def test_logs_are_exported_and_otel_records_filtered(self) -> None:
exporter = InMemoryLogRecordExporter()
resource = Resource.create({"service.name": "test"})
provider = LoggerProvider(resource=resource)
provider.add_log_record_processor(SimpleLogRecordProcessor(exporter))
handler = LoggingHandler(level=logging.NOTSET, logger_provider=provider)
handler.addFilter(_NoOtelLogs())
root = logging.getLogger()
previous_level = root.level
root.addHandler(handler)
root.setLevel(logging.DEBUG)
try:
logging.getLogger("tavolo.test").info("exported message")
logging.getLogger("opentelemetry.test").info("filtered message")
bodies = [r.log_record.body for r in exporter.get_finished_logs()]
self.assertIn("exported message", bodies)
self.assertNotIn("filtered message", bodies)
records = exporter.get_finished_logs()
self.assertEqual(
"test", records[0].resource.attributes["service.name"]
)
finally:
root.removeHandler(handler)
handler.close()
root.setLevel(previous_level)
provider.shutdown()
if __name__ == "__main__":
unittest.main()
+3 -3
View File
@@ -131,7 +131,7 @@ class WebSocketTest(unittest.TestCase):
with self.assertRaises(WebSocketDisconnect) as caught:
async with aconnect_ws("/ws/games/no-such-game", ws_client):
pass
self.assertEqual(4404, caught.exception.code)
self.assertEqual(404, caught.exception.code)
@async_test
async def test_non_player_is_closed(self) -> None:
@@ -146,7 +146,7 @@ class WebSocketTest(unittest.TestCase):
with self.assertRaises(WebSocketDisconnect) as caught:
async with aconnect_ws(f"/ws/games/{game_id}", ws_client):
pass
self.assertEqual(4403, caught.exception.code)
self.assertEqual(403, caught.exception.code)
@async_test
async def test_unauthenticated_is_closed(self) -> None:
@@ -156,7 +156,7 @@ class WebSocketTest(unittest.TestCase):
with self.assertRaises(WebSocketDisconnect) as caught:
async with aconnect_ws("/ws/games/whatever", ws_client):
pass
self.assertEqual(4401, caught.exception.code)
self.assertEqual(401, caught.exception.code)
async def _seed_last_play_state(hand_ack_timeout: int = 30) -> str:
+39 -9
View File
@@ -22,6 +22,19 @@ pub async fn me() -> Result<Option<User>, String> {
resp.json().await.map(Some).map_err(|e| e.to_string())
}
/// Fetch the configured OIDC login providers (public endpoint).
pub async fn auth_providers() -> Result<Vec<AuthProvider>, String> {
let resp = Request::get("/api/auth/providers")
.send()
.await
.map_err(|e| e.to_string())?;
if !resp.ok() {
return Err(server_error(resp.status()));
}
let page: AuthProvidersPage = resp.json().await.map_err(|e| e.to_string())?;
Ok(page.providers)
}
/// Fetch the card games the platform can host (for the creation dropdown).
pub async fn game_types() -> Result<Vec<GameTypeInfo>, String> {
let resp = Request::get("/api/game-types")
@@ -87,16 +100,33 @@ pub async fn join_preview(code: &str) -> Result<Option<JoinPreview>, String> {
resp.json().await.map(Some).map_err(|e| e.to_string())
}
#[allow(dead_code)]
pub async fn game_state(game_id: &str) -> Result<GameView, String> {
let resp = Request::get(&format!("/api/games/{game_id}"))
.send()
.await
.map_err(|e| e.to_string())?;
if !resp.ok() {
return Err(server_error(resp.status()));
/// Why a game connection failed before the first snapshot arrived.
///
/// A refused WebSocket handshake is invisible to the browser: the close
/// event reports an abnormal closure (1006) with no HTTP status, so the
/// game page asks the REST API to tell a deliberate rejection from a
/// transient network problem.
pub enum GameEntry {
/// Not logged in (`401` from `GET /api/games/{id}`).
Unauthenticated,
/// Unknown game (`404`) or the user is not seated (`403`).
Unavailable,
/// The API could not be asked (or answered unexpectedly): treat the
/// connection loss as transient and retry.
Transient,
}
/// Classify a failed game connection via `GET /api/games/{game_id}`,
/// mirroring the statuses used to refuse the websocket upgrade.
pub async fn probe_game(game_id: &str) -> GameEntry {
match Request::get(&format!("/api/games/{game_id}")).send().await {
Ok(resp) => match resp.status() {
401 => GameEntry::Unauthenticated,
403 | 404 => GameEntry::Unavailable,
_ => GameEntry::Transient,
},
Err(_) => GameEntry::Transient,
}
resp.json().await.map_err(|e| e.to_string())
}
pub async fn my_matches(cursor: Option<&str>) -> Result<MatchesPage, String> {
+73
View File
@@ -291,6 +291,13 @@ pub fn fmt_points(points: f64) -> String {
}
}
/// Render a calendar instant as `YYYY-MM-DD HH:MM`. `month` is 0-based
/// (the JavaScript `Date` convention); the other components are as
/// returned by `Date`'s local-time getters.
pub fn fmt_datetime_parts(year: u32, month: u32, day: u32, hour: u32, minute: u32) -> String {
format!("{year:04}-{:02}-{day:02} {hour:02}:{minute:02}", month + 1)
}
/// The caller's Elo rating for one game type (GET /api/me/ratings).
#[derive(Debug, Clone, Deserialize)]
#[allow(dead_code)]
@@ -337,6 +344,19 @@ pub struct GameTypesPage {
pub results: Vec<GameTypeInfo>,
}
/// One configured OIDC login provider (GET /api/auth/providers).
#[derive(Debug, Clone, PartialEq, Deserialize)]
pub struct AuthProvider {
pub name: String,
pub login_url: String,
}
#[derive(Debug, Clone, Deserialize)]
pub struct AuthProvidersPage {
#[serde(default)]
pub providers: Vec<AuthProvider>,
}
/// A join-code lookup result (GET /api/games/preview): the game's lobby
/// payload plus its display name, for the pre-join summary. The options
/// echoed by the engine land in `details` under their `options_schema`
@@ -544,6 +564,21 @@ pub fn card_asset(code: &str) -> String {
pub const CARD_BACK: &str = "/assets/cards/back.svg";
/// Display order key for a hand: rank ascending (01..10, i.e. 1..7 then
/// Fante, Cavallo, Re), then suit (denari, coppe, spade, bastoni).
/// Malformed codes sort last.
pub fn hand_sort_key(code: &str) -> (u8, u8) {
let rank = code.get(..2).and_then(|r| r.parse().ok()).unwrap_or(u8::MAX);
let suit = match code.get(2..3) {
Some("D") => 0,
Some("C") => 1,
Some("S") => 2,
Some("B") => 3,
_ => u8::MAX,
};
(rank, suit)
}
/// Human-friendly rank+suit label, e.g. `07D` -> "7 of denari".
pub fn card_label(code: &str) -> String {
if code.len() != 3 {
@@ -593,6 +628,38 @@ mod tests {
assert_eq!("7 di denari", card_label("07D"));
}
#[test]
fn hand_sort_orders_ranks_ascending() {
let mut hand = vec![
"10B".to_string(),
"01S".to_string(),
"07D".to_string(),
"08C".to_string(),
];
hand.sort_by_key(|c| hand_sort_key(c));
assert_eq!(vec!["01S", "07D", "08C", "10B"], hand);
}
#[test]
fn hand_sort_breaks_ties_by_suit() {
let mut hand = vec![
"02B".to_string(),
"02S".to_string(),
"02C".to_string(),
"02D".to_string(),
"01S".to_string(),
];
hand.sort_by_key(|c| hand_sort_key(c));
assert_eq!(vec!["01S", "02D", "02C", "02S", "02B"], hand);
}
#[test]
fn hand_sort_puts_malformed_codes_last() {
let mut hand = vec!["07D".to_string(), "xx".to_string(), "AB1".to_string()];
hand.sort_by_key(|c| hand_sort_key(c));
assert_eq!(vec!["07D", "xx", "AB1"], hand);
}
fn scopone_match() -> MatchSummary {
serde_json::from_value(json!({
"id": "m1",
@@ -658,6 +725,12 @@ mod tests {
assert_eq!("2.5", fmt_points(2.5));
}
#[test]
fn fmt_datetime_parts_pads_and_offsets_month() {
assert_eq!("2026-09-19 05:02", fmt_datetime_parts(2026, 8, 19, 5, 2));
assert_eq!("2026-01-01 00:00", fmt_datetime_parts(2026, 0, 1, 0, 0));
}
#[test]
fn leaderboard_points_accept_floats() {
let e: LeaderboardEntry = serde_json::from_value(json!({
+33 -15
View File
@@ -4,11 +4,13 @@ use std::cell::RefCell;
use std::rc::Rc;
use sycamore::prelude::*;
use wasm_bindgen_futures::spawn_local;
use crate::api;
use crate::components::card::{card_back, card_img};
use crate::components::summary::{hand_summary_modal, summary_rows};
use crate::components::toast::toast;
use crate::model::{card_label, GameView, MoveView, PlayerView, Scores, ServerMessage};
use crate::model::{card_label, hand_sort_key, GameView, MoveView, PlayerView, Scores, ServerMessage};
use crate::ws::{self, GameSocket};
fn send_play(socket: Signal<Option<GameSocket>>, card: String, capture: Option<Vec<String>>) {
@@ -135,23 +137,38 @@ fn start_connect(id: Rc<String>, ctx: ConnCtx, alive: Rc<Cell<bool>>) {
let on_close = {
let id = id.clone();
let alive = alive.clone();
move |code: Option<u16>| {
move |_code: Option<u16>| {
if !alive.get() {
return;
}
ctx.closed.set(true);
match code {
Some(4401) => {
ctx.fatal.set(true);
ctx.error
.set(Some("Session expired — please log in again.".to_string()));
}
Some(4403) | Some(4404) => {
ctx.fatal.set(true);
ctx.error
.set(Some("This game is no longer available.".to_string()));
}
_ => schedule_retry(id.clone(), ctx, alive.clone()),
if ctx.game.get_clone().is_none() && ctx.attempts.get() == 0 {
// The connection ended before any snapshot arrived. A
// refused handshake looks exactly like a network drop to
// the browser (close code 1006, no HTTP status), so ask
// the API whether the server deliberately rejected us —
// in which case retrying is pointless — or the loss was
// transient. Only on the first failure: later attempts
// are governed by the retry backoff.
let id = id.clone();
let alive = alive.clone();
spawn_local(async move {
match api::probe_game(&id).await {
api::GameEntry::Unauthenticated => {
ctx.fatal.set(true);
ctx.error
.set(Some("Session expired — please log in again.".to_string()));
}
api::GameEntry::Unavailable => {
ctx.fatal.set(true);
ctx.error
.set(Some("This game is no longer available.".to_string()));
}
api::GameEntry::Transient => schedule_retry(id, ctx, alive),
}
});
} else {
schedule_retry(id.clone(), ctx, alive.clone());
}
}
};
@@ -576,7 +593,8 @@ fn table_view(
let viewer = player_for_seat(&game, viewer_seat);
let my_captured = viewer.as_ref().map(|p| p.captured_count).unwrap_or(0);
let my_scope = viewer.as_ref().map(|p| p.scope).unwrap_or(0);
let hand = viewer.and_then(|p| p.hand).unwrap_or_default();
let mut hand = viewer.and_then(|p| p.hand).unwrap_or_default();
hand.sort_by_key(|c| hand_sort_key(c));
let current_selection = selected.get_clone();
let hand_cards = hand
.into_iter()
+18 -6
View File
@@ -68,6 +68,23 @@ fn game_name(game_types: &[GameTypeInfo], game_type: &str) -> String {
.unwrap_or_else(|| game_type.to_string())
}
/// The match's finish instant in the browser's local timezone, as
/// `YYYY-MM-DD HH:MM`. Falls back to the raw UTC prefix when the browser
/// cannot parse the timestamp.
fn local_finished(iso: &str) -> String {
let date = js_sys::Date::new(&wasm_bindgen::JsValue::from_str(iso));
if date.get_time().is_nan() {
return iso.replace('T', " ").chars().take(16).collect();
}
crate::model::fmt_datetime_parts(
date.get_full_year(),
date.get_month(),
date.get_date(),
date.get_hours(),
date.get_minutes(),
)
}
#[component]
pub fn HistoryPage() -> View {
let page = create_signal(Option::<MatchesPage>::None);
@@ -131,12 +148,7 @@ pub fn HistoryPage() -> View {
.get_clone()
.into_iter()
.map(|m| {
let finished = m
.finished_at
.replace('T', " ")
.chars()
.take(16)
.collect::<String>();
let finished = local_finished(&m.finished_at);
let game = game_name(&names, &m.game_type);
let mut cells: Vec<View> = team_columns(&m)
.iter()
+53 -10
View File
@@ -14,7 +14,9 @@ use sycamore_router::navigate;
use crate::api;
use crate::components::card::card_back;
use crate::components::toast::toast;
use crate::model::{GameTypeInfo, JoinPreview, OptionField, OptionKind, PlayerRating, User};
use crate::model::{
AuthProvider, GameTypeInfo, JoinPreview, OptionField, OptionKind, PlayerRating, User,
};
/// Used when the game-types fetch fails: match creation must still work.
fn fallback_game_types() -> Vec<GameTypeInfo> {
@@ -28,6 +30,15 @@ fn fallback_game_types() -> Vec<GameTypeInfo> {
}]
}
/// Human-readable label for a provider name ("google" -> "Google").
fn provider_label(name: &str) -> String {
let mut chars = name.chars();
match chars.next() {
Some(first) => first.to_uppercase().collect::<String>() + chars.as_str(),
None => String::new(),
}
}
fn json_to_string(value: &serde_json::Value) -> String {
match value {
serde_json::Value::String(s) => s.clone(),
@@ -283,6 +294,8 @@ fn join_preview_panel(
pub fn LobbyPage() -> View {
// Outer None = still loading; Some(None) = logged out.
let user = create_signal(Option::<Option<User>>::None);
// Configured login providers, for the logged-out card's buttons.
let auth_providers = create_signal(Vec::<AuthProvider>::new());
// The player's Elo ratings, one row per game type played.
let ratings = create_signal(Vec::<PlayerRating>::new());
let error = create_signal(Option::<String>::None);
@@ -400,6 +413,12 @@ pub fn LobbyPage() -> View {
}
});
spawn_local(async move {
if let Ok(list) = api::auth_providers().await {
auth_providers.set(list);
}
});
let on_create = move |_| {
let game_type = selected_game.get_clone();
let opts: serde_json::Map<String, serde_json::Value> =
@@ -462,15 +481,39 @@ pub fn LobbyPage() -> View {
}
p(class="text-secondary") { "Log in with your account to play." }
// rel="external" opts out of the SPA router's click
// interception: this must be a full page navigation
// to the backend's OIDC login endpoint.
a(
class="btn btn-warning btn-lg w-100",
href="/auth/login",
rel="external",
) {
"Log in"
}
// interception: these must be full page navigations
// to the backend's OIDC login endpoints.
(move || {
let providers = auth_providers.get_clone();
if providers.is_empty() {
// Providers could not be fetched: the
// generic route redirects to the first
// configured provider.
view! {
a(
class="btn btn-warning btn-lg w-100",
href="/auth/login",
rel="external",
) {
"Log in"
}
}
} else {
view! {
div(class="d-grid gap-2") {
(providers.iter().cloned().map(|p| view! {
a(
class="btn btn-warning btn-lg w-100",
href=p.login_url,
rel="external",
) {
"Log in with " (provider_label(&p.name))
}
}).collect::<Vec<_>>())
}
}
}
})
}
}
}
+5 -2
View File
@@ -55,8 +55,11 @@ impl GameSocket {
/// could not be created.
///
/// `on_close` fires exactly once when the connection ends; it receives the
/// server close code when one was sent (e.g. 4401 unauthenticated, 4403 not
/// seated, 4404 unknown game) or `None` for an abnormal network loss.
/// server close code when one was sent (e.g. 1000 for a normal closure) or
/// `None` for an abnormal loss. A *refused* handshake is not distinguishable
/// here: browsers report it as an abnormal closure (1006) and never expose
/// the HTTP status the server answered with, so callers must classify the
/// failure out of band.
pub fn connect(
game_id: &str,
on_message: impl Fn(ServerMessage) + 'static,