tests/core/test_storage_mixin.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 """StorageMixin + ``storage`` config section tests.16 17 Two layers: the capability mixin (``storage=``/``storage_key=`` peeled by18 ``AsgiServer``, the plain ``BaseServer`` never gaining a ``storage`` attribute)19 and the config path (a recipe's ``storage`` section reaching the server's20 ``StorageManager`` through ``AsgiServer(config=...)``, an explicit kwarg still21 winning over the configured one).22 23 ``storage=`` takes exactly three shapes, one test class each: ``None`` (the24 single ``site:`` mount on the deployment directory), a ``StorageManager``25 (adopted as-is) and a ``list[dict]`` (genro-storage's own mount configuration,26 passed through to ``configure()``). Encryption is declared per WRITE, so the27 key test asserts both halves at once: a credential write lands as an envelope28 on disk while a session file in the same tree stays plain, and both read back.29 """30 31 from __future__ import annotations32 33 from pathlib import Path34 from typing import Any35 36 import pytest37 from cryptography.fernet import Fernet38 39 from genro_storage import StorageManager40 from genro_storage.exceptions import (41 StorageConfigError,42 StorageError,43 StorageNotFoundError,44 )45 46 from tests.storage_support import site_storage47 48 from genro_asgi import (49 AsgiConfigBuilder,50 AsgiServer,51 BaseApplication,52 BaseServer,53 ConfigurationHandler,54 StorageMixin,55 )56 from genro_asgi.middleware.base import BaseMiddleware57 from genro_asgi.types import Receive, Scope, Send58 59 60 class ShopApp(BaseApplication):61 """Minimal primary app for the config-driven tests."""62 63 async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:64 await send({"type": "http.response.start", "status": 200, "headers": []})65 await send({"type": "http.response.body", "body": b"shop"})66 67 68 @pytest.fixture69 def key() -> str:70 """A single fresh Fernet key."""71 return Fernet.generate_key().decode()72 73 74 def chain_types(server: AsgiServer) -> list[str]:75 """The class names of the middlewares in the server's chain, outermost first."""76 names: list[str] = []77 node: object = server.middleware_chain78 while isinstance(node, BaseMiddleware):79 names.append(type(node).__name__)80 node = node.app81 return names82 83 84 class TestMixinComposition:85 """The three shapes ``storage=`` accepts, plus the composition without the mixin."""86 87 def test_default_storage_is_the_site_mount_on_the_deployment_directory(self) -> None:88 server = AsgiServer(applications=[BaseApplication(mount="")])89 assert isinstance(server.storage, StorageManager)90 assert server.storage.get_mount_names() == ["site"]91 assert server.storage.node("site:pyproject.toml").exists() # anchored to the cwd92 93 def test_manager_instance_is_adopted_as_is(self, tmp_path: Path) -> None:94 provided = site_storage(tmp_path)95 server = AsgiServer(applications=[BaseApplication(mount="")], storage=provided)96 assert server.storage is provided97 98 def test_mount_list_is_passed_through_to_configure(self, tmp_path: Path) -> None:99 (tmp_path / "data").mkdir() # genro-storage mounts an existing directory100 server = AsgiServer(101 applications=[BaseApplication(mount="")],102 storage=[{"name": "data", "protocol": "local", "base_path": str(tmp_path / "data")}],103 )104 node = server.storage.node("data:hello.txt")105 node.write_text("hi")106 assert node.read_text() == "hi"107 108 def test_base_server_has_no_storage_attribute(self) -> None:109 assert not hasattr(BaseServer(applications=[BaseApplication(mount="")]), "storage")110 111 112 class TestStorageKey:113 def test_encrypted_and_plain_writes_share_one_tree(self, tmp_path: Path, key: str) -> None:114 """A credential is an envelope on disk, a session next to it stays plain."""115 server = AsgiServer(116 applications=[BaseApplication(mount="")],117 storage=site_storage(tmp_path),118 storage_key=key,119 )120 assert server.storage.encryption_active121 credential = server.storage.node("site:users/admin.json")122 credential.write_text('{"k": 1}', encrypted=True)123 session = server.storage.node("site:sessions/abc.json")124 session.write_text('{"s": 2}')125 126 assert (tmp_path / "users" / "admin.json").read_bytes().startswith(b"#GNRE1:")127 assert (tmp_path / "sessions" / "abc.json").read_bytes() == b'{"s": 2}'128 assert credential.read_text() == '{"k": 1}'129 assert session.read_text() == '{"s": 2}'130 131 def test_encrypted_write_without_key_material_raises(self, tmp_path: Path) -> None:132 """Dormancy is loud: ``encrypted=True`` with no key installed fails at the write."""133 server = AsgiServer(134 applications=[BaseApplication(mount="")],135 storage=site_storage(tmp_path),136 )137 with pytest.raises(StorageError):138 server.storage.node("site:users/admin.json").write_text("{}", encrypted=True)139 140 def test_empty_storage_key_raises_at_construction(self, tmp_path: Path) -> None:141 with pytest.raises(StorageConfigError):142 AsgiServer(143 applications=[BaseApplication(mount="")],144 storage=site_storage(tmp_path),145 storage_key="",146 )147 148 def test_omitted_storage_key_leaves_encryption_dormant(self, tmp_path: Path) -> None:149 server = AsgiServer(150 applications=[BaseApplication(mount="")],151 storage=site_storage(tmp_path),152 )153 assert not server.storage.encryption_active154 155 156 class TestBareMixin:157 def test_mixin_over_base_server_exposes_storage(self) -> None:158 class Srv(StorageMixin, BaseServer):159 pass160 161 server = Srv(applications=[BaseApplication(mount="")])162 assert isinstance(server.storage, StorageManager)163 164 165 class TestConfigDriven:166 def test_recipe_storage_section_serves_declared_mounts(self, tmp_path: Path) -> None:167 data_dir = tmp_path / "data"168 data_dir.mkdir()169 170 class StorageConfig(AsgiConfigBuilder):171 def main(self, root: Any) -> None:172 cfg = root.configuration()173 cfg.server(host="127.0.0.1", port=8000)174 cfg.storage(app=StorageManager).local(name="data", base_path=str(data_dir))175 cfg.applications(default="shop").application(code="shop", app_class=ShopApp)176 177 server = AsgiServer(config=StorageConfig)178 node = server.storage.node("data:file.txt")179 node.write_text("x")180 assert node.read_text() == "x"181 182 def test_recipe_storage_key_installs_encryption(self, tmp_path: Path, key: str) -> None:183 secure_root = tmp_path184 185 class StorageConfig(AsgiConfigBuilder):186 def main(self, root: Any) -> None:187 cfg = root.configuration()188 cfg.server(host="127.0.0.1", port=8000)189 cfg.storage(app=StorageManager, storage_key=key).local(190 name="site", base_path=str(secure_root)191 )192 cfg.applications(default="shop").application(code="shop", app_class=ShopApp)193 194 server = AsgiServer(config=StorageConfig)195 assert server.storage.encryption_active196 197 def test_explicit_kwarg_wins_over_the_configured_one(self, tmp_path: Path) -> None:198 data_dir = tmp_path / "data"199 data_dir.mkdir()200 override_dir = tmp_path / "override"201 override_dir.mkdir()202 203 class StorageConfig(AsgiConfigBuilder):204 def main(self, root: Any) -> None:205 cfg = root.configuration()206 cfg.server(host="127.0.0.1", port=8000)207 cfg.storage(app=StorageManager).local(name="data", base_path=str(data_dir))208 cfg.applications(default="shop").application(code="shop", app_class=ShopApp)209 210 server = AsgiServer(211 config=StorageConfig,212 storage=[{"name": "only", "protocol": "local", "base_path": str(override_dir)}],213 )214 assert server.storage.node("only:w.txt") is not None215 with pytest.raises(StorageNotFoundError, match="data"):216 server.storage.node("data:w.txt")217 218 219 class TestDefaultLayoutFallback:220 """An empty mount list means "the default layout", not "no storage"."""221 222 def test_an_empty_mount_list_falls_back_to_the_site_mount(self) -> None:223 server = AsgiServer(applications=[BaseApplication(mount="")], storage=[])224 assert server.storage.get_mount_names() == ["site"]225 226 def test_a_key_only_recipe_section_still_serves_the_site_mount(self, key: str) -> None:227 """The recipe declares the key and no mount; the default layout applies."""228 229 class KeyOnlyConfig(AsgiConfigBuilder):230 def main(self, root: Any) -> None:231 cfg = root.configuration()232 cfg.storage(app=StorageManager, storage_key=key)233 cfg.applications(default="shop").application(code="shop", app_class=ShopApp)234 235 server = AsgiServer(config=ConfigurationHandler(KeyOnlyConfig))236 assert server.storage.get_mount_names() == ["site"]237 assert server.storage.encryption_active