tests/core/test_db.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 """DB handler tests (core 1b Phase 6): ``AsgiDbHandlerBase`` proxy, the server16 registry, and the ``databases`` section of a configured server.17 """18 19 from __future__ import annotations20 21 from typing import Any22 23 import pytest24 25 from genro_asgi import (26 AsgiConfigBuilder,27 AsgiDbHandlerBase,28 AsgiServer,29 BaseApplication,30 BaseServer,31 )32 33 34 class FakeDb:35 """Minimal db: holds params, exposes execute and closeConnection."""36 37 def __init__(self, **params: Any) -> None:38 self.params = params39 self.closed = False40 41 def execute(self, sql: str) -> str:42 return f"ran: {sql}"43 44 def closeConnection(self) -> None:45 self.closed = True46 47 48 class TestAsgiDbHandlerBase:49 def test_proxies_attribute(self) -> None:50 """A non-underscore attribute is fetched from the wrapped db."""51 db = FakeDb(dbname="shop")52 handler = AsgiDbHandlerBase(db)53 assert handler.params is db.params54 55 def test_proxies_method(self) -> None:56 """A method call is forwarded to the wrapped db."""57 handler = AsgiDbHandlerBase(FakeDb())58 assert handler.execute("SELECT 1") == "ran: SELECT 1"59 60 def test_close_connection_delegates(self) -> None:61 """closeConnection delegates to the wrapped db when present."""62 db = FakeDb()63 AsgiDbHandlerBase(db).closeConnection()64 assert db.closed is True65 66 def test_close_connection_noop_when_absent(self) -> None:67 """closeConnection is a no-op when the db has none."""68 69 class Bare:70 pass71 72 AsgiDbHandlerBase(Bare()).closeConnection() # must not raise73 74 def test_underscore_attributes_not_proxied(self) -> None:75 """Underscore names raise AttributeError instead of proxying (no recursion)."""76 handler = AsgiDbHandlerBase(FakeDb())77 with pytest.raises(AttributeError):78 handler._missing79 80 def test_missing_attribute_raises(self) -> None:81 """A public attribute the db lacks raises AttributeError."""82 handler = AsgiDbHandlerBase(FakeDb())83 with pytest.raises(AttributeError):84 handler.nonexistent85 86 def test_repr_shows_wrapped_type(self) -> None:87 """repr names the handler and the wrapped db type."""88 assert repr(AsgiDbHandlerBase(FakeDb())) == "AsgiDbHandlerBase(FakeDb)"89 90 def test_subclass_inherits_proxy(self) -> None:91 """A custom handler subclass keeps the proxy behaviour."""92 93 class CustomHandler(AsgiDbHandlerBase):94 pass95 96 handler = CustomHandler(FakeDb(dbname="x"))97 assert handler.params == {"dbname": "x"}98 assert isinstance(handler, AsgiDbHandlerBase)99 100 101 class TestDatabaseRegistry:102 def test_add_database_registers_by_code(self) -> None:103 server = BaseServer(applications=[BaseApplication(mount="")])104 handler = AsgiDbHandlerBase(FakeDb())105 server.add_database("shop", handler)106 assert server.databases == {"shop": handler}107 108 def test_add_database_duplicate_code_raises(self) -> None:109 server = BaseServer(applications=[BaseApplication(mount="")])110 server.add_database("shop", AsgiDbHandlerBase(FakeDb()))111 with pytest.raises(ValueError):112 server.add_database("shop", AsgiDbHandlerBase(FakeDb()))113 114 def test_databases_empty_by_default(self) -> None:115 assert BaseServer(applications=[BaseApplication(mount="")]).databases == {}116 117 118 # --- the databases section of a configured server ---119 120 121 class ShopApp(BaseApplication):122 pass123 124 125 class TwoDatabaseConfig(AsgiConfigBuilder):126 """A recipe with two ``database`` entries over ``FakeDb``."""127 128 def main(self, root: Any) -> None:129 cfg = root.configuration()130 cfg.server(host="127.0.0.1", port=8000)131 cfg.applications(default="shop").application(code="shop", app_class=ShopApp)132 self.databases_section(cfg)133 134 def databases_section(self, cfg: Any) -> None:135 """Two handlers: the default one and an explicit ``db_handler_class``."""136 dbs = cfg.databases()137 dbs.database(code="shop", db_class=FakeDb, dbname="shop")138 dbs.database(139 code="reports", db_class=FakeDb, db_handler_class=AsgiDbHandlerBase, dbname="ro"140 )141 142 143 class TestConfiguredDatabases:144 def test_configured_server_registers_both_handlers(self) -> None:145 server = AsgiServer(config=TwoDatabaseConfig)146 assert isinstance(server, AsgiServer)147 assert set(server.databases) == {"shop", "reports"}148 shop = server.databases["shop"]149 assert isinstance(shop, AsgiDbHandlerBase)150 assert shop.params == {"dbname": "shop"}151 assert shop.execute("SELECT 1") == "ran: SELECT 1"152 reports = server.databases["reports"]153 assert reports.params == {"dbname": "ro"}154 155 def test_database_missing_db_class_raises(self) -> None:156 class MissingDbClassConfig(AsgiConfigBuilder):157 def main(self, root: Any) -> None:158 cfg = root.configuration()159 cfg.server(host="127.0.0.1", port=8000)160 cfg.applications(default="shop").application(code="shop", app_class=ShopApp)161 cfg.databases().database(code="shop")162 163 with pytest.raises(ValueError):164 AsgiServer(config=MissingDbClassConfig)