tests/spa/orchestration/test_orchestration_console.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 debug door: one eval, any process of the pool, over the lane there is."""16 17 from __future__ import annotations18 19 import pytest20 21 from genro_asgi_multiworker_spa.spa_console import SpaConsole, SpaConsoleMcpApplication22 from genro_asgi_multiworker_spa.orchestration.spa_worker import GUEST_PREFIX23 24 25 async def test_the_commander_answers_about_itself(worker_commander_lane):26     worker_commander_lane.commander.record_connection_user("cid-a", "guest_legacy1")27 28     assert (29         await worker_commander_lane.commander.eval_in_target("commander", "len(commander.user_map)")30         == "1"31     )32 33 34 async def test_a_worker_answers_over_the_lane(worker_commander_lane):35     worker_commander_lane.worker.add_connection("a1b2")36     worker_commander_lane.worker.add_page("page-0", "a1b2")37 38     connections = await worker_commander_lane.commander.eval_in_target(39         "standard_0001", "len(worker.connection_register)"40     )41     pages = await worker_commander_lane.commander.eval_in_target(42         "standard_0001", "sorted(worker.page_register.keys())"43     )44 45     assert connections == "1"46     assert pages == "['page-0']"47 48 49 async def test_an_expression_that_fails_in_the_child_travels_back_as_the_error(50     worker_commander_lane,51 ):52     with pytest.raises(RuntimeError, match="NameError"):53         await worker_commander_lane.commander.eval_in_target("standard_0001", "no_such_name")54 55 56 async def test_an_unknown_target_names_the_ones_there_are(worker_commander_lane):57     with pytest.raises(KeyError, match="commander, standard_0001"):58         await worker_commander_lane.commander.eval_in_target("standard_9999", "1")59 60 61 class XT_Server:62     """The one attribute of a server the console reads: its applications."""63 64     def __init__(self, applications):65         self.applications = applications66 67 68 async def test_the_tools_reach_the_front_and_list_the_targets(worker_commander_lane):69     from genro_asgi_multiworker_spa.spa_app import SpaApplication70 71     spa_front = SpaApplication.__new__(SpaApplication)72     spa_front._commander = worker_commander_lane.commander73     console_app = SpaConsoleMcpApplication(code="console")74     console_app.server = XT_Server({"shop": spa_front, "other": object()})75     console = SpaConsole(console_app)76 77     worker_commander_lane.worker.add_connection("a1b2")78     guest = f"{GUEST_PREFIX}a1b2"79 80     assert await console.targets() == {"shop": ["commander", "standard_0001"]}81 82     answer = await console.eval("sorted(worker.user_register.keys())", "standard_0001")83     assert answer == {"target": "standard_0001", "repr": f"['{guest}']"}