woggioni
a89bd06010
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.
318 lines
13 KiB
Python
318 lines
13 KiB
Python
"""Unit tests for the database DSN assembly in :mod:`tavolo.config`.
|
|
|
|
``Settings.from_env`` is called directly with a fully replaced
|
|
``os.environ`` so no test leaks its ``DATABASE_*`` overrides into the
|
|
suite (``tests/__init__.py`` sets ``DATABASE_URL=sqlite://:memory:``
|
|
globally for the application tests).
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
from tavolo.config import Settings
|
|
|
|
|
|
def _settings(env: dict) -> Settings:
|
|
with patch.dict(os.environ, env, clear=True):
|
|
return Settings.from_env()
|
|
|
|
|
|
class DatabaseUrlTests(unittest.TestCase):
|
|
def test_defaults_assemble_from_parts(self):
|
|
settings = _settings({})
|
|
self.assertEqual(
|
|
settings.database_url,
|
|
"postgres://tavolo:password@localhost/tavolo",
|
|
)
|
|
|
|
def test_components_override_defaults(self):
|
|
settings = _settings({
|
|
"DATABASE_ENGINE": "postgres",
|
|
"DATABASE_HOST": "db.internal",
|
|
"DATABASE_PORT": "5433",
|
|
"DATABASE_NAME": "cards",
|
|
"DATABASE_USER": "scopa",
|
|
"DATABASE_PASSWORD": "s3cret",
|
|
})
|
|
self.assertEqual(
|
|
settings.database_url,
|
|
"postgres://scopa:s3cret@db.internal:5433/cards",
|
|
)
|
|
|
|
def test_options_are_appended_as_query_string(self):
|
|
settings = _settings({"DATABASE_OPTIONS": "ssl=require"})
|
|
self.assertEqual(
|
|
settings.database_url,
|
|
"postgres://tavolo:password@localhost/tavolo?ssl=require",
|
|
)
|
|
|
|
def test_options_leading_question_mark_is_stripped(self):
|
|
settings = _settings({"DATABASE_OPTIONS": "?ssl=require"})
|
|
self.assertEqual(
|
|
settings.database_url,
|
|
"postgres://tavolo:password@localhost/tavolo?ssl=require",
|
|
)
|
|
|
|
def test_credentials_are_percent_encoded(self):
|
|
settings = _settings({
|
|
"DATABASE_USER": "u@x",
|
|
"DATABASE_PASSWORD": "p@ss/word:1",
|
|
})
|
|
self.assertEqual(
|
|
settings.database_url,
|
|
"postgres://u%40x:p%40ss%2Fword%3A1@localhost/tavolo",
|
|
)
|
|
|
|
def test_database_url_takes_precedence_over_parts(self):
|
|
settings = _settings({
|
|
"DATABASE_URL": "sqlite://:memory:",
|
|
"DATABASE_HOST": "db.internal",
|
|
"DATABASE_PASSWORD": "ignored",
|
|
})
|
|
self.assertEqual(settings.database_url, "sqlite://:memory:")
|
|
|
|
def test_empty_database_url_falls_back_to_parts(self):
|
|
settings = _settings({"DATABASE_URL": ""})
|
|
self.assertEqual(
|
|
settings.database_url,
|
|
"postgres://tavolo:password@localhost/tavolo",
|
|
)
|
|
|
|
|
|
class CorsSettingsTests(unittest.TestCase):
|
|
def test_cors_disabled_by_default(self):
|
|
settings = _settings({})
|
|
self.assertIsNone(settings.cors_allow_origins)
|
|
self.assertIsNone(settings.cors_allow_origin_regex)
|
|
self.assertIsNone(settings.cors_allow_methods)
|
|
self.assertIsNone(settings.cors_allow_headers)
|
|
self.assertFalse(settings.cors_allow_credentials)
|
|
self.assertIsNone(settings.cors_expose_headers)
|
|
self.assertEqual(600, settings.cors_max_age)
|
|
|
|
def test_allow_origins_parses_comma_separated_list(self):
|
|
settings = _settings({
|
|
"CORS_ALLOW_ORIGINS": "https://a.example, https://b.example ,,https://c.example",
|
|
})
|
|
self.assertEqual(
|
|
("https://a.example", "https://b.example", "https://c.example"),
|
|
settings.cors_allow_origins,
|
|
)
|
|
|
|
def test_allow_origins_star_is_passed_through(self):
|
|
settings = _settings({"CORS_ALLOW_ORIGINS": "*"})
|
|
self.assertEqual(("*",), settings.cors_allow_origins)
|
|
|
|
def test_allow_origin_regex_is_passed_through(self):
|
|
settings = _settings({"CORS_ALLOW_ORIGIN_REGEX": r"https://.*\.example\.com"})
|
|
self.assertEqual(r"https://.*\.example\.com", settings.cors_allow_origin_regex)
|
|
|
|
def test_allow_methods_and_headers_parse_as_lists(self):
|
|
settings = _settings({
|
|
"CORS_ALLOW_METHODS": "GET,POST",
|
|
"CORS_ALLOW_HEADERS": "Authorization, X-Custom-Header",
|
|
"CORS_EXPOSE_HEADERS": "X-Total-Count",
|
|
})
|
|
self.assertEqual(("GET", "POST"), settings.cors_allow_methods)
|
|
self.assertEqual(("Authorization", "X-Custom-Header"), settings.cors_allow_headers)
|
|
self.assertEqual(("X-Total-Count",), settings.cors_expose_headers)
|
|
|
|
def test_allow_credentials_parses_boolean(self):
|
|
for value in ("1", "true", "TRUE", "yes", "on"):
|
|
self.assertTrue(_settings({"CORS_ALLOW_CREDENTIALS": value}).cors_allow_credentials)
|
|
for value in ("0", "false", "no", "off", "anything-else"):
|
|
self.assertFalse(_settings({"CORS_ALLOW_CREDENTIALS": value}).cors_allow_credentials)
|
|
|
|
def test_max_age_parses_int(self):
|
|
settings = _settings({"CORS_MAX_AGE": "3600"})
|
|
self.assertEqual(3600, settings.cors_max_age)
|
|
|
|
|
|
class OTelSettingsTests(unittest.TestCase):
|
|
def test_otel_disabled_by_default(self):
|
|
settings = _settings({})
|
|
self.assertFalse(settings.otel_enabled)
|
|
self.assertEqual("tavolo", settings.otel_service_name)
|
|
self.assertIsNone(settings.otel_exporter_endpoint)
|
|
self.assertIsNone(settings.otel_exporter_headers)
|
|
|
|
def test_otel_enabled_parses_boolean(self):
|
|
for value in ("1", "true", "TRUE", "yes", "on"):
|
|
self.assertTrue(_settings({"OTEL_ENABLED": value}).otel_enabled)
|
|
for value in ("0", "false", "no", "off", "anything-else"):
|
|
self.assertFalse(_settings({"OTEL_ENABLED": value}).otel_enabled)
|
|
|
|
def test_otel_settings_are_passed_through(self):
|
|
settings = _settings({
|
|
"OTEL_SERVICE_NAME": "cards",
|
|
"OTEL_EXPORTER_OTLP_ENDPOINT": "http://collector:4318",
|
|
"OTEL_EXPORTER_OTLP_HEADERS": "Authorization=Bearer t, X-Tenant=one",
|
|
})
|
|
self.assertEqual("cards", settings.otel_service_name)
|
|
self.assertEqual("http://collector:4318", settings.otel_exporter_endpoint)
|
|
self.assertEqual(
|
|
("Authorization=Bearer t", "X-Tenant=one"),
|
|
settings.otel_exporter_headers,
|
|
)
|
|
|
|
def test_otel_excluded_paths_defaults_to_health_endpoint(self):
|
|
self.assertEqual(("/api/health",), _settings({}).otel_excluded_paths)
|
|
|
|
def test_otel_excluded_paths_parses_comma_separated_list(self):
|
|
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()
|