tests/core/test_config_env.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 """The read door reaching the environment and the applications.16 17 Two properties of the resolver model, one file. **Resolvers resolve in place**:18 a recipe stores an ``EnvResolver`` where a value would go and the read stack19 resolves it at READ time, so the value the runtime consumes is the environment's20 — the layer the old ``^pointer`` model silently dropped for ``storage_key``,21 which is why the encryption round-trip below is a regression test and not a22 nicety. **Applications read their own subtree**: ``app.config(path)`` prefixes23 ``applications.<code>.`` and delegates to the server's handler, so an app holds24 an address in the tree, never a slice of it.25 """26 27 from __future__ import annotations28 29 from pathlib import Path30 from typing import Any31 32 import pytest33 from genro_storage import StorageManager34 from genro_storage.exceptions import StorageConfigError35 from cryptography.fernet import Fernet36 from genro_bag.resolvers import EnvResolver37 from genro_builders.builder import element38 39 from genro_asgi import (40 AsgiConfigBuilder,41 AsgiServer,42 BaseApplication,43 ConfigError,44 ConfigurationHandler,45 )46 from genro_asgi.application import ApplicationGrammar47 from genro_asgi.types import Receive, Scope, Send48 49 HOST_ENV_VAR = "GENRO_TEST_HOST"50 PORT_ENV_VAR = "GENRO_TEST_PORT"51 STORAGE_KEY_ENV_VAR = "GENRO_TEST_STORAGE_KEY"52 ADMIN_PW_ENV_VAR = "GENRO_TEST_ENV_ADMIN_PW"53 BASIC_PW_ENV_VAR = "GENRO_TEST_BASIC_PW"54 DB_PW_ENV_VAR = "GENRO_TEST_DB_PW"55 56 57 class ShopGrammar(ApplicationGrammar):58 """A richer app grammar: the ``catalog`` block on top of ``parameters``."""59 60 @element(node_label="catalog")61 def catalog(self, title: str | None = None, page_size: int = 20) -> None:62 """The catalog block, read back as ``applications.<code>.catalog.<attr>``."""63 64 65 class ShopApp(BaseApplication):66 """App with its own grammar, answering ``shop``."""67 68 grammar = ShopGrammar69 70 async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:71 await send({"type": "http.response.start", "status": 200, "headers": []})72 await send({"type": "http.response.body", "body": b"shop"})73 74 75 class PlainApp(BaseApplication):76 """App with the inherited minimal grammar."""77 78 async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:79 await send({"type": "http.response.start", "status": 200, "headers": []})80 await send({"type": "http.response.body", "body": b"plain"})81 82 83 class AppReadsConfig(AsgiConfigBuilder):84 """One app whose mounted subtree carries both a ``parameters`` and a ``catalog``."""85 86 def main(self, root: Any) -> None:87 cfg = root.configuration()88 cfg.server(host="127.0.0.1", port=8000)89 self.applications_section(cfg)90 91 def applications_section(self, cfg: Any) -> None:92 """``shop`` claims the root and declares its own vocabulary."""93 apps = cfg.applications(default="shop")94 app = apps.application(code="shop", mount="", app_class=ShopApp)95 app.parameters(currency="EUR")96 app.catalog(title="Outlet")97 98 99 @pytest.fixture100 def key() -> str:101 """A single fresh Fernet key."""102 return Fernet.generate_key().decode()103 104 105 class TestServerAttributesFromTheEnvironment:106 """``host``/``port`` supplied by resolvers sitting in the attributes."""107 108 def test_host_and_typed_port_resolve_at_read_time(109 self, monkeypatch: pytest.MonkeyPatch110 ) -> None:111 monkeypatch.setenv(HOST_ENV_VAR, "10.0.0.7")112 monkeypatch.setenv(PORT_ENV_VAR, "9443")113 114 class EnvServerConfig(AsgiConfigBuilder):115 def main(self, root: Any) -> None:116 cfg = root.configuration()117 cfg.server(118 host=EnvResolver(HOST_ENV_VAR),119 port=EnvResolver(PORT_ENV_VAR, dtype="L"),120 )121 cfg.applications().application(code="shop", mount="", app_class=ShopApp)122 123 server = AsgiServer(config=EnvServerConfig)124 assert server.config_host == "10.0.0.7"125 assert server.config_port == 9443 # dtype="L" → a real int126 127 def test_the_environment_is_read_not_frozen_into_the_recipe(128 self, monkeypatch: pytest.MonkeyPatch129 ) -> None:130 monkeypatch.setenv(HOST_ENV_VAR, "first.example.com")131 132 class EnvHostConfig(AsgiConfigBuilder):133 def main(self, root: Any) -> None:134 cfg = root.configuration()135 cfg.server(host=EnvResolver(HOST_ENV_VAR), port=8000)136 cfg.applications().application(code="shop", mount="", app_class=ShopApp)137 138 handler = ConfigurationHandler(EnvHostConfig)139 assert handler("server.host") == "first.example.com"140 monkeypatch.setenv(HOST_ENV_VAR, "second.example.com")141 assert handler("server.host") == "second.example.com"142 143 144 class TestStorageKeyFromTheEnvironment:145 """The regression the pointer model caused: encryption silently disarmed."""146 147 def test_resolved_storage_key_arms_encryption_end_to_end(148 self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch, key: str149 ) -> None:150 monkeypatch.setenv(STORAGE_KEY_ENV_VAR, key)151 vault_dir = tmp_path / "vault"152 vault_dir.mkdir()153 154 class VaultConfig(AsgiConfigBuilder):155 def setup(self, data: Any) -> None:156 """The mount path travels through the datastore, not a closure."""157 data["vault_path"] = str(vault_dir)158 159 def main(self, root: Any) -> None:160 cfg = root.configuration()161 cfg.server(host="127.0.0.1", port=8000)162 cfg.storage(163 app=StorageManager,164 storage_key=EnvResolver(STORAGE_KEY_ENV_VAR),165 ).local(166 name="vault", base_path=self.data["vault_path"], default_encrypted=True167 )168 cfg.applications().application(code="shop", mount="", app_class=ShopApp)169 170 server = AsgiServer(config=VaultConfig)171 assert server.storage.encryption_active172 node = server.storage.node("vault:secret.txt")173 node.write_text("top-secret")174 assert (vault_dir / "secret.txt").read_bytes() != b"top-secret"175 assert node.read_text() == "top-secret"176 177 def test_a_storage_key_resolving_empty_is_a_boot_error(178 self, monkeypatch: pytest.MonkeyPatch179 ) -> None:180 monkeypatch.delenv(STORAGE_KEY_ENV_VAR, raising=False)181 182 class EmptyKeyConfig(AsgiConfigBuilder):183 def main(self, root: Any) -> None:184 cfg = root.configuration()185 cfg.server(host="127.0.0.1", port=8000)186 cfg.storage(187 app=StorageManager,188 storage_key=EnvResolver(STORAGE_KEY_ENV_VAR, default=""),189 ).memory(name="scratch")190 cfg.applications().application(code="shop", mount="", app_class=ShopApp)191 192 # An empty string is NOT "missing": the recipe promised an encryption193 # key, so a resolution that yields nothing is a boot error, never a194 # silent downgrade to plaintext.195 with pytest.raises(StorageConfigError, match="empty key material"):196 AsgiServer(config=EmptyKeyConfig)197 198 199 class TestSecretsFromTheEnvironment:200 """``admin_password`` as a node value, ``basic_user.password`` as an attribute."""201 202 def test_admin_password_node_value_resolves(203 self, monkeypatch: pytest.MonkeyPatch204 ) -> None:205 monkeypatch.setenv(ADMIN_PW_ENV_VAR, "boot-s3cret")206 207 class AdminConfig(AsgiConfigBuilder):208 def main(self, root: Any) -> None:209 cfg = root.configuration()210 cfg.authentication().admin_password(EnvResolver(ADMIN_PW_ENV_VAR))211 cfg.applications().application(code="shop", mount="", app_class=ShopApp)212 213 handler = ConfigurationHandler(AdminConfig)214 assert handler.identity_kwargs()["admin_password"] == "boot-s3cret"215 216 def test_basic_user_password_attribute_resolves_into_the_auth_entries(217 self, monkeypatch: pytest.MonkeyPatch218 ) -> None:219 monkeypatch.setenv(BASIC_PW_ENV_VAR, "attr-s3cret")220 221 class BasicConfig(AsgiConfigBuilder):222 def main(self, root: Any) -> None:223 cfg = root.configuration()224 creds = cfg.authentication().credentials()225 creds.basic_user(226 username="admin",227 password=EnvResolver(BASIC_PW_ENV_VAR),228 tags="admin",229 )230 cfg.applications().application(code="shop", mount="", app_class=ShopApp)231 232 entries = ConfigurationHandler(BasicConfig).auth_entries()233 assert entries is not None234 assert entries["basic"]["admin"]["password"] == "attr-s3cret"235 236 def test_admin_password_resolving_non_string_is_a_boot_error(237 self, monkeypatch: pytest.MonkeyPatch238 ) -> None:239 # dtype="L" makes the resolver deliver an int: the recipe line is240 # legal (it IS a resolver), so the type check belongs to the fold.241 monkeypatch.setenv(ADMIN_PW_ENV_VAR, "12345")242 243 class TypedConfig(AsgiConfigBuilder):244 def main(self, root: Any) -> None:245 cfg = root.configuration()246 cfg.authentication().admin_password(247 EnvResolver(ADMIN_PW_ENV_VAR, dtype="L")248 )249 cfg.applications().application(code="shop", mount="", app_class=ShopApp)250 251 with pytest.raises(ConfigError, match="must resolve to a string"):252 ConfigurationHandler(TypedConfig).identity_kwargs()253 254 255 class RecordingDb:256 """A ``db_class`` stand-in: the fold hands it the connection params."""257 258 def __init__(self, **params: Any) -> None:259 self.params = params260 261 262 class TestOpenAttributesFromTheEnvironment:263 """A resolver in an OPEN element's ``**kwargs`` resolves through the bulk read."""264 265 def test_database_password_resolves_in_the_open_kwargs(266 self, monkeypatch: pytest.MonkeyPatch267 ) -> None:268 monkeypatch.setenv(DB_PW_ENV_VAR, "db-s3cret")269 270 class DbConfig(AsgiConfigBuilder):271 def main(self, root: Any) -> None:272 cfg = root.configuration()273 cfg.databases().database(274 code="main",275 db_class=RecordingDb,276 dsn="postgres://localhost/main",277 password=EnvResolver(DB_PW_ENV_VAR),278 )279 cfg.applications().application(code="shop", mount="", app_class=ShopApp)280 281 [descriptor] = ConfigurationHandler(DbConfig).databases()282 params = descriptor["params"]283 assert params["password"] == "db-s3cret"284 assert params["dsn"] == "postgres://localhost/main"285 286 287 class TestApplicationSideReads:288 """``app.config(path)`` addresses ``applications.<code>.<path>``."""289 290 def test_written_value_in_the_mounted_subtree(self) -> None:291 server = AsgiServer(config=AppReadsConfig)292 shop = server.applications["shop"]293 assert shop.config("parameters.currency") == "EUR"294 assert shop.config("catalog.title") == "Outlet"295 296 def test_signature_default_of_the_mounted_grammar(self) -> None:297 server = AsgiServer(config=AppReadsConfig)298 # Never written by the recipe: the read walks up to ShopGrammar.catalog.299 assert server.applications["shop"].config("catalog.page_size") == 20300 301 def test_call_site_default_answers_an_unwritten_attribute(self) -> None:302 server = AsgiServer(config=AppReadsConfig)303 assert server.applications["shop"].config("parameters.locale", default="it") == "it"304 305 def test_a_missing_path_raises_the_noisy_key_error(self) -> None:306 server = AsgiServer(config=AppReadsConfig)307 with pytest.raises(KeyError, match="applications.shop.parameters.locale"):308 server.applications["shop"].config("parameters.locale")309 310 def test_each_app_reads_only_its_own_prefix(self) -> None:311 class TwoAppsConfig(AsgiConfigBuilder):312 def main(self, root: Any) -> None:313 cfg = root.configuration()314 apps = cfg.applications(default="shop")315 apps.application(code="shop", mount="", app_class=ShopApp).parameters(316 currency="EUR"317 )318 apps.application(code="plain", app_class=PlainApp).parameters(319 currency="USD"320 )321 322 server = AsgiServer(config=TwoAppsConfig)323 assert server.applications["shop"].config("parameters.currency") == "EUR"324 assert server.applications["plain"].config("parameters.currency") == "USD"325 326 327 class TestUnconfiguredServer:328 """An app on a bare server has nothing to read."""329 330 def test_call_site_default_answers(self) -> None:331 app = ShopApp(mount="")332 AsgiServer(applications=[app])333 assert app.config("catalog.title", default="none") == "none"334 335 def test_without_a_default_it_raises(self) -> None:336 app = ShopApp(mount="")337 AsgiServer(applications=[app])338 with pytest.raises(KeyError, match="not attached to a configured server"):339 app.config("catalog.title")340 341 def test_a_detached_app_raises_too(self) -> None:342 assert ShopApp().config("catalog.title", default="none") == "none"