tests/core/test_config.py¶
Source from this local checkout, regenerated when the reader rebuilds.
Line links use #L<number>; a GitHub line range opens its first line.
1 # Copyright 2025 Softwell S.r.l.2 #3 # Licensed under the Apache License, Version 2.0 (the "License");4 # you may not use this file except in compliance with the License.5 # You may obtain a copy of the License at6 #7 # https://www.apache.org/licenses/LICENSE-2.08 #9 # Unless required by applicable law or agreed to in writing, software10 # distributed under the License is distributed on an "AS IS" BASIS,11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.12 # See the License for the specific language governing permissions and13 # limitations under the License.14 15 """Config tests: the ``asgiconfig`` dialect, the read door, the self-configuring server.16 17 A recipe subclasses ``AsgiConfigBuilder`` and declares the site sections under18 one ``configuration`` root; ``AsgiServer(config=source)`` builds its own19 ``ConfigurationHandler`` over that source, derives its kwargs from it and stays20 reachable as ``server.config``. Requests are driven at the ASGI level (no21 uvicorn), the same style as ``test_session.py``.22 """23 24 from __future__ import annotations25 26 import base6427 from pathlib import Path28 from typing import Any29 30 from cryptography.fernet import Fernet31 import pytest32 from genro_bag.resolvers import EnvResolver33 from genro_storage import StorageManager34 35 from genro_asgi import (36 AsgiConfigBuilder,37 AsgiServer,38 BaseApplication,39 ConfigError,40 ConfigurationHandler,41 )42 from genro_asgi.__main__ import AppsRegistry43 from genro_asgi.config import HOME_ENV, BaseConfiguration, DefaultConfig44 from genro_asgi.exceptions import HTTPUnauthorized45 from genro_asgi.middleware.base import BaseMiddleware46 from genro_asgi.storage_mixin import DEFAULT_SITE_MOUNT47 from genro_asgi.types import Message, Receive, Scope, Send48 49 ADMIN_PW_ENV_VAR = "GENRO_TEST_ADMIN_PW"50 OIDC_SECRET_ENV_VAR = "GENRO_TEST_OIDC_SECRET"51 52 53 class ShopApp(BaseApplication):54 """Root app: answers ``shop``."""55 56 async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:57 await send({"type": "http.response.start", "status": 200, "headers": []})58 await send({"type": "http.response.body", "body": b"shop"})59 60 61 class ApiApp(BaseApplication):62 """Secondary app: answers ``api``."""63 64 async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:65 await send({"type": "http.response.start", "status": 200, "headers": []})66 await send({"type": "http.response.body", "body": b"api"})67 68 69 class TwoAppConfig(AsgiConfigBuilder):70 """Two apps (shop on the root, api secondary), cors + basic auth, host/port.71 72 Declares ``external_url`` too — the whole-site recipe of these tests, and a73 site that configures an OIDC provider must name its public base address (the74 absolute ``redirect_uri`` prefix) or the server refuses to boot. The75 without-``external_url`` case is covered on its own in ``test_oidc.py``.76 """77 78 def main(self, root: Any) -> None:79 cfg = root.configuration()80 cfg.server(host="0.0.0.0", port=9100, external_url="https://shop.example.com")81 cfg.middleware(cors=True)82 self.authentication_section(cfg)83 self.applications_section(cfg)84 85 def authentication_section(self, cfg: Any) -> None:86 """One Basic user, handed to ``AuthCore`` through ``credentials``."""87 creds = cfg.authentication().credentials()88 creds.basic_user(username="admin", password="secret", tags="admin")89 90 def applications_section(self, cfg: Any) -> None:91 """``shop`` claims the site root, ``api`` answers its own mount."""92 apps = cfg.applications(default="shop")93 apps.application(code="shop", mount="", app_class=ShopApp)94 apps.application(code="api", app_class=ApiApp)95 96 97 def chain_types(server: AsgiServer) -> list[str]:98 """The class names of the middlewares in the server's chain, outermost first."""99 names: list[str] = []100 node: object = server.middleware_chain101 while isinstance(node, BaseMiddleware):102 names.append(type(node).__name__)103 node = node.app104 return names105 106 107 def basic_header(username: str, password: str) -> list[tuple[bytes, bytes]]:108 """An ``Authorization: Basic`` header list for the given credentials."""109 token = base64.b64encode(f"{username}:{password}".encode()).decode()110 return [(b"authorization", f"Basic {token}".encode())]111 112 113 async def http_get(server: AsgiServer, path: str) -> bytes:114 """Drive one GET through ``server`` at the ASGI level; return the response body."""115 scope: Scope = {"type": "http", "method": "GET", "path": path, "headers": []}116 sent: list[Message] = []117 118 async def receive() -> Message:119 return {"type": "http.request"}120 121 async def send(message: Message) -> None:122 sent.append(message)123 124 await server(scope, receive, send)125 return b"".join(m.get("body", b"") for m in sent if m["type"] == "http.response.body")126 127 128 async def http_status_headers(129 server: AsgiServer, path: str130 ) -> tuple[int, list[tuple[bytes, bytes]]]:131 """Drive one GET through ``server``; return its status and response headers."""132 scope: Scope = {"type": "http", "method": "GET", "path": path, "headers": []}133 sent: list[Message] = []134 135 async def receive() -> Message:136 return {"type": "http.request"}137 138 async def send(message: Message) -> None:139 sent.append(message)140 141 await server(scope, receive, send)142 start = next(m for m in sent if m["type"] == "http.response.start")143 return start["status"], start["headers"]144 145 146 class TestSelfConfiguringServer:147 def test_server_section_reaches_the_serve_defaults(self) -> None:148 server = AsgiServer(config=TwoAppConfig)149 assert server.config_host == "0.0.0.0"150 assert server.config_port == 9100151 assert server.external_url == "https://shop.example.com"152 153 def test_default_app_answers_the_root_others_are_mounts(self) -> None:154 server = AsgiServer(config=TwoAppConfig)155 assert isinstance(server.root_application, ShopApp)156 assert server.root_application.mount == ""157 assert set(server.applications) == {"shop", "api", "_server"}158 assert isinstance(server.applications["api"], ApiApp)159 160 def test_a_bare_server_has_no_configuration(self) -> None:161 assert AsgiServer(applications=[ShopApp(mount="")]).config is None162 163 def test_the_handler_stays_reachable_as_the_read_door(self) -> None:164 server = AsgiServer(config=TwoAppConfig)165 assert isinstance(server.config, ConfigurationHandler)166 assert server.config("server.host") == "0.0.0.0"167 assert server.config("server.port") == 9100168 169 def test_an_explicit_kwarg_wins_over_the_configured_one(self) -> None:170 server = AsgiServer(config=TwoAppConfig, port=0)171 assert server.config_port == 0172 assert server.config_host == "0.0.0.0" # untouched kwargs still apply173 174 def test_a_recipe_instance_is_accepted(self) -> None:175 assert AsgiServer(config=TwoAppConfig(name="site")).config_port == 9100176 177 def test_a_ready_handler_is_adopted_as_is(self) -> None:178 handler = ConfigurationHandler(TwoAppConfig)179 server = AsgiServer(config=handler)180 assert server.config is handler181 182 def test_a_config_py_path_is_loaded(self, tmp_path: Path) -> None:183 module = tmp_path / "config.py"184 module.write_text(185 "from genro_asgi.config import AsgiConfigBuilder\n"186 "\n"187 "\n"188 "class ServerConfiguration(AsgiConfigBuilder):\n"189 " def main(self, root):\n"190 " cfg = root.configuration()\n"191 " cfg.server(host='127.0.0.1', port=8123)\n"192 )193 server = AsgiServer(config=module)194 assert server.config_port == 8123195 assert set(server.applications) == {"_server"}196 197 198 class TestDemux:199 async def test_serves_both_apps(self) -> None:200 server = AsgiServer(config=TwoAppConfig)201 assert await http_get(server, "/") == b"shop"202 assert await http_get(server, "/api") == b"api"203 204 205 class TestMiddlewareChain:206 def test_chain_contains_cors_and_errors(self) -> None:207 types = chain_types(AsgiServer(config=TwoAppConfig))208 assert "CORSMiddleware" in types209 assert "ErrorMiddleware" in types210 211 def test_an_explicit_switch_off_survives_the_read(self) -> None:212 class NoCorsConfig(AsgiConfigBuilder):213 def main(self, root: Any) -> None:214 cfg = root.configuration()215 cfg.middleware(cors=False)216 cfg.applications().application(code="shop", mount="", app_class=ShopApp)217 218 assert "CORSMiddleware" not in chain_types(AsgiServer(config=NoCorsConfig))219 220 221 class TestCredentials:222 def test_basic_user_is_verified_by_the_auth_core(self) -> None:223 server = AsgiServer(config=TwoAppConfig)224 scope: Scope = {"headers": basic_header("admin", "secret")}225 avatar = server.authenticate(scope)226 assert avatar is not None227 assert avatar.identity == "admin"228 assert "admin" in avatar.tags229 230 def test_wrong_password_raises_unauthorized(self) -> None:231 server = AsgiServer(config=TwoAppConfig)232 scope: Scope = {"headers": basic_header("admin", "wrong")}233 with pytest.raises(HTTPUnauthorized):234 server.authenticate(scope)235 236 def test_bearer_token_is_verified_by_its_identity(self) -> None:237 class BearerConfig(AsgiConfigBuilder):238 def main(self, root: Any) -> None:239 cfg = root.configuration()240 creds = cfg.authentication().credentials()241 creds.bearer_token(identity="svc", token="sk_live_xyz", tags="api")242 cfg.applications().application(code="shop", mount="", app_class=ShopApp)243 244 server = AsgiServer(config=BearerConfig)245 scope: Scope = {"headers": [(b"authorization", b"Bearer sk_live_xyz")]}246 avatar = server.authenticate(scope)247 assert avatar is not None248 assert avatar.identity == "svc"249 assert avatar.tags == ["api"]250 251 def test_jwt_entries_stay_an_ordered_list(self) -> None:252 class JwtConfig(AsgiConfigBuilder):253 def main(self, root: Any) -> None:254 cfg = root.configuration()255 creds = cfg.authentication().credentials()256 creds.jwt(name="hmac", secret="topsecret")257 creds.jwt(name="rsa", public_key="PUBKEY", algorithm="RS256")258 cfg.applications().application(code="shop", mount="", app_class=ShopApp)259 260 entries = ConfigurationHandler(JwtConfig).auth_entries()261 assert [entry["name"] for entry in entries["jwt"]] == ["hmac", "rsa"]262 assert entries["jwt"][0]["algorithm"] == "HS256" # signature default263 assert entries["jwt"][1]["public_key"] == "PUBKEY"264 265 def test_no_credentials_section_arms_no_backend(self) -> None:266 class BareConfig(AsgiConfigBuilder):267 def main(self, root: Any) -> None:268 root.configuration().applications().application(269 code="shop", mount="", app_class=ShopApp270 )271 272 assert ConfigurationHandler(BareConfig).auth_entries() is None273 274 275 class TestSession:276 async def test_session_attached_after_a_request(self) -> None:277 server = AsgiServer(config=TwoAppConfig)278 scope: Scope = {"type": "http", "method": "GET", "path": "/", "headers": []}279 sent: list[Message] = []280 281 async def receive() -> Message:282 return {"type": "http.request"}283 284 async def send(message: Message) -> None:285 sent.append(message)286 287 await server(scope, receive, send)288 assert scope.get("session") is not None289 assert server.session(scope) is scope["session"]290 291 def test_session_child_reaches_the_store_ttl(self) -> None:292 class SessionConfig(AsgiConfigBuilder):293 def main(self, root: Any) -> None:294 cfg = root.configuration()295 cfg.server(host="127.0.0.1", port=8000).session(ttl=1234)296 cfg.applications().application(code="shop", mount="", app_class=ShopApp)297 298 server = AsgiServer(config=SessionConfig)299 assert server.session_store.create().meta["ttl"] == 1234300 301 302 class TestShutdownTimeout:303 async def test_recipe_shutdown_timeout_reaches_uvicorn(self, monkeypatch) -> None:304 # One endless response (an SSE stream a client never closes) used to hold305 # uvicorn's shutdown for ever, so the lifespan shutdown never ran and the306 # applications were never stopped (measured 2026-09-08). The bound is a307 # server setpoint and travels to uvicorn as timeout_graceful_shutdown.308 class BoundedConfig(AsgiConfigBuilder):309 def main(self, root: Any) -> None:310 cfg = root.configuration()311 cfg.server(host="127.0.0.1", port=8000, shutdown_timeout_seconds=2)312 cfg.applications().application(code="shop", mount="", app_class=ShopApp)313 314 captured: list[Any] = []315 316 class XT_Server:317 def __init__(self, config: Any) -> None:318 captured.append(config)319 320 def run(self) -> None:321 pass322 323 import genro_asgi.server as server_module324 325 monkeypatch.setattr(server_module.uvicorn, "Server", XT_Server)326 server = AsgiServer(config=BoundedConfig)327 assert server.shutdown_timeout_seconds == 2.0328 server.serve()329 assert captured[0].timeout_graceful_shutdown == 2.0330 331 def test_the_default_is_five_seconds(self) -> None:332 assert AsgiServer(config=TwoAppConfig).shutdown_timeout_seconds == 5.0333 334 335 class TestMaxThreads:336 async def test_recipe_max_threads_reaches_the_pool(self) -> None:337 class SizedPoolConfig(AsgiConfigBuilder):338 def main(self, root: Any) -> None:339 cfg = root.configuration()340 cfg.server(host="127.0.0.1", port=8000, max_threads=2)341 cfg.applications().application(code="shop", mount="", app_class=ShopApp)342 343 server = AsgiServer(config=SizedPoolConfig)344 await server.run_sync(lambda: None)345 assert server.pool.executor._max_workers == 2346 347 348 class TestGrammarValidation:349 def test_unknown_tag_raises(self) -> None:350 class BadConfig(AsgiConfigBuilder):351 def main(self, root: Any) -> None:352 root.configuration().nonexistent(foo=1)353 354 with pytest.raises(AttributeError):355 ConfigurationHandler(BadConfig)356 357 def test_a_section_outside_the_root_is_rejected(self) -> None:358 class LooseConfig(AsgiConfigBuilder):359 def main(self, root: Any) -> None:360 root.server(host="127.0.0.1")361 362 with pytest.raises(ValueError, match="parent_tags"):363 ConfigurationHandler(LooseConfig)364 365 def test_application_without_app_class_rejected_by_grammar(self) -> None:366 class NoClassConfig(AsgiConfigBuilder):367 def main(self, root: Any) -> None:368 root.configuration().applications().application(code="shop")369 370 with pytest.raises(ValueError, match="app_class"):371 ConfigurationHandler(NoClassConfig)372 373 def test_a_mount_without_base_path_is_rejected_by_the_foreign_grammar(self) -> None:374 # The storage subtree is validated by genro-storage's own signatures,375 # not by this dialect: the error comes from THERE.376 class NoBasePathConfig(AsgiConfigBuilder):377 def main(self, root: Any) -> None:378 root.configuration().storage(app=StorageManager).local(name="data")379 380 with pytest.raises(ValueError, match="base_path"):381 ConfigurationHandler(NoBasePathConfig)382 383 def test_storage_without_app_is_rejected_by_the_grammar(self) -> None:384 # ``app`` cannot be defaulted in the signature: the subbuilder385 # reference reads the CALL SITE, so an omitted ``app`` would silently386 # leave the node a leaf of this dialect. It is required instead.387 class NoAppConfig(AsgiConfigBuilder):388 def main(self, root: Any) -> None:389 root.configuration().storage()390 391 with pytest.raises(ValueError, match="app"):392 ConfigurationHandler(NoAppConfig)393 394 def test_an_empty_application_code_is_a_boot_error(self) -> None:395 # code="" would file the subtree under an empty label while the app396 # registers under its class-name fallback: the read door would then397 # never reach the written values, so the fold refuses to boot.398 class EmptyCodeConfig(AsgiConfigBuilder):399 def main(self, root: Any) -> None:400 cfg = root.configuration()401 cfg.applications().application(code="", mount="", app_class=ShopApp)402 403 with pytest.raises(ConfigError, match="non-empty"):404 AsgiServer(config=EmptyCodeConfig)405 406 def test_a_second_server_section_is_rejected(self) -> None:407 class TwiceConfig(AsgiConfigBuilder):408 def main(self, root: Any) -> None:409 cfg = root.configuration()410 cfg.server(host="127.0.0.1")411 cfg.server(host="0.0.0.0")412 413 with pytest.raises(ValueError):414 ConfigurationHandler(TwiceConfig)415 416 417 class TestSkippedSections:418 def test_openapi_and_databases_boot_without_error(self) -> None:419 class OrchestrationConfig(AsgiConfigBuilder):420 def main(self, root: Any) -> None:421 cfg = root.configuration()422 cfg.server(host="127.0.0.1", port=8000)423 cfg.applications(default="shop").application(424 code="shop", mount="", app_class=ShopApp425 )426 cfg.databases().database(code="default", db_class=object)427 cfg.openapi(title="Demo", version="1.0")428 429 server = AsgiServer(config=OrchestrationConfig)430 assert isinstance(server.root_application, ShopApp)431 assert set(server.applications) == {"shop", "_server"}432 assert server.config("openapi.title") == "Demo"433 434 435 class TestSingleAppNoDefault:436 def test_lone_app_answers_its_own_mount_not_the_root(self) -> None:437 # Nothing elects an application: a lone app derives its mount from its438 # code like every other, so the site root stays unclaimed.439 class OneAppConfig(AsgiConfigBuilder):440 def main(self, root: Any) -> None:441 root.configuration().applications().application(442 code="only", app_class=ShopApp443 )444 445 server = AsgiServer(config=OneAppConfig)446 assert set(server.applications) == {"only", "_server"}447 assert server.root_application is None448 assert isinstance(server.application_at("only"), ShopApp)449 450 def test_lone_app_claims_the_root_by_declaring_an_empty_mount(self) -> None:451 # The compatibility mechanism: one app served at unchanged URLs.452 class RootAppConfig(AsgiConfigBuilder):453 def main(self, root: Any) -> None:454 root.configuration().applications().application(455 code="only", mount="", app_class=ShopApp456 )457 458 server = AsgiServer(config=RootAppConfig)459 assert isinstance(server.root_application, ShopApp)460 assert server.root_application.code == "only"461 462 463 class TestDefaultRedirect:464 async def test_root_redirects_to_the_default_when_nobody_claims_it(self) -> None:465 class MountsOnlyConfig(AsgiConfigBuilder):466 def main(self, root: Any) -> None:467 apps = root.configuration().applications(default="shop")468 apps.application(code="shop", app_class=ShopApp)469 apps.application(code="api", app_class=ApiApp)470 471 server = AsgiServer(config=MountsOnlyConfig)472 assert server.root_application is None473 assert server.default_application is server.applications["shop"]474 status, headers = await http_status_headers(server, "/")475 assert status == 307476 assert dict(headers)[b"location"] == b"/shop/"477 478 def test_a_default_naming_no_application_is_a_boot_error(self) -> None:479 class GhostDefaultConfig(AsgiConfigBuilder):480 def main(self, root: Any) -> None:481 root.configuration().applications(default="ghost").application(482 code="shop", app_class=ShopApp483 )484 485 with pytest.raises(ValueError, match="ghost"):486 AsgiServer(config=GhostDefaultConfig)487 488 489 def storage_site_config(base_path: Path) -> type[AsgiConfigBuilder]:490 """A site recipe with an ``idstore`` mount and the key the credential stores need."""491 492 class StorageSiteConfig(AsgiConfigBuilder):493 def setup(self, data: Any) -> None:494 """The mount path travels through the datastore, not a closure."""495 data["base_path"] = str(base_path)496 497 def main(self, root: Any) -> None:498 cfg = root.configuration()499 cfg.server(host="127.0.0.1", port=8000)500 cfg.storage(501 app=StorageManager, storage_key=Fernet.generate_key().decode()502 ).local(name="idstore", base_path=self.data["base_path"])503 cfg.applications(default="shop").application(504 code="shop", mount="", app_class=ShopApp505 )506 self.identity_section(cfg)507 508 def identity_section(self, cfg: Any) -> None:509 """Bootstrap admin plus both identity stores on ``idstore``."""510 auth = cfg.authentication()511 auth.admin_password(EnvResolver(ADMIN_PW_ENV_VAR))512 auth.users(mount="idstore", prefix="users")513 auth.tokens(mount="idstore", prefix="api_keys")514 515 return StorageSiteConfig516 517 518 class TestStorageSection:519 """``storage`` → genro-storage's own ``list[dict]`` plus the section key."""520 521 def test_the_section_flattens_to_genro_storage_configuration(522 self, monkeypatch: pytest.MonkeyPatch523 ) -> None:524 monkeypatch.setenv("GENRO_TEST_STORAGE_KEY", "k1,k2")525 526 class StorageConfig(AsgiConfigBuilder):527 def main(self, root: Any) -> None:528 self.storage_section(root.configuration())529 530 def storage_section(self, cfg: Any) -> None:531 s = cfg.storage(532 app=StorageManager,533 storage_key=EnvResolver("GENRO_TEST_STORAGE_KEY"),534 )535 s.local(name="site", base_path=".")536 s.s3(name="uploads", bucket="shop-media", default_encrypted="shopspa")537 538 mounts, storage_key = ConfigurationHandler(StorageConfig).storage_config()539 assert storage_key == "k1,k2"540 assert mounts == [541 {"name": "site", "protocol": "local", "base_path": "."},542 {543 "name": "uploads",544 "protocol": "s3",545 "bucket": "shop-media",546 "default_encrypted": "shopspa",547 },548 ]549 550 def test_no_storage_section_leaves_the_default_manager(self) -> None:551 assert ConfigurationHandler(TwoAppConfig).storage_config() is None552 553 554 class TestIdentitySection:555 """``authentication`` → the identity kwargs ``AuthMixin`` peels."""556 557 def test_no_identity_configured_leaves_the_stores_unwired(self) -> None:558 server = AsgiServer(config=TwoAppConfig)559 assert server.user_store is None560 assert server.api_key_store is None561 562 def test_stores_and_bootstrap_admin_reach_the_server(563 self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch564 ) -> None:565 monkeypatch.setenv(ADMIN_PW_ENV_VAR, "s3cret")566 server = AsgiServer(config=storage_site_config(tmp_path))567 assert server.user_store is not None568 assert server.api_key_store is not None569 admin = server.user_store.get("admin")570 assert admin is not None571 assert "SUPERADMIN" in admin["tags"]572 573 def test_admin_password_literal_is_rejected_by_the_grammar(self) -> None:574 class LiteralConfig(AsgiConfigBuilder):575 def main(self, root: Any) -> None:576 root.configuration().authentication().admin_password("plain-secret")577 578 with pytest.raises(ValueError, match="node_value"):579 AsgiServer(config=LiteralConfig)580 581 def test_admin_password_resolving_empty_is_a_boot_error(582 self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch583 ) -> None:584 monkeypatch.delenv(ADMIN_PW_ENV_VAR, raising=False)585 with pytest.raises(ConfigError, match="resolved empty"):586 AsgiServer(config=storage_site_config(tmp_path))587 588 def test_a_second_users_element_is_rejected_by_the_grammar(self) -> None:589 class DoubledConfig(AsgiConfigBuilder):590 def main(self, root: Any) -> None:591 auth = root.configuration().authentication()592 auth.users(mount="one")593 auth.users(mount="two")594 595 with pytest.raises(ValueError):596 ConfigurationHandler(DoubledConfig)597 598 599 class LoginSurfaceConfig(TwoAppConfig):600 """The two-app site plus a lockout policy and two OIDC providers."""601 602 def authentication_section(self, cfg: Any) -> None:603 """The login surface: policy, one confidential and one public provider."""604 auth = cfg.authentication()605 auth.login(max_attempts=3, backoff=10)606 oidc = auth.oidc()607 oidc.provider(608 code="corp",609 issuer="https://idp.example.com",610 client_id="corp-client",611 client_secret=EnvResolver(OIDC_SECRET_ENV_VAR),612 scopes="openid profile",613 identity_claim="preferred_username",614 tags=["staff"],615 )616 oidc.provider(617 code="public",618 issuer="https://accounts.example.org",619 client_id="pub-client",620 )621 622 623 class TestLoginSurface:624 """``authentication.login``/``.oidc`` → ``server_app=`` → the ``_server`` app."""625 626 def test_the_login_surface_reaches_the_server_app(627 self, monkeypatch: pytest.MonkeyPatch628 ) -> None:629 monkeypatch.setenv(OIDC_SECRET_ENV_VAR, "oidc-s3cret")630 app = AsgiServer(config=LoginSurfaceConfig).applications["_server"]631 assert app.login_policy == {"max_attempts": 3, "backoff": 10}632 assert set(app.oidc_providers) == {"corp", "public"}633 corp = app.oidc_providers["corp"]634 assert corp["client_secret"] == "oidc-s3cret"635 assert corp["scopes"] == "openid profile"636 assert corp["identity_claim"] == "preferred_username"637 assert corp["tags"] == ["staff"]638 639 def test_provider_defaults_apply_per_provider(640 self, monkeypatch: pytest.MonkeyPatch641 ) -> None:642 monkeypatch.setenv(OIDC_SECRET_ENV_VAR, "oidc-s3cret")643 app = AsgiServer(config=LoginSurfaceConfig).applications["_server"]644 assert app.oidc_providers["public"] == {645 "issuer": "https://accounts.example.org",646 "client_id": "pub-client",647 "scopes": "openid email profile",648 "identity_claim": "email",649 "tags": [],650 }651 652 def test_no_login_section_leaves_the_bare_app(self) -> None:653 app = AsgiServer(config=TwoAppConfig).applications["_server"]654 assert app.login_policy == {}655 assert app.oidc_providers == {}656 657 def test_a_provider_without_a_code_is_rejected_by_the_collection(self) -> None:658 class NoCodeConfig(AsgiConfigBuilder):659 def main(self, root: Any) -> None:660 root.configuration().authentication().oidc().provider(661 issuer="https://idp.example.com", client_id="x"662 )663 664 with pytest.raises(ValueError, match="code"):665 ConfigurationHandler(NoCodeConfig)666 667 def test_a_duplicate_provider_code_is_rejected_by_the_collection(self) -> None:668 class DoubledCodeConfig(AsgiConfigBuilder):669 def main(self, root: Any) -> None:670 oidc = root.configuration().authentication().oidc()671 oidc.provider(code="corp", issuer="https://a.example.com", client_id="a")672 oidc.provider(code="corp", issuer="https://b.example.com", client_id="b")673 674 with pytest.raises(ValueError, match="corp"):675 ConfigurationHandler(DoubledCodeConfig)676 677 678 class TestTasksConfig:679 """The ``tasks()`` child of ``server`` lifts to the ``tasks=`` kwarg."""680 681 def test_tasks_disabled_via_recipe(self) -> None:682 class TasksOffConfig(AsgiConfigBuilder):683 def main(self, root: Any) -> None:684 cfg = root.configuration()685 cfg.server(host="127.0.0.1", port=8000).tasks(enabled=False)686 cfg.applications().application(code="shop", mount="", app_class=ShopApp)687 688 server = AsgiServer(config=TasksOffConfig)689 assert server.tasks_enabled is False690 with pytest.raises(RuntimeError, match="disabled"):691 server.tasks692 693 def test_tuning_reaches_scheduler_and_store(self) -> None:694 class TunedConfig(AsgiConfigBuilder):695 def main(self, root: Any) -> None:696 cfg = root.configuration()697 cfg.server(host="127.0.0.1", port=8000).tasks(tick_seconds=5, mount="site")698 cfg.applications().application(code="shop", mount="", app_class=ShopApp)699 700 server = AsgiServer(config=TunedConfig)701 assert server.tasks_enabled is True # enabled defaults on702 assert server.tasks.scheduler.tick_seconds == 5.0703 assert server.tasks.task_store.mount == "site" # explicit override704 705 def test_direct_dict_kwarg(self) -> None:706 server = AsgiServer(applications=[ShopApp(mount="")],707 tasks={"enabled": True, "tick_seconds": 3})708 assert server.tasks.scheduler.tick_seconds == 3.0709 assert server.tasks_config == {"tick_seconds": 3} # enabled peeled away710 711 def test_a_child_under_tasks_is_rejected_by_the_grammar(self) -> None:712 class StrayChildConfig(AsgiConfigBuilder):713 def main(self, root: Any) -> None:714 cfg = root.configuration()715 cfg.server(host="127.0.0.1", port=8000).tasks().middleware()716 717 with pytest.raises(ValueError, match="parent"):718 ConfigurationHandler(StrayChildConfig)719 720 721 class ParametrizedShop(ShopApp):722 """An app whose grammar is the inherited minimal one (``parameters``)."""723 724 code = "shop"725 726 727 class TestMountedAppGrammar:728 """``application(app_class=...)`` mounts ``app_class.grammar`` for the subtree."""729 730 def test_the_apps_own_subtree_is_read_through_the_handler(self) -> None:731 class ParamConfig(AsgiConfigBuilder):732 def main(self, root: Any) -> None:733 cfg = root.configuration()734 cfg.server(host="127.0.0.1", port=8000)735 shop = cfg.applications(default="shop").application(736 code="shop", mount="", app_class=ParametrizedShop737 )738 shop.parameters(theme="dark", max_items=10)739 740 server = AsgiServer(config=ParamConfig)741 assert server.config("applications.shop.parameters.theme") == "dark"742 assert server.config("applications.shop.parameters.max_items") == 10743 744 def test_the_envelope_attributes_are_the_apps_constructor_kwargs(self) -> None:745 class KwargConfig(AsgiConfigBuilder):746 def main(self, root: Any) -> None:747 root.configuration().applications().application(748 code="outlet", mount="outlet", app_class=ShopApp749 )750 751 server = AsgiServer(config=KwargConfig)752 assert server.applications["outlet"].mount == "outlet"753 754 def test_an_undeclared_child_of_the_mounted_grammar_raises(self) -> None:755 class StrayConfig(AsgiConfigBuilder):756 def main(self, root: Any) -> None:757 shop = root.configuration().applications().application(758 code="shop", mount="", app_class=ShopApp759 )760 shop.catalog(title="x")761 762 with pytest.raises(AttributeError):763 ConfigurationHandler(StrayConfig)764 765 766 class TestReadStack:767 """The four layers, on this dialect."""768 769 def test_written_value_wins(self) -> None:770 assert ConfigurationHandler(TwoAppConfig)("server.port") == 9100771 772 def test_signature_default_is_resolved_at_read_time(self) -> None:773 class ProviderConfig(AsgiConfigBuilder):774 def main(self, root: Any) -> None:775 root.configuration().authentication().oidc().provider(776 code="corp", issuer="https://idp.example.com"777 )778 779 handler = ConfigurationHandler(ProviderConfig)780 assert handler("authentication.oidc.corp.scopes") == "openid email profile"781 782 def test_call_site_default_applies_to_an_unwritten_value(self) -> None:783 handler = ConfigurationHandler(TwoAppConfig)784 assert handler("server.max_threads", default=7) == 7785 786 def test_a_missing_path_is_a_noisy_key_error(self) -> None:787 handler = ConfigurationHandler(TwoAppConfig)788 with pytest.raises(KeyError, match="server.tls"):789 handler("server.tls")790 791 792 def write_defaults_recipe(793 base_dir: Path, filename: str = "config.py", mount_path: str = "/srv/deployment"794 ) -> Path:795 """A recipe file in *base_dir* deviating from the package defaults.796 797 ``mount_path`` needs to be a directory that EXISTS only where the recipe798 reaches a real ``StorageManager`` — genro-storage's local backend validates799 the anchor at ``configure()`` time, never at recipe time.800 """801 path = base_dir / filename802 path.write_text(803 "from typing import Any\n"804 "\n"805 "from genro_asgi.config import BaseConfiguration\n"806 "\n"807 "\n"808 "class DeploymentConfiguration(BaseConfiguration):\n"809 " def server_section(self, cfg: Any) -> None:\n"810 " cfg.server(host='10.0.0.1', port=9999)\n"811 "\n"812 " def storage_mounts(self, section: Any) -> None:\n"813 f" section.local(name='site', base_path={mount_path!r})\n",814 encoding="utf-8",815 )816 return path817 818 819 class TestParentRecipes:820 """``BaseConfiguration`` + the declared defaults layer + the site recipe.821 822 ``DefaultConfig.parents_for()`` is what ``AsgiServer`` hands the handler: the823 package defaults lowest, the recipe's own defaults source over them, the site824 recipe last and winning.825 """826 827 def test_a_site_inherits_the_default_site_mount_and_adds_its_key(828 self, tmp_path: Path829 ) -> None:830 class KeyOnlyConfig(BaseConfiguration):831 storage_key = "k1"832 833 parents = DefaultConfig(tmp_path).parents_for(KeyOnlyConfig)834 mounts, storage_key = ConfigurationHandler(KeyOnlyConfig, parents=parents).storage_config()835 assert storage_key == "k1"836 assert mounts == [{**DEFAULT_SITE_MOUNT, "base_path": str(Path.cwd())}]837 838 def test_only_the_package_defaults_are_layered_without_a_defaults_recipe(839 self, tmp_path: Path840 ) -> None:841 assert DefaultConfig(tmp_path).parents_for(BaseConfiguration) == [BaseConfiguration]842 843 def test_the_conventional_recipe_joins_the_chain_when_its_file_exists(844 self, tmp_path: Path845 ) -> None:846 declared = write_defaults_recipe(tmp_path)847 assert DefaultConfig(tmp_path).parents_for(BaseConfiguration) == [848 BaseConfiguration,849 declared,850 ]851 852 def test_the_defaults_layer_overrides_the_base_and_loses_to_the_site(853 self, tmp_path: Path854 ) -> None:855 write_defaults_recipe(tmp_path)856 857 class SiteConfig(AsgiConfigBuilder):858 """Says one thing only: the layers under it supply everything else."""859 860 def main(self, root: Any) -> None:861 root.configuration().server(host="127.0.0.1")862 863 parents = DefaultConfig(tmp_path).parents_for(SiteConfig)864 handler = ConfigurationHandler(SiteConfig, parents=parents)865 assert handler("server.host") == "127.0.0.1" # the site wins866 assert handler("server.port") == 9999 # the defaults layer holds867 mounts, _ = handler.storage_config() # over the package default868 assert mounts == [{**DEFAULT_SITE_MOUNT, "base_path": "/srv/deployment"}]869 870 def test_a_key_only_section_without_parents_yields_no_mount(self) -> None:871 """The guard: a storage section with no mount child is not a crash."""872 873 class KeyOnlyConfig(AsgiConfigBuilder):874 def main(self, root: Any) -> None:875 root.configuration().storage(app=StorageManager, storage_key="k1")876 877 mounts, storage_key = ConfigurationHandler(KeyOnlyConfig).storage_config()878 assert mounts == []879 assert storage_key == "k1"880 881 882 class TestDeclaredDefaultConfig:883 """``default_config`` on the recipe: which defaults source, declared by the recipe.884 885 Unset (or ``True``) takes the conventional ``<base_dir>/config.py`` when it is886 there, ``False`` takes nothing, a path takes that file and must find it.887 """888 889 def test_unset_takes_the_conventional_file(self, tmp_path: Path) -> None:890 declared = write_defaults_recipe(tmp_path)891 892 class SiteConfig(BaseConfiguration):893 pass894 895 assert SiteConfig.default_config is None896 assert DefaultConfig(tmp_path).parents_for(SiteConfig) == [BaseConfiguration, declared]897 898 def test_true_reads_the_conventional_file_like_an_unset_attribute(899 self, tmp_path: Path900 ) -> None:901 declared = write_defaults_recipe(tmp_path)902 903 class SiteConfig(BaseConfiguration):904 default_config = True905 906 assert DefaultConfig(tmp_path).parents_for(SiteConfig) == [BaseConfiguration, declared]907 908 def test_false_refuses_the_layer_even_when_the_file_is_there(self, tmp_path: Path) -> None:909 write_defaults_recipe(tmp_path)910 911 class SiteConfig(BaseConfiguration):912 default_config = False913 914 assert DefaultConfig(tmp_path).parents_for(SiteConfig) == [BaseConfiguration]915 916 def test_an_explicit_path_is_layered_from_wherever_it_lives(self, tmp_path: Path) -> None:917 elsewhere = write_defaults_recipe(tmp_path, filename="shared_defaults.py")918 919 class SiteConfig(BaseConfiguration):920 default_config = str(elsewhere)921 922 parents = DefaultConfig(tmp_path).parents_for(SiteConfig)923 assert parents == [BaseConfiguration, elsewhere]924 assert ConfigurationHandler(SiteConfig, parents=parents)("server.port") == 9999925 926 def test_an_explicit_path_that_does_not_exist_is_a_config_error(self, tmp_path: Path) -> None:927 missing = tmp_path / "absent.py"928 929 class SiteConfig(BaseConfiguration):930 default_config = missing931 932 with pytest.raises(ConfigError, match="does not exist"):933 DefaultConfig(tmp_path).parents_for(SiteConfig)934 935 def test_a_config_py_source_declares_its_own_default_config(self, tmp_path: Path) -> None:936 """The attribute is read off the recipe class a path source defines."""937 write_defaults_recipe(tmp_path)938 site = tmp_path / "site.py"939 site.write_text(940 "from genro_asgi.config import BaseConfiguration\n"941 "\n"942 "\n"943 "class SiteConfiguration(BaseConfiguration):\n"944 " default_config = False\n",945 encoding="utf-8",946 )947 assert DefaultConfig(tmp_path).parents_for(site) == [BaseConfiguration]948 949 def test_a_config_py_source_must_define_exactly_one_recipe(self, tmp_path: Path) -> None:950 site = tmp_path / "site.py"951 site.write_text("value = 1\n", encoding="utf-8")952 with pytest.raises(ConfigError, match="exactly one ConfigBuilder subclass"):953 DefaultConfig(tmp_path).parents_for(site)954 955 956 class TestHomeResolution:957 """``base_dir``: the explicit argument, then ``GENRO_ASGI_HOME``, then ``~``."""958 959 def test_the_env_var_is_the_default_base_dir(960 self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch961 ) -> None:962 monkeypatch.setenv(HOME_ENV, str(tmp_path))963 assert DefaultConfig().base_dir == tmp_path964 assert DefaultConfig().path == tmp_path / "config.py"965 966 def test_the_explicit_argument_wins_over_the_env_var(967 self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch968 ) -> None:969 monkeypatch.setenv(HOME_ENV, str(tmp_path / "from_env"))970 assert DefaultConfig(tmp_path / "explicit").base_dir == tmp_path / "explicit"971 972 def test_the_home_directory_is_the_last_resort(self, monkeypatch: pytest.MonkeyPatch) -> None:973 monkeypatch.delenv(HOME_ENV, raising=False)974 assert DefaultConfig().base_dir == Path.home() / ".genroasgi"975 976 def test_the_cli_registry_follows_the_same_variable(977 self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch978 ) -> None:979 monkeypatch.setenv(HOME_ENV, str(tmp_path))980 assert AppsRegistry().base_dir == tmp_path981 assert AppsRegistry().apps_dir == tmp_path / "apps"982 983 984 class TestServerLayersTheDeclaredDefaults:985 """The production wiring: ``AsgiServer(config=...)`` layers what the recipe declares."""986 987 def test_the_server_reads_the_conventional_defaults_recipe(988 self, genro_asgi_home: Path989 ) -> None:990 write_defaults_recipe(genro_asgi_home, mount_path=str(genro_asgi_home))991 992 class SiteConfig(AsgiConfigBuilder):993 def main(self, root: Any) -> None:994 root.configuration().server(host="127.0.0.1")995 996 server = AsgiServer(config=SiteConfig)997 assert server.config is not None998 assert server.config("server.host") == "127.0.0.1" # the site wins999 assert server.config("server.port") == 9999 # from the defaults layer1000 1001 def test_a_recipe_declining_the_layer_sees_only_the_package_defaults(1002 self, genro_asgi_home: Path1003 ) -> None:1004 write_defaults_recipe(genro_asgi_home)1005 1006 class SiteConfig(BaseConfiguration):1007 default_config = False1008 1009 server = AsgiServer(config=SiteConfig)1010 assert server.config is not None1011 with pytest.raises(KeyError, match="server.port"):1012 server.config("server.port")