tests/spa/orchestration/test_orchestration_placement.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 a user lands: the walk down from the hottest, and the four refusals.16 17 The subject here is the JUDGEMENT, not the processes: the workers are real18 ``WorkerHandler`` over a real group and a real vertex, but none of them has a19 process under it — what a placement reads is the ``state`` and the last photo,20 and both are written straight in, the way the machine writes them. The processes21 have their own tests, one file over.22 23 The occupancy of this group is measured against a round million of bytes — the24 whole concession is the group's quota and the whole quota is what one worker may25 hold — so a photo of 780_000 bytes is a worker standing at 78% and the arithmetic26 of every refusal is readable in the numbers themselves.27 """28 29 from __future__ import annotations30 31 import time as real_time32 33 import pytest34 35 from genro_asgi_multiworker_spa.orchestration import (36 AssignmentRefused,37 GroupHandler,38 NoRoomError,39 SpaCommander,40 WorkerHandler,41 WorkerQuittingError,42 )43 44 #: What one worker of these groups reads as full.45 MEMORY_CEILING = 1_000_00046 47 48 49 def minted(commander, cid: str) -> str:50 """The identity the site would baptise for this cookie, learned by the vertex.51 52 The old mint died with the doctrine (the cookie routes, the site names):53 tests stage the junction the fold of ``new_connection`` would have written.54 """55 user = f"guest_{cid}"56 commander.record_connection_user(cid, user)57 return user58 59 60 @pytest.fixture61 def commander(tmp_path):62 return SpaCommander(tmp_path / "frozen_users")63 64 65 @pytest.fixture66 def group(commander, tmp_path):67 """A group whose ONE worker may hold the whole quota: the arithmetic below68 reads occupancies against the quota itself (the core default sizes a group69 for ``worker_max_number`` workers, pinned away here on purpose)."""70 return GroupHandler(71 commander,72 "standard",73 memory_concession_bytes=MEMORY_CEILING,74 worker_memory_max_percent=100.0,75 instance_dir=tmp_path / "i",76 frozen_users_path=tmp_path / "frozen_users",77 entry_module="never.launched",78 )79 80 81 def worker_at(group, name: str, occupancy_percent: float, state: str = "running"):82 """One real worker of the group, standing at that occupancy, with no process under it."""83 worker_handler = WorkerHandler(group, name, **group.worker_settings)84 worker_handler.state = state85 worker_handler.worker_snapshot = {86 "rss_bytes": int(MEMORY_CEILING * occupancy_percent / 100)87 }88 group.worker_handler_map[name] = worker_handler89 return worker_handler90 91 92 def warm(worker_handler, cpu_temperature_percent: float) -> None:93 """Declare a fresh filtered temperature on a worker, as the meter would."""94 worker_handler.cpu_temperature_percent = cpu_temperature_percent95 worker_handler.cpu_temperature_sampled_at = real_time.monotonic()96 worker_handler.cpu_temperature_interval_seconds = 0.197 98 99 def newcomer(commander, cid: str = "cid-a") -> str:100 """A user the vertex has minted and nobody has ever measured."""101 return minted(commander, cid)102 103 104 async def test_a_photo_reads_as_the_percentage_it_is_and_never_over_full(group):105 assert group.get_memory_occupancy_percent(None) == 0.0106 assert group.get_memory_occupancy_percent({}) == 0.0107 assert group.get_memory_occupancy_percent({"rss_bytes": MEMORY_CEILING // 4}) == 25.0108 assert group.get_memory_occupancy_percent({"rss_bytes": 3 * MEMORY_CEILING}) == 100.0109 110 111 async def test_a_worker_reads_as_full_at_its_own_share_of_the_group_quota(commander, tmp_path):112 group = GroupHandler(113 commander,114 "standard",115 memory_concession_bytes=MEMORY_CEILING,116 memory_max_percent=50.0,117 worker_memory_max_percent=50.0,118 instance_dir=tmp_path / "i",119 frozen_users_path=tmp_path / "f",120 entry_module="never.launched",121 )122 123 # The cascade: half the concession is this group's quota, half of that quota124 # is what one of its workers may hold — so a quarter of the machine is a125 # worker of this group standing at its full.126 assert group.memory_quota_bytes == MEMORY_CEILING / 2127 assert group.get_memory_occupancy_percent({"rss_bytes": MEMORY_CEILING // 4}) == 100.0128 assert group.get_memory_occupancy_percent({"rss_bytes": MEMORY_CEILING // 8}) == 50.0129 130 131 async def test_the_hottest_worker_that_still_takes_him_is_the_one_that_gets_him(group, commander):132 warm(worker_at(group, "standard_0001", 10.0), 10.0)133 warm(worker_at(group, "standard_0002", 20.0), 20.0)134 warm(worker_at(group, "standard_0003", 60.0), 60.0)135 user = newcomer(commander)136 137 assert await group.assign_user(user) == "standard_0003"138 assert group.user_worker_map == {user: "standard_0003"}139 140 141 async def test_the_walk_goes_past_the_one_with_no_room_and_stops_at_the_next(group, commander):142 warm(worker_at(group, "standard_0001", 10.0), 10.0)143 warm(worker_at(group, "standard_0002", 50.0), 20.0)144 # The hottest is past the memory veto of 80: he does not fit, and the walk145 # goes on to the next one down the temperature.146 warm(worker_at(group, "standard_0003", 85.0), 60.0)147 user = newcomer(commander)148 149 assert await group.assign_user(user) == "standard_0002"150 151 152 async def test_two_placements_in_a_row_go_to_two_workers(group, commander):153 warm(worker_at(group, "standard_0001", 0.0), 5.0)154 warm(worker_at(group, "standard_0002", 70.0), 40.0)155 first = newcomer(commander, "cid-a")156 second = newcomer(commander, "cid-b")157 158 # The hotter one takes the first; having just admitted, it is skipped for159 # the admission interval and the second lands on the other.160 assert await group.assign_user(first) == "standard_0002"161 assert await group.assign_user(second) == "standard_0001"162 163 164 async def test_a_worker_with_no_room_refuses_with_the_class_that_says_so(group):165 worker_handler = worker_at(group, "standard_0002", 85.0)166 worker_at(group, "standard_0001", 0.0)167 168 with pytest.raises(NoRoomError, match="stands at 85.0% of memory"):169 worker_handler.assign_user("mario")170 171 172 async def test_a_worker_on_its_way_out_refuses_with_the_class_that_says_it_will_not(group):173 for state in ("quitting", "quitted", "aborted"):174 worker_handler = worker_at(group, f"standard_{state}", 0.0, state=state)175 176 with pytest.raises(WorkerQuittingError):177 worker_handler.assign_user("mario")178 179 180 async def test_a_worker_that_has_not_presented_itself_yet_takes_nobody(group):181 worker_handler = worker_at(group, "standard_0001", 0.0, state="starting")182 183 with pytest.raises(AssignmentRefused) as refusal:184 worker_handler.assign_user("mario")185 186 assert type(refusal.value) is AssignmentRefused187 188 189 async def test_a_worker_whose_process_has_ended_is_nobodys_candidate(group, commander):190 worker_at(group, "standard_0001", 0.0, state="quitted")191 user = newcomer(commander)192 193 with pytest.raises(AssignmentRefused):194 await group.assign_user(user)195 196 assert group.living_workers == []197 assert group.reception is None198 199 200 async def test_when_nobody_admits_the_wake_rings_and_the_base_rises(group, commander):201 worker_at(group, "standard_0001", 85.0)202 worker_at(group, "standard_0002", 85.0)203 user = newcomer(commander)204 assert group.ping_now_event.is_set() is False205 206 with pytest.raises(AssignmentRefused) as refusal:207 await group.assign_user(user)208 209 assert type(refusal.value) is AssignmentRefused210 assert refusal.value.user == user211 assert group.ping_now_event.is_set() is True212 assert group.user_worker_map == {}