tests/spa/test_spa_app_profiles.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 SpaApplication's profiles: the boot read and the router.16 17 Design sections 1, 2, 7, 10; test matrix T1 T2 T3 T4 T5 T8 T9 T18 T19 T23. The18 stage is the real thing — a real recipe, a real lifespan, a real vertex with one19 real group — with only the PROCESSES left out: what these tests ask is what20 configuration the pool is running on, and no child is needed to answer that.21 """22 23 from __future__ import annotations24 25 import base6426 import json27 import logging28 from pathlib import Path29 from typing import Any30 31 import pytest32 33 from tests.spa.orchestration.frame_helpers import http_reply34 35 from genro_asgi import AsgiServer36 from genro_asgi_multiworker_spa.spa_app import ORCHESTRATION_ROOT, SpaApplication37 from genro_asgi.config.builder import AsgiConfigBuilder38 from genro_asgi.lifespan import FatalBootError39 from genro_asgi.orchestration_profile_store import (40 OrchestrationProfileContentError,41 OrchestrationProfileNotFoundError,42 OrchestrationProfileStore,43 )44 from genro_asgi.server import STOPPING45 from genro_asgi_multiworker_spa.orchestration import SpaCommander46 from genro_asgi_multiworker_spa.orchestration.group_policy import GroupPolicyError47 from genro_asgi_multiworker_spa.orchestration.spa_commander import SingleGroupRequired48 49 SITE_BODY = b"the hosted site answered"50 51 52 class QuietCommander(SpaCommander):53 """The real vertex, minus the processes: nothing is launched, nothing forked."""54 55 def __init__(self, *args: Any, **kwargs: Any) -> None:56 super().__init__(*args, **kwargs)57 self.started = False58 59 async def start(self) -> None:60 self.started = True61 62 async def stop(self) -> None:63 self.started = False64 65 async def serve_request(66 self, cid: str | None, http: dict[str, Any], *, hold_timeout: float67 ) -> dict[str, Any]:68 """What the hosted site answers, so a fall-through is visible from outside."""69 return http_reply(http, {70 "result": {71 "status": 200,72 "headers": [],73 "body": base64.b64encode(SITE_BODY).decode(),74 }75 })76 77 78 79 class ProfiledFront(SpaApplication):80 """The front under test, with a pool that costs nothing to build."""81 82 commander_class = QuietCommander83 84 85 def pool_recipe(86 root: Path,87 groups: tuple[str, ...] = ("standard",),88 orchestration: dict[str, Any] | None = None,89 env_settings: dict[str, Any] | None = None,90 ) -> type[AsgiConfigBuilder]:91 """A recipe with the whole orchestration subtree and as many groups as asked for."""92 93 class PoolConfig(AsgiConfigBuilder):94 def main(self, configuration_root: Any) -> None:95 cfg = configuration_root.configuration()96 applications = cfg.applications()97 front_kwargs: dict[str, Any] = {}98 if env_settings is not None:99 front_kwargs["env_settings"] = env_settings100 front = applications.application(101 code="site0", mount="", app_class=ProfiledFront, **front_kwargs102 )103 commander = front.orchestration(**(orchestration or {})).commander(104 frozen_users_path=str(root / "frozen_users"),105 instance_dir=str(root / "i"),106 )107 if not groups:108 return109 collection = commander.groups(default=groups[0])110 for name in groups:111 collection.group(112 name=name,113 entry_module="never.launched",114 worker_memory_admission_percent=70.0,115 worker_max_number=3,116 )117 118 return PoolConfig119 120 121 def pool_server(122 root: Path,123 groups: tuple[str, ...] = ("standard",),124 *,125 env_settings: dict[str, Any] | None = None,126 **orchestration: Any,127 ) -> AsgiServer:128 """A server built the way production builds one: everything through the recipe.129 130 The three words go on the orchestration node, ``env_settings`` on the131 application element — it is a runtime dict and no grammar declares it — and132 the front is instantiated by the server out of that recipe alone.133 """134 if "profiles_path" in orchestration:135 orchestration["profiles_path"] = str(orchestration["profiles_path"])136 return AsgiServer(config=pool_recipe(root, groups, orchestration, env_settings))137 138 139 async def boot(server: AsgiServer) -> None:140 """Start the front directly, the refusal coming out unwrapped.141 142 ``on_startup`` declares any boot failure fatal by raising ``FatalBootError``143 around the refusal; what these tests assert is the refusal itself, so the144 cause comes back out as it was raised.145 """146 try:147 await server.applications["site0"].on_startup()148 except FatalBootError as fatal:149 assert fatal.__cause__ is not None150 raise fatal.__cause__ from None151 152 153 async def lifespan_startup(server: AsgiServer) -> list[dict[str, Any]]:154 """Drive the ASGI lifespan startup and return what the server sent back."""155 sent: list[dict[str, Any]] = []156 inbox = [{"type": "lifespan.startup"}, {"type": "lifespan.shutdown"}]157 158 async def receive() -> dict[str, Any]:159 return inbox.pop(0)160 161 async def send(message: dict[str, Any]) -> None:162 sent.append(message)163 164 await server.lifespan( # type: ignore[operator]165 {"type": "lifespan"}, receive, send166 )167 return sent168 169 170 async def ask(171 server: AsgiServer,172 path: str,173 *,174 method: str = "GET",175 body: dict[str, Any] | bytes | None = None,176 ) -> tuple[int, Any]:177 """One request through the whole server; the status and the decoded answer."""178 headers = [(b"accept", b"application/json")]179 payload = b""180 if body is not None:181 payload = body if isinstance(body, bytes) else json.dumps(body).encode()182 headers.append((b"content-type", b"application/json"))183 scope = {184 "type": "http",185 "method": method,186 "path": path,187 "query_string": b"",188 "headers": headers,189 }190 sent: list[dict[str, Any]] = []191 192 async def receive() -> dict[str, Any]:193 return {"type": "http.request", "body": payload, "more_body": False}194 195 async def send(message: dict[str, Any]) -> None:196 sent.append(message)197 198 await server(scope, receive, send)199 status = next(m["status"] for m in sent if m["type"] == "http.response.start")200 raw = b"".join(m.get("body", b"") for m in sent if m["type"] == "http.response.body")201 try:202 return status, json.loads(raw)203 except (json.JSONDecodeError, UnicodeDecodeError):204 return status, raw205 206 207 def written(folder: Path, name: str, profile: dict[str, Any]) -> None:208 """Store one profile the way the archive stores it."""209 OrchestrationProfileStore(folder).write(name, profile)210 211 212 # -- the boot read: four levels ----------------------------------------------213 214 215 async def test_boot_precedence_four_levels(tmp_path):216 # wf:contract: T1 — at boot the effective configuration composes217 # wf:contract: defaults ⊕ recipe_settings ⊕ profile ⊕ env_settings: the218 # wf:contract: profile overrides the recipe and env_settings overrides the219 # wf:contract: profile, key by key.220 folder = tmp_path / "profiles"221 written(folder, "fast", {"worker_memory_admission_percent": 60.0, "worker_min_life_seconds": 5.0})222 server = pool_server(223 tmp_path,224 profiles_path=folder,225 profile_name="fast",226 env_settings={"worker_min_life_seconds": 9.0},227 )228 229 await boot(server)230 231 policy = server.applications["site0"].commander.configured_group.policy232 # The profile wins over the recipe...233 assert policy.worker_memory_admission_percent == 60.0234 # ...the environment wins over the profile...235 assert policy.worker_min_life_seconds == 9.0236 # ...the recipe still owns what nobody above it named...237 assert policy.worker_max_number == 3238 # ...and the dataclass default owns what nobody named at all.239 assert policy.cpu_close_percent is None240 241 242 async def test_boot_without_named_profile_unchanged(tmp_path):243 # wf:contract: T2 — no profile_name means no profile level: behaviour244 # wf:contract: identical to today, generation 1, last_apply source "boot".245 server = pool_server(tmp_path)246 247 await boot(server)248 249 commander = server.applications["site0"].commander250 assert commander.active_profile is None251 assert commander.configuration_generation == 1252 assert commander.last_apply["source"] == "boot"253 assert commander.last_apply["digest"] is None254 # The recipe alone decided, and its own level is kept for the next apply.255 policy = commander.configured_group.policy256 assert policy.worker_memory_admission_percent == 70.0257 assert policy.worker_min_life_seconds == 60.0258 assert commander.recipe_settings == {259 "worker_memory_admission_percent": 70.0,260 "worker_max_number": 3,261 }262 263 264 async def test_boot_failure_missing_named_profile(tmp_path):265 # wf:contract: T3 — a named profile that does not exist makes on_startup266 # wf:contract: raise: the lifespan fails and the server does not start.267 server = pool_server(tmp_path, profiles_path=tmp_path / "profiles", profile_name="nowhere")268 269 with pytest.raises(OrchestrationProfileNotFoundError):270 await boot(server)271 272 # Nothing was built: there is no pool to serve with.273 with pytest.raises(RuntimeError):274 server.applications["site0"].commander275 276 # And the lifespan itself fails, so the server does not start.277 fresh = pool_server(tmp_path, profiles_path=tmp_path / "profiles", profile_name="nowhere")278 sent = await lifespan_startup(fresh)279 assert not any(message["type"] == "lifespan.startup.complete" for message in sent)280 281 282 async def test_boot_failure_invalid_profile(tmp_path, caplog):283 # wf:contract: T4 — corrupt JSON, non-object, oversize, symlink or schema284 # wf:contract: violation on the named profile: on_startup raises, the285 # wf:contract: violations are in the message on the spa app module logger.286 folder = tmp_path / "profiles"287 written(folder, "wrong", {"worker_memory_admission_percent": 200.0})288 server = pool_server(tmp_path, profiles_path=folder, profile_name="wrong")289 290 with caplog.at_level(logging.ERROR, logger="genro_asgi_multiworker_spa.spa_app"):291 with pytest.raises(GroupPolicyError) as refused:292 await boot(server)293 294 assert len(refused.value.violations) == 1295 said = caplog.text296 for violation in refused.value.violations:297 assert violation in said298 299 # The same boot fails on a file that is not a JSON object at all.300 (folder / "text.json").write_text('"not an object"')301 broken = pool_server(tmp_path, profiles_path=folder, profile_name="text")302 with pytest.raises(OrchestrationProfileContentError):303 await boot(broken)304 305 306 async def test_zero_or_multi_group_rejection(tmp_path):307 # wf:contract: T9 — a named profile with 0 or 2 groups fails the boot; a hot308 # wf:contract: apply on such a composition answers 409; without a profile and309 # wf:contract: without the gate a multi-group composition boots as today.310 folder = tmp_path / "profiles"311 written(folder, "fast", {"worker_memory_admission_percent": 60.0})312 313 for groups in ((), ("standard", "heavy")):314 server = pool_server(tmp_path, groups, profiles_path=folder, profile_name="fast")315 with pytest.raises(SingleGroupRequired):316 await boot(server)317 318 # Two groups, the gate on and nothing named: the boot is today's, and the319 # apply is what refuses — it is the one that needs a single group.320 gated = pool_server(tmp_path, ("standard", "heavy"), control_enabled=True)321 await boot(gated)322 assert set(gated.applications["site0"].commander.group_map) == {"standard", "heavy"}323 324 status, answer = await ask(325 gated, f"/{ORCHESTRATION_ROOT}/apply", method="POST", body={"worker_memory_admission_percent": 60.0}326 )327 assert status == 409328 assert "exactly one group" in answer["error"]329 330 # And with the gate off it boots as today, both groups on their own policy.331 plain = pool_server(tmp_path, ("standard", "heavy"))332 await boot(plain)333 assert set(plain.applications["site0"].commander.group_map) == {"standard", "heavy"}334 335 336 # -- the router: the gate, the three routes, the answers ----------------------337 338 339 async def test_router_gate_off_and_on(tmp_path):340 # wf:contract: T18 — gate off: _orchestration/* does not resolve natively341 # wf:contract: (the path goes to the hosted site); gate on: the three routes342 # wf:contract: resolve under _orchestration.343 closed = pool_server(tmp_path)344 await boot(closed)345 front = closed.applications["site0"]346 assert ORCHESTRATION_ROOT not in front.internal_roots347 348 status, answer = await ask(closed, f"/{ORCHESTRATION_ROOT}/status")349 assert status == 200350 assert answer == SITE_BODY351 352 opened = pool_server(tmp_path, control_enabled=True)353 await boot(opened)354 gated = opened.applications["site0"]355 assert ORCHESTRATION_ROOT in gated.internal_roots356 for route in ("apply", "reload", "status"):357 assert gated.resolves_natively(f"/{ORCHESTRATION_ROOT}/{route}") is True358 359 360 async def test_http_contract_success_and_errors(tmp_path):361 # wf:contract: T19 — 200 carries the six fields (outcome, source,362 # wf:contract: active_profile, generation, changed_settings,363 # wf:contract: effective_settings); 400 invalid body/profile with violations;364 # wf:contract: 404 reload of a missing profile; 400 reload with no name and365 # wf:contract: no active profile ("nothing to reload"); 409 not exactly one366 # wf:contract: group; 503 commander not started or server not RUNNING.367 folder = tmp_path / "profiles"368 server = pool_server(tmp_path, profiles_path=folder, control_enabled=True)369 await boot(server)370 server_commander = server.applications["site0"].commander371 apply_path = f"/{ORCHESTRATION_ROOT}/apply"372 reload_path = f"/{ORCHESTRATION_ROOT}/reload"373 374 status, answer = await ask(375 server, apply_path, method="POST", body={"worker_memory_admission_percent": 65.0}376 )377 assert status == 200378 assert set(answer) == {379 "outcome",380 "source",381 "active_profile",382 "generation",383 "changed_settings",384 "effective_settings",385 }386 assert answer["outcome"] == "applied"387 assert answer["source"] == "inline"388 assert answer["active_profile"] is None389 assert answer["generation"] == 2390 assert answer["changed_settings"] == {"worker_memory_admission_percent": 65.0}391 assert answer["effective_settings"]["worker_memory_admission_percent"] == 65.0392 393 # 400 — the body is a JSON object the schema refuses, and every violation is said.394 status, answer = await ask(395 server,396 apply_path,397 method="POST",398 body={"worker_memory_admission_percent": 200.0, "unknown_setpoint": 1},399 )400 assert status == 400401 assert "worker_memory_admission_percent" in answer["error"]402 assert "unknown_setpoint" in answer["error"]403 404 # 404 — a reload of a name the folder does not hold.405 status, answer = await ask(server, reload_path, method="POST", body={"name": "nowhere"})406 assert status == 404407 408 # 400 — nothing to reload: no name, and the inline apply left no active profile.409 status, answer = await ask(server, reload_path, method="POST", body={})410 assert status == 400411 assert "nothing to reload" in answer["error"]412 413 # 409 — the machine has no single group the setpoints could govern.414 several = pool_server(tmp_path, ("standard", "heavy"), control_enabled=True)415 await boot(several)416 status, answer = await ask(several, apply_path, method="POST", body={})417 assert status == 409418 419 # Before the boot the root is not claimed at all: the gate is mounted last,420 # once the pool is up, so an unstarted front leaves the path to the site.421 unbooted = pool_server(tmp_path, control_enabled=True)422 assert ORCHESTRATION_ROOT not in unbooted.applications["site0"].internal_roots423 424 # 503 — the pool is gone under a mounted gate, and the server that left425 # RUNNING takes nothing.426 server.applications["site0"]._commander = None427 status, answer = await ask(server, apply_path, method="POST", body={})428 assert status == 503429 server.applications["site0"]._commander = server_commander430 server.state = STOPPING431 status, answer = await ask(server, apply_path, method="POST", body={})432 assert status == 503433 434 435 async def test_profile_level_replacement(tmp_path):436 # wf:contract: T5 — apply of P1 then P2 missing one of P1's keys: that key437 # wf:contract: returns to the env_settings, recipe_settings or default value,438 # wf:contract: in that order of precedence.439 folder = tmp_path / "profiles"440 written(folder, "p1", {"worker_memory_admission_percent": 60.0, "worker_min_life_seconds": 5.0})441 written(folder, "p2", {"worker_min_life_seconds": 30.0})442 server = pool_server(443 tmp_path,444 profiles_path=folder,445 control_enabled=True,446 env_settings={"worker_max_users": 16},447 )448 await boot(server)449 reload_path = f"/{ORCHESTRATION_ROOT}/reload"450 451 status, first = await ask(server, reload_path, method="POST", body={"name": "p1"})452 assert status == 200453 assert first["effective_settings"]["worker_memory_admission_percent"] == 60.0454 455 status, second = await ask(server, reload_path, method="POST", body={"name": "p2"})456 assert status == 200457 # P1's key is not carried over: the recipe level owns it again...458 assert second["effective_settings"]["worker_memory_admission_percent"] == 70.0459 # ...P2's own key is in force...460 assert second["effective_settings"]["worker_min_life_seconds"] == 30.0461 # ...the environment still wins on its key...462 assert second["effective_settings"]["worker_max_users"] == 16463 # ...and a key nobody ever named is the default.464 assert second["effective_settings"]["cpu_close_percent"] is None465 assert second["active_profile"] == "p2"466 assert second["source"] == "profile"467 468 469 async def test_invalid_apply_all_or_nothing(tmp_path):470 # wf:contract: T8 — one violation means the state is untouched, generation471 # wf:contract: does not move, and the response is 400 with the complete472 # wf:contract: violations list.473 server = pool_server(tmp_path, control_enabled=True)474 await boot(server)475 commander = server.applications["site0"].commander476 before = commander.configured_group.policy477 478 status, answer = await ask(479 server,480 f"/{ORCHESTRATION_ROOT}/apply",481 method="POST",482 body={"worker_min_life_seconds": 5.0, "worker_memory_admission_percent": 99.0},483 )484 485 assert status == 400486 # The one violation is the cross rule, and the valid key did not land either.487 assert "worker_memory_admission_percent" in answer["error"]488 assert commander.configured_group.policy is before489 assert commander.configuration_generation == 1490 assert commander.last_apply["outcome"].startswith("rejected: ")491 492 493 async def test_status_introspection(tmp_path):494 # wf:contract: T23 — GET _orchestration/status renders active_profile,495 # wf:contract: generation, last_apply and effective_settings coherent with496 # wf:contract: the last apply, read-only, no lock taken.497 folder = tmp_path / "profiles"498 written(folder, "fast", {"worker_memory_admission_percent": 62.0})499 server = pool_server(tmp_path, profiles_path=folder, control_enabled=True)500 await boot(server)501 commander = server.applications["site0"].commander502 503 status, applied = await ask(504 server, f"/{ORCHESTRATION_ROOT}/reload", method="POST", body={"name": "fast"}505 )506 assert status == 200507 508 status, seen = await ask(server, f"/{ORCHESTRATION_ROOT}/status")509 510 assert status == 200511 assert seen["active_profile"] == "fast"512 assert seen["generation"] == applied["generation"]513 assert seen["last_apply"]["outcome"] == "applied"514 assert seen["last_apply"]["source"] == "profile"515 assert seen["last_apply"]["digest"] is not None516 assert seen["effective_settings"] == applied["effective_settings"]517 # Read-only: nothing moved, and the apply lock was never taken.518 assert commander.configuration_generation == applied["generation"]519 assert commander._configuration_lock.locked() is False520 521 522 async def test_the_retirement_quiet_travels_the_four_levels(tmp_path):523 # wf:contract: cpu_retirement_quiet_seconds is a setpoint like any other:524 # wf:contract: the recipe writes it under the group, a stored profile525 # wf:contract: overrides it, env_settings overrides the profile, and what526 # wf:contract: nobody names falls back to the dataclass default.527 folder = tmp_path / "profiles"528 written(folder, "quiet", {"cpu_retirement_quiet_seconds": 30.0})529 530 # The recipe alone.531 plain = pool_server(tmp_path)532 await boot(plain)533 assert plain.applications["site0"].commander.configured_group.policy.\534 cpu_retirement_quiet_seconds == 60.0535 536 # The profile over the recipe.537 profiled = pool_server(tmp_path, profiles_path=folder, profile_name="quiet")538 await boot(profiled)539 assert profiled.applications["site0"].commander.configured_group.policy.\540 cpu_retirement_quiet_seconds == 30.0541 542 # And the environment over the profile.543 overridden = pool_server(544 tmp_path,545 profiles_path=folder,546 profile_name="quiet",547 env_settings={"cpu_retirement_quiet_seconds": 7.5},548 )549 await boot(overridden)550 commander = overridden.applications["site0"].commander551 assert commander.configured_group.policy.cpu_retirement_quiet_seconds == 7.5552 553 # It is readable from outside like every other setpoint.554 front = overridden.applications["site0"]555 assert front.settings_status["effective_settings"]["cpu_retirement_quiet_seconds"] == 7.5556 557 558 async def test_a_profile_with_a_negative_quiet_fails_the_boot(tmp_path):559 # wf:contract: the quiet is validated with the rest of the policy: a stored560 # wf:contract: profile carrying a negative one does not start the server.561 folder = tmp_path / "profiles"562 written(folder, "wrong", {"cpu_retirement_quiet_seconds": -1.0})563 server = pool_server(tmp_path, profiles_path=folder, profile_name="wrong")564 565 with pytest.raises(GroupPolicyError) as refused:566 await boot(server)567 568 assert any("cpu_retirement_quiet_seconds" in v for v in refused.value.violations)