tests/spa/orchestration/test_orchestration_observation.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 observation stream: register mutations pushed worker → commander → watcher."""16 17 from __future__ import annotations18 19 import asyncio20 21 22 from .conftest import wait_for23 24 25 async def test_nobody_watching_leaves_the_worker_silent(worker_commander_lane):26     assert worker_commander_lane.worker.observation_on is False27     assert worker_commander_lane.commander.observation_watched is False28 29     await worker_commander_lane.verb("add_connection", "a1b2")30 31     assert worker_commander_lane.worker.observation_on is False32 33 34 async def test_a_watcher_switches_the_worker_on_and_hears_a_birth(worker_commander_lane):35     queue: asyncio.Queue = asyncio.Queue()36     await worker_commander_lane.commander.subscribe_observation(queue)37     assert worker_commander_lane.worker.observation_on is True38 39     await worker_commander_lane.verb("add_connection", "a1b2")40     await wait_for(lambda: queue.qsize() >= 2)41     await asyncio.sleep(0)42 43     # The vertex reports the same births when it folds the announcement; what44     # this test listens for is the worker's own report.45     heard = [queue.get_nowait() for _ in range(queue.qsize())]46     events = {event["kind"]: event for event in heard if event["source"] == "standard_0001"}47     assert set(events) == {"new_user", "new_connection"}48     assert events["new_connection"]["data"]["user"].startswith("guest_")49 50 51 async def test_the_last_watcher_leaving_switches_the_worker_off(worker_commander_lane):52     queue: asyncio.Queue = asyncio.Queue()53     other: asyncio.Queue = asyncio.Queue()54     await worker_commander_lane.commander.subscribe_observation(queue)55     await worker_commander_lane.commander.subscribe_observation(other)56 57     await worker_commander_lane.commander.unsubscribe_observation(queue)58     assert worker_commander_lane.worker.observation_on is True59 60     await worker_commander_lane.commander.unsubscribe_observation(other)61     assert worker_commander_lane.worker.observation_on is False62     assert worker_commander_lane.commander.observation_watched is False63 64 65 async def test_the_fold_at_the_vertex_is_published_too(worker_commander_lane):66     queue: asyncio.Queue = asyncio.Queue()67     await worker_commander_lane.commander.subscribe_observation(queue)68 69     worker_commander_lane.worker_handler.read_envelope(70         {"worker_events": [{"op": "new_user", "worker": "standard_0001", "user": "u1"}]}71     )72 73     event = await queue.get()74     assert event["kind"] == "new_user"75     assert event["source"] == "commander"