tests/core/test_server_monitor.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 ``_server/monitor`` section: the page, the snapshot, the panels.16 17 Requests drive a REAL ``AsgiServer`` at the ASGI level: the ``_server`` app is18 auto-mounted, so the monitor lives at ``/_server/monitor/...``. A test19 middleware (order 500, after the real AuthMiddleware) stamps a fixed identity20 on the scope, so the ``SERVER_ADMIN`` gate is exercised with a real avatar.21 22 What the suite pins:23 24 - the gate — an anonymous request is challenged (401, which the error25 middleware turns into a login) and a wrong-tag one refused (403), on every26 route of the section;27 - the aggregate — one entry per mounted application, keyed by mount, the28 ``_server`` app itself absent (the monitor is its face, not a tab);29 - the contract — an application that overrides ``app_snapshot``/``app_panel``30 is carried through verbatim, one that overrides nothing still shows up with31 its identity facts and the generic panel;32 - the shipped panel — an app declaring ``panel_source`` gets its ``src`` filled33 in and its module served, while an explicit ``src`` is left alone;34 - the page — served as HTML at the section root;35 - the bootstrap admin — it carries ``SERVER_ADMIN``, so a freshly installed36 server is observable by the identity that configures it.37 """38 39 from __future__ import annotations40 41 import json42 from typing import Any43 44 from genro_asgi import AsgiServer, Avatar, BaseApplication, UserStore45 from genro_asgi.middleware.base import BaseMiddleware46 from genro_asgi.types import Message, Scope47 48 49 class MemoryUserStore(UserStore):50 """In-memory ``UserStore`` backend: the contract suite over a dict."""51 52 __slots__ = ("_records",)53 54 def __init__(self) -> None:55 self._records: dict[str, dict[str, Any]] = {}56 57 def load_all(self) -> list[dict[str, Any]]:58 return list(self._records.values())59 60 def get(self, identity: str) -> dict[str, Any] | None:61 return self._records.get(identity)62 63 def save(self, record: dict[str, Any]) -> None:64 self._records[record["identity"]] = record65 66 def delete(self, identity: str) -> bool:67 return self._records.pop(identity, None) is not None68 69 70 class StampAuthMiddleware(BaseMiddleware):71 """Test middleware (order 500): stamps a fixed identity on ``scope["auth"]``."""72 73 middleware_order = 50074 75 def __init__(self, app: Any, server: Any, *, avatar: Avatar | None = None, **options: Any):76 self._avatar = avatar77 super().__init__(app, server, **options)78 79 async def __call__(self, scope: Any, receive: Any, send: Any) -> None:80 scope["auth"] = self._avatar81 await self.app(scope, receive, send)82 83 84 class RichApplication(BaseApplication):85 """An app that declares its own monitor face, panel included."""86 87 code = "rich"88 89 @property90 def app_snapshot(self) -> dict[str, Any]:91 return {**super().app_snapshot, "orders": 7}92 93 @property94 def app_panel(self) -> dict[str, Any]:95 return {"panel": "orders", "src": "./orders_panel.js"}96 97 98 class ShippingApplication(BaseApplication):99 """An app that ships its panel module instead of serving it itself."""100 101 code = "shipping"102 103 @property104 def app_panel(self) -> dict[str, Any]:105 return {"panel": "shipping"}106 107 @property108 def panel_source(self) -> str:109 return "export default { render(target, context) {} };"110 111 112 SERVER_ADMIN = Avatar("ops", ["SERVER_ADMIN"])113 MONITOR_ROUTES = ("/_server/monitor/", "/_server/monitor/snapshot", "/_server/monitor/panels")114 115 116 def make_server(avatar: Avatar | None, *applications: BaseApplication) -> AsgiServer:117 """A server whose chain stamps ``avatar``, mounting ``applications``."""118 return AsgiServer(119 applications=list(applications) or [BaseApplication(mount="")],120 middleware={"stamp": {"avatar": avatar}},121 middleware_registry={"stamp": StampAuthMiddleware},122 )123 124 125 async def drive(server: AsgiServer, path: str, accept: bytes | None = None) -> list[Message]:126 """One GET through ``server``, query string split off the path.127 128 The ``http_request`` fixture puts the whole string in ``path``; ``panel``129 is addressed by query, so it needs the ASGI split.130 """131 path, _, query = path.partition("?")132 headers = [(b"accept", accept)] if accept else []133 scope: Scope = {134 "type": "http",135 "method": "GET",136 "path": path,137 "query_string": query.encode(),138 "headers": headers,139 }140 sent: list[Message] = []141 142 async def receive() -> Message:143 return {"type": "http.request"}144 145 async def send(message: Message) -> None:146 sent.append(message)147 148 await server(scope, receive, send)149 return sent150 151 152 def payload(sent: list[Message]) -> Any:153 body = b"".join(m.get("body", b"") for m in sent if m["type"] == "http.response.body")154 return json.loads(body)155 156 157 class TestMonitorGate:158 """``auth_rule="SERVER_ADMIN"`` on the page and both data endpoints."""159 160 async def test_anonymous_is_challenged(self, http_request, response_status) -> None:161 """No identity: 401, the status ``ErrorMiddleware`` turns into a login."""162 server = make_server(None)163 for path in MONITOR_ROUTES:164 assert response_status(await http_request(server, path)) == 401165 166 async def test_wrong_tags_are_forbidden(self, http_request, response_status) -> None:167 server = make_server(Avatar("bob", ["SUPERADMIN"]))168 for path in MONITOR_ROUTES:169 assert response_status(await http_request(server, path)) == 403170 171 async def test_server_admin_is_allowed(self, http_request, response_status) -> None:172 server = make_server(SERVER_ADMIN)173 for path in MONITOR_ROUTES:174 assert response_status(await http_request(server, path)) == 200175 176 async def test_a_browser_lands_on_the_login_page(177 self, http_request, response_status, response_headers178 ) -> None:179 """The whole point of the 401: an operator opening the monitor gets a login.180 181 ``ErrorMiddleware`` negotiates the challenge only when the server has a182 login surface, which ``ServerApplication`` always registers.183 """184 sent = await http_request(185 make_server(None), "/_server/monitor/", headers=[(b"accept", b"text/html")]186 )187 assert response_status(sent) == 302188 location = response_headers(sent)[b"location"].decode()189 assert location.startswith("/_server/login_page?next=")190 assert "monitor" in location191 192 193 class TestMonitorPage:194 """The shell itself, served at the section root."""195 196 async def test_page_is_html(self, http_request, response_headers, response_body) -> None:197 sent = await http_request(make_server(SERVER_ADMIN), "/_server/monitor/")198 assert response_headers(sent)[b"content-type"].startswith(b"text/html")199 assert b"genro" in response_body(sent)200 201 202 class TestSnapshot:203 """The polled aggregate: server facts plus one entry per application."""204 205 async def test_server_facts(self, http_request) -> None:206 server = make_server(SERVER_ADMIN, BaseApplication(mount=""))207 facts = payload(await http_request(server, "/_server/monitor/snapshot"))["server"]208 assert facts["pid"] > 0209 assert "monitor" in facts["sections"]210 211 async def test_one_entry_per_application(self, http_request) -> None:212 server = make_server(SERVER_ADMIN, BaseApplication(mount=""), RichApplication())213 apps = payload(await http_request(server, "/_server/monitor/snapshot"))["apps"]214 assert set(apps) == {"", "rich"}215 216 async def test_server_app_is_not_a_tab(self, http_request) -> None:217 server = make_server(SERVER_ADMIN)218 apps = payload(await http_request(server, "/_server/monitor/snapshot"))["apps"]219 assert "_server" not in apps220 221 async def test_identity_facts_by_default(self, http_request) -> None:222 server = make_server(SERVER_ADMIN, BaseApplication(code="shop", mount="shop"))223 apps = payload(await http_request(server, "/_server/monitor/snapshot"))["apps"]224 assert apps["shop"] == {225 "class": "BaseApplication",226 "code": "shop",227 "mount": "shop",228 }229 230 async def test_an_app_extends_its_own_entry(self, http_request) -> None:231 server = make_server(SERVER_ADMIN, RichApplication())232 apps = payload(await http_request(server, "/_server/monitor/snapshot"))["apps"]233 assert apps["rich"]["orders"] == 7234 assert apps["rich"]["code"] == "rich"235 236 237 class TestPanels:238 """The descriptors: who draws what, fetched once."""239 240 async def test_generic_by_default(self, http_request) -> None:241 server = make_server(SERVER_ADMIN, BaseApplication(code="shop", mount="shop"))242 panels = payload(await http_request(server, "/_server/monitor/panels"))243 assert panels["shop"] == {"panel": "generic"}244 245 async def test_a_declared_panel_carries_its_module(self, http_request) -> None:246 server = make_server(SERVER_ADMIN, RichApplication())247 panels = payload(await http_request(server, "/_server/monitor/panels"))248 assert panels["rich"] == {"panel": "orders", "src": "./orders_panel.js"}249 250 async def test_a_shipped_module_gets_its_src_filled_in(self, http_request) -> None:251 """Declaring ``panel_source`` is enough: the app publishes no route."""252 server = make_server(SERVER_ADMIN, ShippingApplication())253 panels = payload(await http_request(server, "/_server/monitor/panels"))254 assert panels["shipping"] == {255 "panel": "shipping",256 "src": "/_server/monitor/panel?app=shipping",257 }258 259 async def test_an_explicit_src_wins(self, http_request) -> None:260 """An app free to serve the module itself keeps its own address."""261 server = make_server(SERVER_ADMIN, RichApplication())262 panels = payload(await http_request(server, "/_server/monitor/panels"))263 assert panels["rich"]["src"] == "./orders_panel.js"264 265 async def test_panels_and_snapshot_agree(self, http_request) -> None:266 server = make_server(SERVER_ADMIN, BaseApplication(mount=""), RichApplication())267 panels = payload(await http_request(server, "/_server/monitor/panels"))268 apps = payload(await http_request(server, "/_server/monitor/snapshot"))["apps"]269 assert set(panels) == set(apps)270 271 272 class TestPanelModule:273 """``panel``: the module a contributor ships, served as an ES module."""274 275 async def test_the_module_is_served_as_javascript(276 self, response_headers, response_body277 ) -> None:278 server = make_server(SERVER_ADMIN, ShippingApplication())279 sent = await drive(server, "/_server/monitor/panel?app=shipping")280 assert response_headers(sent)[b"content-type"].startswith(b"text/javascript")281 assert b"export default" in response_body(sent)282 283 async def test_an_app_without_a_module_is_not_found(self, response_status) -> None:284 server = make_server(SERVER_ADMIN, BaseApplication(code="plain", mount="plain"))285 sent = await drive(server, "/_server/monitor/panel?app=plain")286 assert response_status(sent) == 404287 288 async def test_an_unknown_app_is_not_found(self, response_status) -> None:289 sent = await drive(make_server(SERVER_ADMIN), "/_server/monitor/panel?app=ghost")290 assert response_status(sent) == 404291 292 async def test_the_module_is_gated_like_the_rest(self, response_status) -> None:293 server = make_server(None, ShippingApplication())294 sent = await drive(server, "/_server/monitor/panel?app=shipping")295 assert response_status(sent) == 401296 297 298 class TestBootstrapAdmin:299 """The admin the server seeds at boot can reach its own monitor."""300 301 async def test_admin_carries_the_monitor_tag(self) -> None:302 store = MemoryUserStore()303 AsgiServer(users=store, admin_password="opspassword")304 assert "SERVER_ADMIN" in store.get("admin")["tags"]