tests/core/test_openapi_application.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 """OpenApiApplication tests (Macro 4 Phase 6).16 17 Requests drive a REAL ``AsgiServer`` composition at the ASGI level (no18 uvicorn). The pydantic + openapi plugins are armed through the server's19 ``plugins`` config (direct mode) or plugged on the mounted class by the app20 (mounted mode). The GET-side helpers come from ``tests/conftest.py``;21 ``json_request`` (local) adds a JSON body.22 """23 24 from __future__ import annotations25 26 import json27 from typing import Any, Callable28 29 import pytest30 from genro_routes import RoutingClass, route31 32 from genro_asgi import AsgiServer, OpenApiApplication, RoutedApplication33 from genro_asgi.types import Message, Scope34 35 36 class SubApi(RoutingClass):37 """External API mounted into an OpenApiApplication under ``api_name``."""38 39 openapi_info = {"title": "Sub API", "version": "3.0.0", "description": "mounted"}40 41 @route()42 def ping(self, name: str = "x") -> dict[str, str]:43 """Echo a name back."""44 return {"pong": name}45 46 @route()47 def make(self, payload: dict) -> dict[str, Any]:48 """Build something from a JSON body."""49 return {"made": payload}50 51 52 class DirectApi(OpenApiApplication):53 """Direct mode: typed @route methods on the app itself."""54 55 openapi_info = {"title": "Direct API", "version": "2.0.0", "description": "direct"}56 57 @route()58 def add(self, x: int = 0, y: int = 0) -> dict[str, int]:59 """Add two integers."""60 return {"sum": x + y}61 62 @route()63 def store(self, body_data: dict | None = None) -> dict[str, Any]:64 """Keep the whole JSON body."""65 return {"stored": body_data}66 67 68 class Empty(RoutedApplication):69 """A do-nothing primary so an OpenApiApplication can mount as a secondary."""70 71 72 def direct_server() -> AsgiServer:73 """A server whose plugin config arms pydantic + openapi on routed apps."""74 return AsgiServer(applications=[DirectApi(mount="")], plugins={"openapi": True, "pydantic": True})75 76 77 def mounted_server(app: OpenApiApplication) -> AsgiServer:78 """A server with ``app`` mounted at its ``mount`` over an empty root app."""79 server = AsgiServer(80 applications=[Empty(mount=""), app],81 plugins={"openapi": True, "pydantic": True},82 )83 return server84 85 86 @pytest.fixture87 def json_request() -> Callable[..., object]:88 """Fixture: drive one JSON-body request through a server at the ASGI level."""89 90 async def _json_request(91 server: object, path: str, body: bytes, method: str = "POST"92 ) -> list[Message]:93 scope: Scope = {94 "type": "http",95 "method": method,96 "path": path,97 "query_string": b"",98 "headers": [(b"content-type", b"application/json")],99 }100 sent: list[Message] = []101 102 async def receive() -> Message:103 return {"type": "http.request", "body": body, "more_body": False}104 105 async def send(message: Message) -> None:106 sent.append(message)107 108 await server(scope, receive, send) # type: ignore[operator]109 return sent110 111 return _json_request112 113 114 class TestDirectMode:115 async def test_schema_is_valid_openapi(116 self, http_request, response_status, response_body117 ) -> None:118 sent = await http_request(direct_server(), "/_meta/schema_json")119 assert response_status(sent) == 200120 doc = json.loads(response_body(sent))121 assert doc["openapi"] == "3.1.0"122 assert doc["info"] == {123 "title": "Direct API",124 "version": "2.0.0",125 "description": "direct",126 }127 assert "/add" in doc["paths"]128 129 async def test_operation_and_input_schema_present(self, http_request, response_body) -> None:130 sent = await http_request(direct_server(), "/_meta/schema_json")131 doc = json.loads(response_body(sent))132 operation = doc["paths"]["/add"]["get"]133 assert operation["operationId"] == "add"134 param_names = {p["name"] for p in operation["parameters"]}135 assert {"x", "y"} <= param_names136 137 async def test_endpoint_is_served(self, http_request, response_status, response_body) -> None:138 scope: Scope = {139 "type": "http",140 "method": "GET",141 "path": "/add",142 "query_string": b"x=1&y=2",143 "headers": [],144 }145 sent: list[Message] = []146 147 async def receive() -> Message:148 return {"type": "http.request"}149 150 async def send(message: Message) -> None:151 sent.append(message)152 153 await direct_server()(scope, receive, send)154 assert response_status(sent) == 200155 assert json.loads(response_body(sent)) == {"sum": 3}156 157 async def test_docs_served(158 self, http_request, response_status, response_headers, response_body159 ) -> None:160 sent = await http_request(direct_server(), "/_meta/docs")161 assert response_status(sent) == 200162 assert response_headers(sent)[b"content-type"] == b"text/html; charset=utf-8"163 body = response_body(sent).decode()164 assert "swagger-ui" in body165 assert '"/_meta/schema_json"' in body166 167 async def test_index_served(self, http_request, response_status, response_body) -> None:168 sent = await http_request(direct_server(), "/_meta/index")169 assert response_status(sent) == 200170 body = response_body(sent).decode()171 assert "Direct API" in body172 assert 'href="/_meta/docs"' in body173 174 175 class TestDocsOff:176 async def test_docs_off_is_404(self, http_request, response_status) -> None:177 server = AsgiServer(178 applications=[DirectApi(mount="", docs="off")], plugins={"openapi": True, "pydantic": True}179 )180 sent = await http_request(server, "/_meta/docs")181 assert response_status(sent) == 404182 183 184 class TestMountedMode:185 async def test_endpoint_under_mount_and_api(186 self, http_request, response_status, response_body187 ) -> None:188 app = OpenApiApplication(code="mount", routing_class=SubApi())189 scope: Scope = {190 "type": "http",191 "method": "GET",192 "path": "/mount/api/ping",193 "query_string": b"name=z",194 "headers": [],195 }196 sent: list[Message] = []197 198 async def receive() -> Message:199 return {"type": "http.request"}200 201 async def send(message: Message) -> None:202 sent.append(message)203 204 await mounted_server(app)(scope, receive, send)205 assert response_status(sent) == 200206 assert json.loads(response_body(sent)) == {"pong": "z"}207 208 async def test_schema_covers_mounted_endpoints(209 self, http_request, response_status, response_body210 ) -> None:211 app = OpenApiApplication(code="mount", routing_class=SubApi())212 sent = await http_request(mounted_server(app), "/mount/_meta/schema_json")213 assert response_status(sent) == 200214 doc = json.loads(response_body(sent))215 # The mounted class's own openapi_info wins over the app default.216 assert doc["info"]["title"] == "Sub API"217 assert "/api/ping" in doc["paths"]218 assert doc["paths"]["/api/ping"]["get"]["operationId"] == "ping"219 220 async def test_module_import_mode(self) -> None:221 app = OpenApiApplication(code="m", module=f"{SubApi.__module__}:SubApi")222 assert app.api_name == "api"223 node = app.route.node("/api/ping")224 assert node.error is None, node.error225 226 227 class TestBodyBinding:228 async def test_json_body_spread_over_scalar_params(229 self, json_request, response_status, response_body230 ) -> None:231 sent = await json_request(direct_server(), "/add", b'{"x": 1, "y": 2, "extra": 9}')232 assert response_status(sent) == 200233 assert json.loads(response_body(sent)) == {"sum": 3}234 235 async def test_body_kept_whole_when_declared(self, json_request, response_body) -> None:236 sent = await json_request(direct_server(), "/store", b'{"x": 1}')237 assert json.loads(response_body(sent)) == {"stored": {"x": 1}}