tests/spa/test_inspector_section.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/inspector`` section: mounted by the env var, and by nothing else.16 17 Contract: the mount IS the gate — without ``GNR_ASGI_INSPECTOR`` the section18 does not exist at all; with it the page is HTML and the census is JSON, keyed by19 the code of the front whose pool it watches. And in neither case does the20 inspector touch the hosted site: no connection cookie ever comes back from it.21 22 The front attaches it, on its own startup, so every server here goes through its23 lifespan — the section does not exist before the pool does.24 """25 26 from __future__ import annotations27 28 import json29 from typing import Any30 31 import pytest32 33 from genro_asgi import AsgiServer, ServerApplication34 from genro_asgi.config.builder import AsgiConfigBuilder35 from genro_asgi.lifespan import FatalBootError36 from genro_asgi_multiworker_spa.inspector_section import INSPECTOR_ENV_VAR37 from genro_asgi_multiworker_spa.orchestration import SpaCommander38 from genro_asgi_multiworker_spa.spa_app import SPA_CONNECTION_ID_COOKIE, SpaApplication39 40 from ..conftest import LifespanRunner41 42 43 class InspectorScriptedCommander(SpaCommander):44     """A pool that comes up without processes: no wire, no beat, no worker."""45 46     async def start(self) -> None:47         pass48 49     async def stop(self) -> None:50         pass51 52 53 class InspectorScriptedFront(SpaApplication):54     """The front under test: the real startup, over a pool that launches nothing."""55 56     commander_class = InspectorScriptedCommander57 58 59 def inspector_recipe_for(root) -> type[AsgiConfigBuilder]:60     """A recipe with one spa front and its pool, mounted at the root."""61 62     class FrontConfig(AsgiConfigBuilder):63         def main(self, configuration_root: Any) -> None:64             cfg = configuration_root.configuration()65             front = cfg.applications().application(66                 code="site", mount="", app_class=InspectorScriptedFront67             )68             commander = front.orchestration().commander(69                 frozen_users_path=str(root / "frozen_users"),70                 instance_dir=str(root / "i"),71             )72             commander.groups(default="standard").group(73                 name="standard", entry_module="never.launched"74             )75 76     return FrontConfig77 78 79 @pytest.fixture80 async def inspector_server(tmp_path, monkeypatch):81     """A started server whose front attached the inspector on startup."""82     monkeypatch.setenv(INSPECTOR_ENV_VAR, "1")83     server = AsgiServer(config=inspector_recipe_for(tmp_path))84     runner = LifespanRunner(server)85     await runner.startup()86     yield server87     await runner.shutdown()88 89 90 @pytest.fixture91 async def plain_server(tmp_path, monkeypatch):92     """The same server with the variable unset: the section is not there."""93     monkeypatch.delenv(INSPECTOR_ENV_VAR, raising=False)94     server = AsgiServer(config=inspector_recipe_for(tmp_path))95     runner = LifespanRunner(server)96     await runner.startup()97     yield server98     await runner.shutdown()99 100 101 async def test_without_the_env_var_the_page_is_not_there(102     plain_server, http_request, response_status103 ):104     sent = await http_request(plain_server, "/_server/inspector/page")105 106     assert response_status(sent) == 404107 108 109 async def test_the_page_is_html(110     inspector_server, http_request, response_status, response_headers, response_body111 ):112     sent = await http_request(inspector_server, "/_server/inspector/page")113 114     assert response_status(sent) == 200115     assert response_headers(sent)[b"content-type"].startswith(b"text/html")116     assert b"worker-grid" in response_body(sent)117 118 119 async def test_the_census_is_json_under_the_front_code(120     inspector_server, http_request, response_status, response_headers, response_body121 ):122     sent = await http_request(inspector_server, "/_server/inspector/census")123 124     assert response_status(sent) == 200125     assert response_headers(sent)[b"content-type"].startswith(b"application/json")126     census = json.loads(response_body(sent))127     assert set(census) == {"site"}128     assert census["site"]["user_map"] == {}129 130 131 async def test_the_inspector_mints_no_cookie(inspector_server, http_request, response_headers):132     for path in ("/_server/inspector/page", "/_server/inspector/census"):133         headers = await http_request(inspector_server, path)134 135         cookie = response_headers(headers).get(b"set-cookie", b"")136         assert SPA_CONNECTION_ID_COOKIE.encode() not in cookie137 138 139 async def test_the_stream_opens_with_the_census(inspector_server, sse_request):140     connection = await sse_request(inspector_server, "/_server/inspector/stream")141 142     frames = await connection.wait_frames(2)143     await connection.close()144 145     assert b"retry: 2000" in frames[0]146     assert b"event: census" in frames[1]147 148 149 async def test_the_page_carries_its_containers_and_its_endpoints(150     inspector_server, http_request, response_body151 ):152     page = response_body(await http_request(inspector_server, "/_server/inspector/page"))153 154     for container_id in ("commander-panel", "worker-grid", "event-log", "last-read", "toggle-stream"):155         assert f'id="{container_id}"'.encode() in page156     assert b'"/census"' in page157     assert b'"/stream"' in page158 159 160 async def test_no_server_app_leaves_the_inspector_unmounted(inspector_server, caplog):161     """The variable set and no ``_server`` to host it: said once, nothing attached."""162     front = inspector_server.applications["site"]163     del inspector_server.applications["_server"]164 165     with caplog.at_level("WARNING"):166         front.mount_inspector()167 168     assert INSPECTOR_ENV_VAR in caplog.text169     assert "nowhere to go" in caplog.text170 171 172 async def test_a_second_front_on_one_server_does_not_boot(tmp_path, monkeypatch):173     """A server has ONE orchestrated application: the second attach is fatal."""174     monkeypatch.setenv(INSPECTOR_ENV_VAR, "1")175     server = AsgiServer(config=inspector_recipe_for(tmp_path))176     front = server.applications["site"]177     await LifespanRunner(server).startup()178 179     second = InspectorScriptedFront(code="site2", mount="other")180     server.register_application(second)181 182     with pytest.raises(FatalBootError, match="already attached"):183         second.mount_inspector()184 185     server_app = server.applications["_server"]186     assert isinstance(server_app, ServerApplication)187     assert server_app.sections["inspector"].application is front