tests/spa/orchestration/test_orchestration_foundations_e2e.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 foundations end to end: a worker is born, works, dies wild, and leaves its traces.16 17 The three foundations are exercised together, in one story, on the real things:18 a real child process (``child_stub``), a real Unix socket, a real deposit on19 disk. Nothing here is doubled except the level above — ``GroupStub``, because20 the GroupHandler is Macro 2's — and the test reads the deposit from the parent21 side through a FreezeHandler of its own, over the same root the child was given.22 23 The story: the handler launches its process, which presents itself — carrying24 its first photo — and is answered with the whole global store; the beat asks25 whether it is alive; the child takes a user's semaphore, parks that user's26 connection under it and gives the semaphore back; it takes it again, and while it27 holds it the process goes mute and the surveillance kills it.28 29 **What the death leaves is the point.** The handler writes ``aborted`` — the30 death nobody was waiting for — rings its group's wake and stops there; the users31 that were on board are still on its list for whoever reads it at that round. The32 parcel is still in the deposit, the semaphore of the interrupted operation is33 still held by a process that no longer exists. Macro 1 cleans nothing: the sweep34 of the traces belongs to the Commander, which does not exist yet, and this test35 is the picture it will inherit.36 37 The sockets and the deposit live under a short ``mkdtemp`` root: the system caps38 a UDS path at about a hundred characters and pytest's own directory is already39 past it, which is the very reason worker names are short.40 """41 42 from __future__ import annotations43 44 import logging45 import time46 from typing import Any47 48 import pytest49 50 from genro_asgi_multiworker_spa.orchestration import FreezeHandler, WorkerHandler51 52 from .child_stub import (53 GO_MUTE_OP,54 RELEASE_LOCK_OP,55 TAKE_LOCK_OP,56 WRITE_CONNECTION_REGISTER_ITEM_OP,57 )58 from .group_stub import GroupStub59 from .conftest import kill_process, wait_for60 61 CHILD_MODULE = "tests.spa.orchestration.child_stub"62 PARKED_CONNECTION = {"cid": "c-1", "pages": ["main", "invoices"]}63 64 65 @pytest.fixture66 def group(short_root):67 return GroupStub(short_root / "frozen_users")68 69 70 @pytest.fixture71 def deposit(short_root):72 """The deposit as the parent reads it — the same root the child is given."""73 return FreezeHandler(short_root / "frozen_users")74 75 76 @pytest.fixture77 async def handler(short_root, group, repo_on_pythonpath):78 """The handler under test; no process and no socket of its own outlives the test."""79 worker_handler = WorkerHandler(80 group,81 "standard_0001",82 instance_dir=short_root / "i",83 frozen_users_path=short_root / "frozen_users",84 entry_module=CHILD_MODULE,85 worker_kwargs={"group": "standard"},86 # Wide at birth: the same value bounds the wait for the presentation,87 # and a fresh interpreter on a loaded machine can take seconds to get88 # there. The mute phase tightens it to what it really measures.89 process_ping_timeout=10.0,90 )91 group.worker_handler = worker_handler92 yield worker_handler93 if worker_handler.process is not None:94 kill_process(worker_handler.process)95 await wait_for(lambda: not worker_handler.process.alive)96 await worker_handler.connector.stop()97 98 99 async def order(handler: WorkerHandler, path: str, data: Any = None) -> Any:100 """Drive one order of the child through the wire and give back what it answered."""101 payload = await handler.connector.call(path, data, timeout=5.0)102 return payload["result"]103 104 105 async def test_a_worker_is_born_works_dies_wild_and_leaves_its_traces_behind(106 handler, group, deposit, caplog107 ):108 caplog.set_level(logging.INFO)109 handler.hosted_users.update({"mario", "anna"})110 111 # It is born: the process presents itself on its handler's own socket, and112 # the presentation already carries its first photo — a live process is never113 # without one.114 await handler.launch_process()115 assert handler.connector.connected is True116 assert handler.worker_snapshot == {"pid": handler.process.pid, "name": "standard_0001"}117 118 # It is alive: the beat asks that and nothing else, and a fresh photo rides119 # the answer.120 await handler.ping_process()121 assert handler.worker_snapshot == {"pid": handler.process.pid, "name": "standard_0001"}122 123 # It takes the semaphore of one of its users. Nothing is announced upward:124 # the lock is the deposit's own mechanism, and the vertex already knows —125 # it suspended mario before any of this.126 assert await order(handler, TAKE_LOCK_OP, {"user": "mario"}) == {"taken": True}127 assert deposit.lock_holder("mario") == "standard_0001"128 assert deposit.read_connection_register_item("mario", "c-1") is None129 130 # It parks that user's connection under the semaphore it holds, and the131 # parent reads back from the deposit exactly what the child wrote.132 written = await order(133 handler,134 WRITE_CONNECTION_REGISTER_ITEM_OP,135 {"user": "mario", "cid": "c-1", "payload": PARKED_CONNECTION},136 )137 assert written == {"written": "c-1"}138 assert deposit.read_connection_register_item("mario", "c-1") == PARKED_CONNECTION139 assert deposit.get_item_header("mario", "c-1") == {140 "writer": "standard_0001",141 "ts": pytest.approx(time.time(), abs=60),142 "cause": "freeze",143 "group": "standard",144 }145 146 # It gives the semaphore back: the operation is over, the parcel stays.147 assert await order(handler, RELEASE_LOCK_OP, {"user": "mario"}) == {"released": "mario"}148 assert deposit.lock_holder("mario") is None149 assert deposit.read_connection_register_item("mario", "c-1") == PARKED_CONNECTION150 151 # It takes the semaphore again — this is the operation the death interrupts.152 assert await order(handler, TAKE_LOCK_OP, {"user": "mario"}) == {"taken": True}153 assert deposit.lock_holder("mario") == "standard_0001"154 155 # It goes mute: up, and no longer serving. That order is the last thing it156 # answers. The beat is repeated once past the timeout and then the process157 # group is killed and its death awaited.158 condemned = handler.process159 handler.process_ping_timeout = 1.0160 assert await order(handler, GO_MUTE_OP) == {"muted": True}161 started = time.monotonic()162 await handler.ping_process()163 elapsed = time.monotonic() - started164 165 assert elapsed < 6 * handler.process_ping_timeout166 assert not condemned.alive167 assert handler.process is None168 169 # The death was nobody's order: the handler writes `aborted`, rings its170 # group's wake, and its job ends there — at that round the group reads the171 # state and the users that were on board.172 await wait_for(lambda: group.wakes == ["aborted"])173 assert handler.state == "aborted"174 assert group.users_on_board == [{"mario", "anna"}]175 assert handler.connector.connected is False176 assert handler.worker_snapshot["pid"] == condemned.pid177 178 # And the deposit is exactly as the dead process left it: the parcel it179 # parked, and the semaphore of the operation it never finished. Macro 1180 # cleans nothing — the traces are the Commander's to sweep.181 assert deposit.user_folders == {deposit.user_to_userkey("mario")}182 assert deposit.read_connection_register_item("mario", "c-1") == PARKED_CONNECTION183 assert deposit.lock_holder("mario") == "standard_0001"