tests/spa/orchestration/test_orchestration_worker_process.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 two questions, asked of a process born each of the two ways.16 17 A spawned process is asked through the Popen its handler holds; a forked one18 through nothing but its pid. Both answer ``alive`` and ``pid``, and only the19 spawned one answers ``exit_code``, because only a parent can read a status.20 """21 22 from __future__ import annotations23 24 import os25 import signal26 import subprocess27 import sys28 import time29 30 import pytest31 32 from genro_asgi_multiworker_spa.orchestration import ForkedProcess, SpawnedProcess, WorkerProcess33 34 #: A child that waits to be killed, so a test can look at it while it is there.35 PATIENT_CHILD = [sys.executable, "-c", "import time; time.sleep(30)"]36 37 #: A pid that belongs to the system, not to us: signalling it is not permitted.38 FOREIGN_PID = 139 40 41 def bury(worker_process: SpawnedProcess) -> None:42     """Kill it and wait until the answer changes; reading ``alive`` reaps it too."""43     os.kill(worker_process.pid, signal.SIGKILL)44     deadline = time.monotonic() + 10.045     while worker_process.alive and time.monotonic() < deadline:46         time.sleep(0.01)47 48 49 def test_the_two_questions_have_no_answer_of_their_own():50     process = WorkerProcess()51 52     with pytest.raises(NotImplementedError):53         process.alive54     with pytest.raises(NotImplementedError):55         process.pid56 57 58 def test_a_spawned_process_answers_from_its_popen():59     popen = subprocess.Popen(PATIENT_CHILD)60     process = SpawnedProcess(popen)61 62     assert process.alive is True63     assert process.pid == popen.pid64     assert process.exit_code is None65 66     bury(process)67 68     assert process.alive is False69     assert process.exit_code is not None70 71 72 def test_a_forked_process_answers_from_its_pid_alone():73     popen = subprocess.Popen(PATIENT_CHILD)74     process = ForkedProcess(popen.pid)75 76     assert process.alive is True77     assert process.pid == popen.pid78     assert not hasattr(process, "exit_code")79 80     os.kill(process.pid, signal.SIGKILL)81     popen.wait()82 83     assert process.alive is False84 85 86 def test_a_forked_process_reads_a_zombie_as_alive():87     """The declared weakness, shown on purpose: this form waits for the reaping.88 89     A pid still holds a zombie until somebody collects it — the template, in the90     real machine. Until then a signal reaches it and this is the only answer the91     pid can give, which is why the WIRE stays the authoritative death.92     """93     popen = subprocess.Popen(PATIENT_CHILD)94     process = ForkedProcess(popen.pid)95     os.kill(process.pid, signal.SIGKILL)96     time.sleep(0.2)97 98     assert process.alive is True  # dead, unburied, and reading as alive99 100     popen.wait()  # the reaping a template would have done101 102     assert process.alive is False103 104 105 def test_a_pid_that_is_no_longer_ours_reads_as_gone():106     assert ForkedProcess(FOREIGN_PID).alive is False