tests/core/test_plugins.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 """Plugin system tests (Macro 4 Phase 5).16 17 Covers the ``PluginMixin`` server capability (default registry, registry18 extension, ``arm_router``), the config-driven arming through19 ``AsgiServer(config=...)``, the no-import-side-effect guarantee (checked in a20 fresh subprocess so it is order-independent) and the ported OpenAPI translator +21 ``router_openapi``.22 """23 24 from __future__ import annotations25 26 import inspect27 import subprocess28 import sys29 from typing import Any30 31 import pytest32 from genro_routes import Router, RoutingClass, route33 from genro_routes.plugins._base_plugin import BasePlugin34 35 from genro_asgi import AsgiServer, OpenAPIPlugin, OpenAPITranslator, RoutedApplication36 from genro_asgi.config import AsgiConfigBuilder37 from genro_asgi.plugin_mixin import PluginMixin, default_plugin_registry38 from genro_asgi.plugins import router_openapi39 40 41 class _MiniBase:42 """Terminal cooperative base: rejects any leftover kwarg."""43 44 def __init__(self, **kwargs: Any) -> None:45 if kwargs:46 raise TypeError(f"unexpected kwargs: {sorted(kwargs)}")47 48 49 class _MiniServer(PluginMixin, _MiniBase):50 """A bare PluginMixin composition, no other capability."""51 52 53 class _CustomPlugin(BasePlugin):54 """A distinct plugin whose code matches its registry key ("customtest")."""55 56 plugin_code = "customtest"57 plugin_description = "test-only plugin"58 59 def configure(self, enabled: bool = True, flag: str | None = None) -> None: # type: ignore[override]60 """No-op configure (storage handled by the wrapper)."""61 62 63 class _Svc(RoutingClass):64 """A tiny external router used as an arming target."""65 66 @route()67 def ping(self) -> dict:68 """Ping."""69 return {"ok": True}70 71 72 class ApiApp(RoutedApplication):73 """Routed app with typed handlers to exercise the schema path."""74 75 @route()76 def add(self, x: int, y: int = 0) -> dict:77 """Add two numbers."""78 return {"sum": x + y}79 80 @route(openapi_method="delete")81 def remove(self, item_id: int) -> dict:82 """Remove an item."""83 return {"deleted": item_id}84 85 @route()86 def make(self, items: list[str]) -> dict:87 """Make from a list."""88 return {"n": len(items)}89 90 91 class TestDefaultRegistry:92 def test_default_registry_holds_openapi(self) -> None:93 assert default_plugin_registry() == {"openapi": OpenAPIPlugin}94 95 def test_default_registry_is_fresh_each_call(self) -> None:96 first = default_plugin_registry()97 first["injected"] = OpenAPIPlugin98 assert "injected" not in default_plugin_registry()99 100 def test_registry_extension_merges_over_default(self) -> None:101 server = _MiniServer(102 plugins={"customtest": True}, plugin_registry={"customtest": _CustomPlugin}103 )104 assert server.plugin_registry["customtest"] is _CustomPlugin105 assert server.plugin_registry["openapi"] is OpenAPIPlugin106 107 108 class TestArmRouter:109 def test_arms_enabled_plugin_on_router(self) -> None:110 server = _MiniServer(plugins={"openapi": True})111 svc = _Svc()112 server.arm_router(svc.route)113 assert "openapi" in {plugin.name for plugin in svc.route.iter_plugins()}114 115 def test_dict_value_tunes_the_fixed_plugin_with_options(self) -> None:116 server = _MiniServer(plugins={"openapi": {"security_scheme": "ApiKey"}})117 # openapi is fixed; the dict value tunes it. pydantic stays in the base.118 assert server.plugins == {"openapi": {"security_scheme": "ApiKey"}, "pydantic": {}}119 svc = _Svc()120 server.arm_router(svc.route) # options reach router.plug without error121 assert "openapi" in {plugin.name for plugin in svc.route.iter_plugins()}122 123 def test_bundled_plugin_plugged_by_name(self) -> None:124 server = _MiniServer(plugins={"pydantic": True})125 svc = _Svc()126 server.arm_router(svc.route)127 assert "pydantic" in {plugin.name for plugin in svc.route.iter_plugins()}128 129 def test_registry_extension_arms_custom_plugin(self) -> None:130 server = _MiniServer(131 plugins={"customtest": True}, plugin_registry={"customtest": _CustomPlugin}132 )133 svc = _Svc()134 server.arm_router(svc.route)135 assert "customtest" in {plugin.name for plugin in svc.route.iter_plugins()}136 137 def test_disabling_a_fixed_plugin_is_a_config_error(self) -> None:138 with pytest.raises(ValueError, match="fixed structure"):139 _MiniServer(plugins={"openapi": False})140 141 def test_an_extra_name_is_retained_beside_the_fixed_base(self) -> None:142 # Retention is resolution-time; whether the name can be armed is decided143 # later by arm_router (see test_unknown_plugin_name_raises).144 server = _MiniServer(plugins={"logging": True})145 assert set(server.plugins) == {"pydantic", "openapi", "logging"}146 147 def test_arming_registers_the_plugin_in_the_router_registry(self) -> None:148 # The mirror of the subprocess test below: importing does not register,149 # arming does.150 server = _MiniServer(plugins={"openapi": True})151 server.arm_router(_Svc().route)152 assert "openapi" in Router.available_plugins()153 154 def test_false_leaves_an_extra_plugin_unarmed(self) -> None:155 server = _MiniServer(156 plugins={"customtest": False}, plugin_registry={"customtest": _CustomPlugin}157 )158 # the extra is dropped; only the fixed base remains.159 assert set(server.plugins) == {"pydantic", "openapi"}160 svc = _Svc()161 server.arm_router(svc.route)162 assert "customtest" not in {plugin.name for plugin in svc.route.iter_plugins()}163 164 def test_arming_twice_is_a_no_op(self) -> None:165 server = _MiniServer(plugins={"openapi": True, "pydantic": True})166 svc = _Svc()167 server.arm_router(svc.route)168 server.arm_router(svc.route) # must not raise on the already-plugged names169 names = [plugin.name for plugin in svc.route.iter_plugins()]170 assert names.count("openapi") == 1171 172 def test_unknown_plugin_name_raises(self) -> None:173 server = _MiniServer(plugins={"does-not-exist": True})174 svc = _Svc()175 with pytest.raises(ValueError):176 server.arm_router(svc.route)177 178 179 class PluginsConfig(AsgiConfigBuilder):180 """Two plugins armed, one explicitly disabled."""181 182 def main(self, root: Any) -> None:183 cfg = root.configuration()184 cfg.server(host="127.0.0.1", port=8000)185 cfg.applications(default="api").application(code="api", mount="", app_class=ApiApp)186 self.plugins_section(cfg)187 188 def plugins_section(self, cfg: Any) -> None:189 """``enabled=False`` leaves a plugin unarmed."""190 plugins = cfg.plugins()191 plugins.plugin(code="openapi")192 plugins.plugin(code="pydantic")193 plugins.plugin(code="logging", enabled=False)194 195 196 class TestConfigDriven:197 def test_configured_server_carries_the_plugins_config(self) -> None:198 server = AsgiServer(config=PluginsConfig)199 assert server.plugins == {"openapi": {}, "pydantic": {}}200 201 def test_configured_server_arms_the_routed_app(self) -> None:202 server = AsgiServer(config=PluginsConfig)203 names = {plugin.name for plugin in server.root_application.route.iter_plugins()}204 assert {"auth", "openapi", "pydantic"} <= names205 206 def test_plugin_options_map_to_a_dict(self) -> None:207 class Recipe(AsgiConfigBuilder):208 def main(self, root: Any) -> None:209 cfg = root.configuration()210 cfg.applications(default="api").application(211 code="api", mount="", app_class=ApiApp212 )213 cfg.plugins().plugin(code="openapi", security_scheme="ApiKey")214 215 server = AsgiServer(config=Recipe)216 # openapi is fixed; the config tunes it. pydantic stays in the base.217 assert server.plugins == {"openapi": {"security_scheme": "ApiKey"}, "pydantic": {}}218 219 220 class TestNoImportSideEffect:221 def test_importing_the_package_does_not_register_openapi(self) -> None:222 code = (223 "import genro_asgi\n"224 "import genro_asgi.plugins.openapi\n"225 "from genro_routes import Router\n"226 "assert 'openapi' not in Router.available_plugins(), Router.available_plugins()\n"227 )228 result = subprocess.run([sys.executable, "-c", code], capture_output=True, text=True)229 assert result.returncode == 0, result.stderr230 231 232 class TestLazyArming:233 def test_unmounted_app_arms_nothing_extra(self) -> None:234 app = ApiApp() # no server → only the auth plug from __init__235 assert {plugin.name for plugin in app.route.iter_plugins()} == {"auth"}236 237 def test_mounted_app_arms_on_first_route_access(self) -> None:238 app = ApiApp(mount="")239 AsgiServer(applications=[app], plugins={"openapi": True})240 names = {plugin.name for plugin in app.route.iter_plugins()}241 assert {"auth", "openapi"} <= names242 243 def test_no_plugins_config_still_arms_the_fixed_base(self) -> None:244 # A mixin-equipped server always arms the fixed base (pydantic/openapi)245 # on top of the app's own ``auth`` plug, even with no plugins config.246 app = ApiApp(mount="")247 AsgiServer(applications=[app])248 assert {plugin.name for plugin in app.route.iter_plugins()} == {249 "auth",250 "pydantic",251 "openapi",252 }253 254 255 class TestTranslator:256 def test_translator_module_imports_no_pydantic(self) -> None:257 # The redesigned translator reads genro-routes' cached neutral blocks;258 # it must not import pydantic nor inspect callables (ratified ruling).259 from genro_asgi.plugins.openapi import translator as translator_module260 261 source = inspect.getsource(translator_module)262 assert "import pydantic" not in source263 assert "from pydantic" not in source264 assert "create_pydantic_model_for_func" not in source265 266 def test_guess_get_for_scalar_fields(self) -> None:267 fields = [268 {"name": "x", "schema": {"type": "integer"}, "required": True, "kind": "pk"},269 {"name": "y", "schema": {"type": "string"}, "required": False, "kind": "pk"},270 ]271 assert OpenAPITranslator.guess_http_method(fields) == "get"272 273 def test_guess_post_for_non_scalar_field(self) -> None:274 fields = [{"name": "items", "schema": {"type": "array"}, "required": True, "kind": "pk"}]275 assert OpenAPITranslator.guess_http_method(fields) == "post"276 277 def test_guess_get_for_optional_scalar_union(self) -> None:278 fields = [279 {280 "name": "n",281 "schema": {"anyOf": [{"type": "integer"}, {"type": "null"}]},282 "required": False,283 "kind": "pk",284 }285 ]286 assert OpenAPITranslator.guess_http_method(fields) == "get"287 288 def test_guess_ignores_untyped_and_var_fields(self) -> None:289 fields = [290 {"name": "a", "schema": None, "required": True, "kind": "positional_or_keyword"},291 {"name": "kwargs", "schema": None, "required": False, "kind": "var_keyword"},292 ]293 assert OpenAPITranslator.guess_http_method(fields) == "get"294 295 def test_ref_schema_is_non_scalar(self) -> None:296 fields = [{"name": "body", "schema": {"$ref": "#/$defs/Item"}, "required": True, "kind": "pk"}]297 assert OpenAPITranslator.guess_http_method(fields) == "post"298 299 def test_schema_to_parameters_marks_required(self) -> None:300 schema = {301 "properties": {"x": {"type": "integer"}, "y": {"type": "integer"}},302 "required": ["x"],303 }304 params = OpenAPITranslator.schema_to_parameters(schema)305 by_name = {p["name"]: p for p in params}306 assert by_name["x"]["required"] is True307 assert by_name["y"]["required"] is False308 assert by_name["x"]["in"] == "query"309 310 311 class TestRouterOpenapi:312 def _armed_service(self) -> ApiApp:313 server = _MiniServer(plugins={"openapi": True, "pydantic": True})314 svc = ApiApp()315 server.arm_router(svc.route)316 return svc317 318 def test_flat_paths_carry_operations_and_schemas(self) -> None:319 spec = router_openapi(self._armed_service().route)320 assert set(spec["paths"]) == {"/add", "/remove", "/make"}321 add_get = spec["paths"]["/add"]["get"]322 assert add_get["operationId"] == "add"323 # Input query params come from the cached request_schema: x required, y not.324 by_name = {p["name"]: p for p in add_get["parameters"]}325 assert by_name["x"]["required"] is True326 assert by_name["y"]["required"] is False327 assert "responses" in add_get328 329 def test_method_override_from_handler_config(self) -> None:330 spec = router_openapi(self._armed_service().route)331 assert "delete" in spec["paths"]["/remove"]332 333 def test_complex_param_becomes_post_request_body(self) -> None:334 spec = router_openapi(self._armed_service().route)335 make_post = spec["paths"]["/make"]["post"]336 assert "requestBody" in make_post337 338 def test_hierarchical_format_preserves_the_tree(self) -> None:339 server = _MiniServer(plugins={"openapi": True, "pydantic": True})340 parent = ApiApp()341 parent.route.add_branches({"name": "sub", "instance": _Svc()})342 server.arm_router(parent.route)343 flat = router_openapi(parent.route)344 assert "/sub/ping" in flat["paths"]345 hierarchical = router_openapi(parent.route, hierarchical=True)346 assert "/add" in hierarchical["paths"]347 assert "sub" in hierarchical["routers"]348 assert "/ping" in hierarchical["routers"]["sub"]["paths"]349 350 def test_empty_router_yields_empty_paths(self) -> None:351 class Empty(RoutingClass):352 pass353 354 assert router_openapi(Empty().route) == {"paths": {}}