tests/spa/orchestration/test_orchestration_audit_destinations.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 """Where the audit of a configuration attempt lands — three destinations (T22).16 17 Design section 9: a boot that fails before the vertex exists can only speak on18 the front's own module logger; anything that reaches the handler leaves an19 orchestration row, refusals included; a body the request layer never hydrated is20 the ONE thing that leaves no row at all.21 """22 23 from __future__ import annotations24 25 import logging26 27 import pytest28 29 from genro_asgi_multiworker_spa.spa_app import ORCHESTRATION_ROOT30 from genro_asgi_multiworker_spa.orchestration.group_policy import GroupPolicyError31 from genro_asgi_multiworker_spa.orchestration.spa_commander import ORDERS_LOGGER_NAME32 33 from ..test_spa_app_profiles import ask, boot, pool_server, written34 35 SPA_APP_LOGGER = "genro_asgi_multiworker_spa.spa_app"36 37 38 async def test_boot_failure_logs_on_spa_app_logger(tmp_path, caplog):39     # wf:contract: T22a — a failed boot logs on the SpaApplication module logger40     # wf:contract: (caplog) with the violations, and leaves NO orchestration log41     # wf:contract: line: the commander does not exist yet.42     folder = tmp_path / "profiles"43     written(folder, "wrong", {"memory_max_percent": 0.0})44     server = pool_server(tmp_path, profiles_path=folder, profile_name="wrong")45     front = server.applications["site0"]46 47     with caplog.at_level(logging.INFO):48         with pytest.raises(GroupPolicyError) as refused:49             await boot(server)50 51     said = [52         record.getMessage() for record in caplog.records if record.name == SPA_APP_LOGGER53     ]54     assert len(said) == 155     for violation in refused.value.violations:56         assert violation in said[0]57     # There is nothing to write an orchestration row with, and none was written.58     assert front._commander is None59     assert not [record for record in caplog.records if record.name == ORDERS_LOGGER_NAME]60 61 62 async def test_reload_handler_errors_audited_as_rejected(tmp_path, caplog):63     # wf:contract: T22b — a reload of a missing or corrupt profile that reaches64     # wf:contract: the handler leaves a commander log_order line with outcome65     # wf:contract: "rejected: ...".66     folder = tmp_path / "profiles"67     server = pool_server(tmp_path, profiles_path=folder, control_enabled=True)68     await boot(server)69     (folder / "corrupt.json").write_text("{not json")70 71     with caplog.at_level(logging.INFO, logger=ORDERS_LOGGER_NAME):72         missing, _ = await ask(73             server, f"/{ORCHESTRATION_ROOT}/reload", method="POST", body={"name": "nowhere"}74         )75         corrupt, _ = await ask(76             server, f"/{ORCHESTRATION_ROOT}/reload", method="POST", body={"name": "corrupt"}77         )78 79     assert (missing, corrupt) == (404, 400)80     rows = [81         record.getMessage()82         for record in caplog.records83         if record.name == ORDERS_LOGGER_NAME and "apply_group_settings" in record.args84     ]85     assert len(rows) == 286     for row in rows:87         assert "outcome=rejected: " in row88     # The vertex's own record says the same, and nothing moved.89     commander = server.applications["site0"].commander90     assert commander.last_apply["outcome"].startswith("rejected: ")91     assert commander.configuration_generation == 192 93 94 async def test_request_parser_400_is_not_orchestration_audit(tmp_path, caplog):95     # wf:contract: T22c — a body the handler cannot take is answered by the96     # wf:contract: dispatcher (422 here: the malformed JSON reaches the handler97     # wf:contract: as a string and pydantic rejects it) and leaves NO98     # wf:contract: orchestration log line: the single exclusion from the audit.99     server = pool_server(tmp_path, control_enabled=True)100     await boot(server)101 102     with caplog.at_level(logging.INFO, logger=ORDERS_LOGGER_NAME):103         status, _ = await ask(104             server, f"/{ORCHESTRATION_ROOT}/apply", method="POST", body=b"{not json"105         )106 107     assert status == 422108     assert not [record for record in caplog.records if record.name == ORDERS_LOGGER_NAME]109     commander = server.applications["site0"].commander110     # The last attempt on record is still the boot: this one never reached the vertex.111     assert commander.last_apply["source"] == "boot"