tests/spa/test_spa_profile_grammar.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 """Contract tests for the ``orchestration`` subtree of the spa front.16 17 The whole orchestration of a spa front hangs under ONE node: the recipe writes18 ``front.orchestration(...)``, the commander under it, the groups under that.19 ``profiles_path``, ``profile_name`` and ``control_enabled`` are words of that20 node and of nothing else — written on the application element they are refused21 by name, with the new path in the message. ``env_settings`` is no grammar word22 at all: it is a dict a Python recipe builds at runtime and hands over as a plain23 constructor kwarg of the application.24 """25 26 from __future__ import annotations27 28 import asyncio29 import json30 import logging31 from pathlib import Path32 from typing import Any33 34 import pytest35 36 from genro_asgi import AsgiServer37 from genro_asgi_multiworker_spa.spa_app import (38 ORCHESTRATION_ROOT,39 WSX_ROOT,40 OrchestrationControl,41 SpaApplication,42 )43 from genro_asgi.config.builder import AsgiConfigBuilder44 from genro_asgi.config.handler import ConfigError, ConfigurationHandler45 from genro_asgi.lifespan import FatalBootError46 from genro_asgi_multiworker_spa.orchestration import SpaCommander47 48 from .test_spa_app_profiles import lifespan_startup49 50 SPA_APP_LOGGER = "genro_asgi_multiworker_spa.spa_app"51 52 53 class QuietCommander(SpaCommander):54 """The real vertex, minus the processes: nothing is launched, nothing forked."""55 56 async def start(self) -> None:57 pass58 59 async def stop(self) -> None:60 pass61 62 63 class GrammarFront(SpaApplication):64 """The front under test, with a pool that costs nothing to build."""65 66 commander_class = QuietCommander67 68 69 def recipe_with(70 root: Path,71 orchestration_words: dict[str, Any],72 with_commander: bool = True,73 app_class: type = GrammarFront,74 ) -> type[AsgiConfigBuilder]:75 """A recipe writing ``orchestration_words`` on the front's orchestration node."""76 77 class GrammarConfig(AsgiConfigBuilder):78 def main(self, configuration_root: Any) -> None:79 cfg = configuration_root.configuration()80 front = cfg.applications().application(81 code="site0", mount="", app_class=app_class82 )83 orchestration = front.orchestration(**orchestration_words)84 if not with_commander:85 return86 commander = orchestration.commander(87 frozen_users_path=str(root / "frozen_users"),88 instance_dir=str(root / "i"),89 )90 commander.groups(default="standard").group(91 name="standard", entry_module="never.launched"92 )93 94 return GrammarConfig95 96 97 def application_recipe_with(root: Path, **front_words: Any) -> type[AsgiConfigBuilder]:98 """The same recipe, with the words written on the APPLICATION element instead."""99 100 class ApplicationConfig(AsgiConfigBuilder):101 def main(self, configuration_root: Any) -> None:102 cfg = configuration_root.configuration()103 front = cfg.applications().application(104 code="site0", mount="", app_class=GrammarFront, **front_words105 )106 commander = front.orchestration().commander(107 frozen_users_path=str(root / "frozen_users"),108 instance_dir=str(root / "i"),109 )110 commander.groups(default="standard").group(111 name="standard", entry_module="never.launched"112 )113 114 return ApplicationConfig115 116 117 async def boot(server: AsgiServer) -> None:118 """Bring the front up the way the lifespan does."""119 await server.applications["site0"].on_startup()120 121 122 def test_the_orchestration_node_carries_the_three_words(tmp_path: Path) -> None:123 # wf:contract: profiles_path, profile_name and control_enabled are words of124 # wf:contract: the orchestration node, read at applications.<code>.orchestration,125 # wf:contract: and they reach the front at boot — never the constructor.126 profiles = tmp_path / "profiles"127 recipe = recipe_with(128 tmp_path,129 {130 "profiles_path": str(profiles),131 "profile_name": "busy_hours",132 "control_enabled": True,133 },134 )135 136 handler = ConfigurationHandler(recipe)137 assert handler.orchestration_kwargs("site0") == {138 "profiles_path": str(profiles),139 "profile_name": "busy_hours",140 "control_enabled": True,141 }142 143 entries, _default = ConfigurationHandler(recipe).applications()144 app_class, app_kwargs = entries[0]145 assert app_class is GrammarFront146 # Nothing of the orchestration travels as a constructor kwarg any more.147 assert app_kwargs == {"code": "site0", "mount": ""}148 149 front = AsgiServer(config=recipe).applications["site0"]150 assert front.profiles_path is None151 assert front.profile_name is None152 assert front.control_enabled is False153 154 155 async def test_the_boot_reads_the_node_onto_the_front(tmp_path: Path) -> None:156 # wf:contract: the boot lands the three words on the front and mounts the157 # wf:contract: control root — last, once the pool is up.158 profiles = tmp_path / "profiles"159 profiles.mkdir()160 (profiles / "busy_hours.json").write_text(json.dumps({"worker_memory_admission_percent": 60.0}))161 server = AsgiServer(162 config=recipe_with(163 tmp_path,164 {165 "profiles_path": str(profiles),166 "profile_name": "busy_hours",167 "control_enabled": True,168 },169 )170 )171 front = server.applications["site0"]172 assert ORCHESTRATION_ROOT not in front.internal_roots173 174 await boot(server)175 176 assert front.profiles_path == str(profiles)177 assert front.profile_name == "busy_hours"178 assert front.control_enabled is True179 assert ORCHESTRATION_ROOT in front.internal_roots180 assert front.commander.active_profile == "busy_hours"181 182 183 def test_the_commander_only_lives_under_the_orchestration(tmp_path: Path) -> None:184 # wf:contract: the old form front.commander(...) is refused by the grammar,185 # wf:contract: and the refusal names the node the commander now belongs to.186 class OldFormConfig(AsgiConfigBuilder):187 def main(self, configuration_root: Any) -> None:188 cfg = configuration_root.configuration()189 front = cfg.applications().application(190 code="site0", mount="", app_class=GrammarFront191 )192 front.commander(frozen_users_path=str(tmp_path / "frozen_users"))193 194 with pytest.raises(ValueError, match="orchestration") as refusal:195 ConfigurationHandler(OldFormConfig)196 assert "parent" in str(refusal.value)197 198 199 @pytest.mark.parametrize(200 "word", ["profiles_path", "profile_name", "orchestration_control"]201 )202 def test_a_moved_word_on_the_application_element_is_refused_by_name(203 tmp_path: Path, word: str204 ) -> None:205 # wf:contract: the three words that moved are refused by NAME on the206 # wf:contract: application element, with the new path in the message — never207 # wf:contract: as a bare unexpected-kwarg TypeError.208 recipe = application_recipe_with(tmp_path, **{word: "x"})209 210 with pytest.raises(ConfigError) as refusal:211 AsgiServer(config=recipe)212 message = str(refusal.value)213 assert word in message214 assert "applications.<code>.orchestration" in message215 216 217 def test_the_old_and_the_new_form_together_are_refused(tmp_path: Path) -> None:218 # wf:contract: a recipe writing a moved word on the application element AND219 # wf:contract: the orchestration node fails — no silent precedence.220 recipe = application_recipe_with(tmp_path, profile_name="busy_hours")221 222 with pytest.raises(ConfigError, match="profile_name"):223 AsgiServer(config=recipe)224 225 226 def test_env_settings_is_not_grammar(tmp_path: Path) -> None:227 # wf:contract: env_settings is not writable from the grammar — neither on the228 # wf:contract: orchestration node nor anywhere else: it stays a runtime dict.229 grammar_file = tmp_path / "site_grammar.json"230 AsgiConfigBuilder.to_grammar(str(grammar_file))231 document = json.dumps(json.load(grammar_file.open()))232 assert "env_settings" not in document233 234 with pytest.raises(Exception) as refusal:235 ConfigurationHandler(236 recipe_with(tmp_path, {"env_settings": {"worker_max_users": 3}})237 ).applications()238 assert "env_settings" in str(refusal.value)239 240 241 async def test_a_front_with_no_orchestration_does_not_boot(tmp_path: Path) -> None:242 # wf:contract: a spa front declared without the orchestration node is an243 # wf:contract: INCOMPLETE configuration: fatal boot, startup.failed, no244 # wf:contract: commander and no control root. Wanting no pool means245 # wf:contract: declaring no spa front.246 class NoPoolConfig(AsgiConfigBuilder):247 def main(self, configuration_root: Any) -> None:248 cfg = configuration_root.configuration()249 cfg.applications().application(code="site0", mount="", app_class=GrammarFront)250 251 server = AsgiServer(config=NoPoolConfig)252 with pytest.raises(FatalBootError, match="no 'orchestration' node"):253 await boot(server)254 255 front = server.applications["site0"]256 assert front._commander is None257 assert ORCHESTRATION_ROOT not in front.internal_roots258 259 sent = await lifespan_startup(AsgiServer(config=NoPoolConfig))260 assert [message["type"] for message in sent][0] == "lifespan.startup.failed"261 262 263 async def test_a_server_with_no_spa_front_at_all_starts(tmp_path: Path) -> None:264 # wf:contract: it is the spa front WITHOUT its orchestration that is265 # wf:contract: incomplete, never a server that declares no spa front: one266 # wf:contract: without any starts the way it always did.267 class NoFrontConfig(AsgiConfigBuilder):268 def main(self, configuration_root: Any) -> None:269 configuration_root.configuration().server(host="127.0.0.1", port=8000)270 271 server = AsgiServer(config=NoFrontConfig)272 assert not [app for app in server.applications.values() if isinstance(app, SpaApplication)]273 274 sent = await lifespan_startup(server)275 assert [message["type"] for message in sent][0] == "lifespan.startup.complete"276 277 278 async def test_an_orchestration_with_no_commander_does_not_boot(tmp_path: Path) -> None:279 # wf:contract: a declared orchestration node MUST carry a commander: profile280 # wf:contract: and control plane with no pool to act on address nothing, so281 # wf:contract: the boot is fatal and the lifespan answers startup.failed.282 server = AsgiServer(283 config=recipe_with(tmp_path, {"control_enabled": True}, with_commander=False)284 )285 286 with pytest.raises(FatalBootError, match="no commander"):287 await boot(server)288 289 front = server.applications["site0"]290 assert front._commander is None291 assert ORCHESTRATION_ROOT not in front.internal_roots292 293 fresh = AsgiServer(294 config=recipe_with(tmp_path, {"control_enabled": True}, with_commander=False)295 )296 sent = await lifespan_startup(fresh)297 assert [message["type"] for message in sent][0] == "lifespan.startup.failed"298 299 300 def counting_front(301 built: list[Any],302 mount_breaks: bool = False,303 failing_starts: int = 0,304 stop_breaks: bool = False,305 ) -> type[SpaApplication]:306 """A front whose vertex counts what the lifecycle does to it.307 308 Args:309 built: every vertex ever constructed lands here, in order.310 mount_breaks: whether ``mount_control`` raises, to exercise the rollback.311 failing_starts: how many of the first vertices raise from ``start`` —312 AFTER arming their clock, the way a half-started pool leaves one.313 stop_breaks: whether ``stop`` raises once it has done its work, to check314 that a cleanup failure never replaces the reason the boot failed.315 """316 317 class CountingCommander(SpaCommander):318 """The real vertex, with a heartbeat that sleeps and no process at all."""319 320 def __init__(self, *args: Any, **kwargs: Any) -> None:321 super().__init__(*args, **kwargs)322 self.starts = 0323 self.stops = 0324 self.quits = 0325 built.append(self)326 327 async def start(self) -> None:328 self.starts += 1329 self._heartbeat_task = asyncio.ensure_future(asyncio.sleep(3600))330 if len(built) <= failing_starts:331 raise RuntimeError("the reception would not come up")332 333 async def stop(self) -> None:334 self.stops += 1335 if self._heartbeat_task is not None:336 self._heartbeat_task.cancel()337 self._heartbeat_task = None338 if stop_breaks:339 raise RuntimeError("the pool would not go down")340 341 async def quit(self) -> None:342 self.quits += 1343 await self.stop()344 345 class CountingFront(SpaApplication):346 commander_class = CountingCommander347 348 def mount_control(self) -> None:349 if mount_breaks:350 raise RuntimeError("the router refused the branch")351 super().mount_control()352 353 return CountingFront354 355 356 def live_heartbeats(built: list[Any]) -> int:357 """How many of those vertices still hold a running clock."""358 return sum(359 1360 for commander in built361 if commander._heartbeat_task is not None and not commander._heartbeat_task.done()362 )363 364 365 async def test_a_root_the_front_already_claims_does_not_boot(tmp_path: Path) -> None:366 # wf:contract: a front whose own router already answers on _orchestration367 # wf:contract: cannot also mount the runtime configuration there. The clash368 # wf:contract: is established BEFORE anything is built, so the boot fails369 # wf:contract: with no vertex constructed, no clock running and the router370 # wf:contract: exactly as the front left it.371 built: list[Any] = []372 373 class ClashingFront(counting_front(built)): # type: ignore[misc]374 """A front that claims the control root for a page of its own."""375 376 def __init__(self, **kwargs: Any) -> None:377 super().__init__(**kwargs)378 self.route.add_branches(379 {"name": ORCHESTRATION_ROOT, "instance": OrchestrationControl(self)}380 )381 382 recipe = recipe_with(tmp_path, {"control_enabled": True}, app_class=ClashingFront)383 server = AsgiServer(config=recipe)384 front = server.applications["site0"]385 before = set(front.internal_roots)386 387 with pytest.raises(FatalBootError, match="already claimed"):388 await boot(server)389 390 assert built == []391 assert front._commander is None392 assert live_heartbeats(built) == 0393 assert set(front.internal_roots) == before394 395 sent = await lifespan_startup(AsgiServer(config=recipe))396 assert [message["type"] for message in sent][0] == "lifespan.startup.failed"397 398 399 async def test_a_mount_that_breaks_takes_the_pool_back_down(tmp_path: Path) -> None:400 # wf:contract: the mount is the last mutation, and an unexpected failure of401 # wf:contract: it rolls the pool back: stop is called, the front holds no402 # wf:contract: vertex, no clock is left running, and the boot is fatal.403 built: list[Any] = []404 server = AsgiServer(405 config=recipe_with(406 tmp_path,407 {"control_enabled": True},408 app_class=counting_front(built, mount_breaks=True),409 )410 )411 front = server.applications["site0"]412 413 with pytest.raises(FatalBootError, match="taken back down"):414 await boot(server)415 416 assert len(built) == 1417 assert (built[0].starts, built[0].stops) == (1, 1)418 assert front._commander is None419 assert live_heartbeats(built) == 0420 assert ORCHESTRATION_ROOT not in front.internal_roots421 422 423 async def test_a_second_startup_builds_no_second_pool(tmp_path: Path) -> None:424 # wf:contract: a startup on a front whose pool is already up does nothing —425 # wf:contract: one vertex, one start, one route, and no orphan clock.426 built: list[Any] = []427 server = AsgiServer(428 config=recipe_with(429 tmp_path, {"control_enabled": True}, app_class=counting_front(built)430 )431 )432 front = server.applications["site0"]433 434 await boot(server)435 await boot(server)436 437 assert len(built) == 1438 assert (built[0].starts, built[0].stops, built[0].quits) == (1, 0, 0)439 assert front.commander is built[0]440 assert live_heartbeats(built) == 1441 # The front's internal roots, each claimed ONCE: the orchestration control442 # and the channel commands of a page (#68 phase 4).443 assert set(front.route.nodes(lazy=True, forbidden=True)["routers"]) == {444 ORCHESTRATION_ROOT,445 WSX_ROOT,446 }447 448 449 async def test_a_startup_after_a_shutdown_builds_a_new_pool_and_no_second_route(450 tmp_path: Path,451 ) -> None:452 # wf:contract: startup → shutdown → startup gives TWO vertices, never at the453 # wf:contract: same time: the first is stopped and let go, its clock is gone,454 # wf:contract: and the control root is not mounted a second time.455 built: list[Any] = []456 server = AsgiServer(457 config=recipe_with(458 tmp_path, {"control_enabled": True}, app_class=counting_front(built)459 )460 )461 front = server.applications["site0"]462 463 await boot(server)464 first = front.commander465 await front.on_shutdown()466 467 assert front._commander is None468 assert (first.starts, first.stops) == (1, 1)469 assert live_heartbeats(built) == 0470 471 await boot(server)472 473 assert len(built) == 2474 assert front.commander is built[1]475 assert front.commander is not first476 assert (built[1].starts, built[1].stops) == (1, 0)477 # Only the new one holds a clock: the first was let go for good.478 assert live_heartbeats(built) == 1479 assert first.stops == 1480 481 assert ORCHESTRATION_ROOT in front.internal_roots482 # The front's internal roots, each claimed ONCE: the orchestration control483 # and the channel commands of a page (#68 phase 4).484 assert set(front.route.nodes(lazy=True, forbidden=True)["routers"]) == {485 ORCHESTRATION_ROOT,486 WSX_ROOT,487 }488 for name in ("apply", "reload", "status"):489 assert front.resolves_natively(f"/{ORCHESTRATION_ROOT}/{name}") is True490 491 await front.on_shutdown()492 493 494 async def test_a_boot_that_fails_leaves_the_router_untouched(tmp_path: Path) -> None:495 # wf:contract: nothing is mounted before the pool is up: a composition the496 # wf:contract: boot refuses claims no root at all, gate on or not.497 folder = tmp_path / "profiles"498 folder.mkdir()499 (folder / "wrong.json").write_text(json.dumps({"worker_memory_admission_percent": 200.0}))500 server = AsgiServer(501 config=recipe_with(502 tmp_path,503 {504 "profiles_path": str(folder),505 "profile_name": "wrong",506 "control_enabled": True,507 },508 )509 )510 511 with pytest.raises(FatalBootError):512 await boot(server)513 514 front = server.applications["site0"]515 assert front._commander is None516 assert ORCHESTRATION_ROOT not in front.internal_roots517 518 519 async def test_a_start_that_raises_leaves_the_front_holding_nothing(tmp_path: Path) -> None:520 # wf:contract: a pool that arms something and then fails to come up is taken521 # wf:contract: back down: one vertex, one start, one stop, no clock left, the522 # wf:contract: front holding none, the router untouched and the boot fatal.523 built: list[Any] = []524 recipe = recipe_with(525 tmp_path,526 {"control_enabled": True},527 app_class=counting_front(built, failing_starts=1),528 )529 server = AsgiServer(config=recipe)530 front = server.applications["site0"]531 before = set(front.internal_roots)532 533 with pytest.raises(FatalBootError, match="could not be brought up") as refused:534 await boot(server)535 536 assert isinstance(refused.value.__cause__, RuntimeError)537 assert len(built) == 1538 assert (built[0].starts, built[0].stops) == (1, 1)539 assert front._commander is None540 assert live_heartbeats(built) == 0541 assert set(front.internal_roots) == before542 assert ORCHESTRATION_ROOT not in front.internal_roots543 544 # The same failure through the real lifespan: the server does not start.545 fresh: list[Any] = []546 sent = await lifespan_startup(547 AsgiServer(548 config=recipe_with(549 tmp_path,550 {"control_enabled": True},551 app_class=counting_front(fresh, failing_starts=1),552 )553 )554 )555 assert [message["type"] for message in sent][0] == "lifespan.startup.failed"556 assert live_heartbeats(fresh) == 0557 558 559 async def test_a_second_attempt_after_a_failed_start_comes_up(tmp_path: Path) -> None:560 # wf:contract: the idempotent guard never protects a pool that failed to561 # wf:contract: start: the next startup builds a NEW vertex and brings it up.562 built: list[Any] = []563 server = AsgiServer(564 config=recipe_with(565 tmp_path,566 {"control_enabled": True},567 app_class=counting_front(built, failing_starts=1),568 )569 )570 front = server.applications["site0"]571 572 with pytest.raises(FatalBootError):573 await boot(server)574 await boot(server)575 576 assert len(built) == 2577 assert front.commander is built[1]578 assert (built[1].starts, built[1].stops) == (1, 0)579 assert live_heartbeats(built) == 1580 assert ORCHESTRATION_ROOT in front.internal_roots581 582 await front.on_shutdown()583 584 585 async def test_a_cleanup_that_breaks_never_hides_the_mount_failure(586 tmp_path: Path, caplog: Any587 ) -> None:588 # wf:contract: when the rollback's own stop raises, the front still ends589 # wf:contract: holding no vertex and mounts nothing, the FatalBootError still590 # wf:contract: carries the MOUNT failure as its cause, and the cleanup591 # wf:contract: failure is readable on the module logger.592 built: list[Any] = []593 server = AsgiServer(594 config=recipe_with(595 tmp_path,596 {"control_enabled": True},597 app_class=counting_front(built, mount_breaks=True, stop_breaks=True),598 )599 )600 front = server.applications["site0"]601 602 with caplog.at_level(logging.ERROR, logger=SPA_APP_LOGGER):603 with pytest.raises(FatalBootError, match="taken back down") as refused:604 await boot(server)605 606 assert str(refused.value.__cause__) == "the router refused the branch"607 assert len(built) == 1608 assert (built[0].starts, built[0].stops) == (1, 1)609 assert front._commander is None610 assert ORCHESTRATION_ROOT not in front.internal_roots611 assert "refused to go back down" in caplog.text612 assert "the pool would not go down" in caplog.text