tests/spa/orchestration/test_orchestration_template_entry.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 template's whole contract: the first line, the refusals, and one real fork.16 17 Everything here drives ``TemplateEntry`` on paper pipes, which is exactly what it18 was given: a line source and somewhere to answer. The last test forks for real —19 the child writes what it found and leaves with ``os._exit``, because a pytest20 process must not be run twice.21 """22 23 from __future__ import annotations24 25 import gc26 import io27 import json28 import os29 import signal30 import threading31 import time32 from typing import Any33 34 import pytest35 36 from genro_asgi_multiworker_spa.orchestration import TemplateEntry37 38 @pytest.fixture(autouse=True)39 def sigchld_put_back():40 """Give ``SIGCHLD`` back to whoever had it, and thaw what was frozen.41 42 ``serve`` installs the reaper on the PROCESS, and here that process is the43 test runner: left in place it would collect every later test's children, and44 a Popen whose child somebody else buried does not behave.45 """46 had = signal.getsignal(signal.SIGCHLD)47 yield48 signal.signal(signal.SIGCHLD, had)49 # And ``serve`` freezes the heap of the PROCESS too, which here is the runner's:50 # left frozen, no later test's garbage would ever be collected.51 gc.unfreeze()52 53 54 TEMPLATE_NAME = "template-standard"55 WORKER_NAME = "standard_0001"56 57 58 class EngineFactory:59 """The seam the deployment fills: a class asked once for the group's engine."""60 61 def __init__(self, mark: str = "plain") -> None:62 self.mark = mark63 64 def build_group_engine(self) -> str:65 """The engine, played here by a string the tests can recognise."""66 return f"engine-{self.mark}"67 68 69 FACTORY = f"{__name__}:EngineFactory"70 71 72 def launch_line(**overrides: Any) -> str:73 """One launch line, with whatever a test wants different in it."""74 config: dict[str, Any] = {"name": TEMPLATE_NAME, "engine_factory": FACTORY}75 config.update(overrides)76 return json.dumps(config) + "\n"77 78 79 def fork_line(name: str = WORKER_NAME) -> str:80 """One fork request: a worker's spawn payload, cut to what these tests read back."""81 return json.dumps(82 {"name": name, "uds_url": "uds:/nowhere.sock", "frozen_users_path": "/nowhere"}83 ) + "\n"84 85 86 def template_on(lines: str) -> tuple[TemplateEntry, io.StringIO]:87 """A template reading those lines, and the paper it answers on."""88 answers = io.StringIO()89 return TemplateEntry(pipe_in=io.StringIO(lines), pipe_out=answers), answers90 91 92 def answers_of(paper: io.StringIO) -> list[dict[str, Any]]:93 """The answer lines, parsed."""94 return [json.loads(line) for line in paper.getvalue().splitlines()]95 96 97 # ----------------------------------------------------------------------98 # The first line99 # ----------------------------------------------------------------------100 101 102 def test_the_launch_line_builds_the_engine_of_the_group():103 template, _ = template_on(launch_line(kwargs={"mark": "standard"}))104 105 template.run()106 107 assert template.name == TEMPLATE_NAME108 assert template.group_engine == "engine-standard"109 110 111 def test_what_the_template_built_is_frozen_before_any_fork():112 template, _ = template_on(launch_line())113 114 template.run()115 116 assert gc.get_freeze_count() > 0117 # And what comes after the freeze is the collector's again, which is what a118 # child allocates while it serves.119 watched = len(gc.get_objects())120 fresh = [[i] for i in range(100)]121 assert len(gc.get_objects()) >= watched + len(fresh)122 123 124 def test_a_launch_line_without_a_factory_ends_the_process():125 template, _ = template_on(json.dumps({"name": TEMPLATE_NAME}) + "\n")126 127 with pytest.raises(SystemExit, match="engine_factory"):128 template.run()129 130 131 def test_a_launch_line_that_is_not_a_json_object_ends_the_process():132 template, _ = template_on("[1, 2, 3]\n")133 134 with pytest.raises(SystemExit, match="must be a JSON object"):135 template.run()136 137 138 def test_a_launch_line_that_is_not_json_ends_the_process():139 template, _ = template_on("not json at all\n")140 141 with pytest.raises(SystemExit, match="not valid JSON"):142 template.run()143 144 145 def test_a_pipe_that_closes_before_the_launch_line_ends_the_process():146 template, _ = template_on("")147 148 with pytest.raises(SystemExit, match="closed before the launch line"):149 template.run()150 151 152 def test_a_factory_that_is_not_a_reference_is_refused():153 template, _ = template_on(launch_line(engine_factory="some.module.EngineFactory"))154 155 with pytest.raises(SystemExit, match="module.path:ClassName"):156 template.run()157 158 159 # ----------------------------------------------------------------------160 # Serving, and refusing161 # ----------------------------------------------------------------------162 163 164 def test_the_pipe_ending_takes_the_template_out():165 template, answers = template_on(launch_line())166 167 assert template.run() == 0168 assert answers_of(answers) == []169 170 171 def test_the_invariant_reads_the_threads_of_this_process():172 template, _ = template_on(launch_line())173 174 assert template.live_thread_count == threading.active_count()175 176 177 def test_a_second_thread_refuses_the_fork_out_loud():178 class NeverForks(TemplateEntry):179 """A template with company, which records the birth it must never reach."""180 181 births: list[dict[str, Any]] = []182 183 @property184 def live_thread_count(self) -> int:185 return 2186 187 def live_as_worker(self, payload: dict[str, Any]) -> None:188 self.births.append(payload)189 190 answers = io.StringIO()191 template = NeverForks(192 pipe_in=io.StringIO(launch_line() + fork_line()), pipe_out=answers193 )194 195 template.run()196 197 answer = answers_of(answers)[0]198 assert "2 threads alive" in answer["error"]199 assert TEMPLATE_NAME in answer["error"]200 assert NeverForks.births == []201 202 203 # ----------------------------------------------------------------------204 # One real fork205 # ----------------------------------------------------------------------206 207 208 def test_the_forked_child_takes_its_own_session_and_finds_payload_and_engine(tmp_path):209 report = tmp_path / "what_the_child_found.json"210 211 class ReportingTemplate(TemplateEntry):212 """A template whose child says what it found instead of serving.213 214 ``become_worker`` is the seam: everything before it — the fork, the session,215 the pipes — is the real thing under test. The thread count is declared216 because a test runner is never a one-thread process, which a template is.217 """218 219 @property220 def live_thread_count(self) -> int:221 return 1222 223 def become_worker(self, payload: dict[str, Any]) -> None:224 report.write_text(225 json.dumps(226 {227 "worker": payload["name"],228 "engine": self.group_engine,229 "session": os.getsid(0),230 "pipes_closed": self.pipe_in.closed and self.pipe_out.closed,231 }232 )233 )234 os._exit(0)235 236 answers = io.StringIO()237 template = ReportingTemplate(238 pipe_in=io.StringIO(launch_line() + fork_line()), pipe_out=answers239 )240 template.run()241 242 pid = answers_of(answers)[0]["pid"]243 assert pid > 0244 deadline = time.monotonic() + 10.0245 while not report.exists() and time.monotonic() < deadline:246 time.sleep(0.01)247 found = json.loads(report.read_text())248 249 assert found["worker"] == WORKER_NAME250 assert found["engine"] == "engine-plain"251 assert found["pipes_closed"] is True252 assert found["session"] != os.getsid(0)