From 9ae4b27158784c7abd4e4ade2d6e066123e61705 Mon Sep 17 00:00:00 2001 From: Sebastien Jourdain Date: Wed, 7 Oct 2026 16:40:33 -0600 Subject: [PATCH 01/19] feat(state): improve update() usability with kwarg support --- src/trame_server/state.py | 8 +++-- tests/test_state.py | 64 +++++++++++++++++++++++++++++++++++++++ 2 files changed, 70 insertions(+), 2 deletions(-) diff --git a/src/trame_server/state.py b/src/trame_server/state.py index 8f246c0..91b3b70 100644 --- a/src/trame_server/state.py +++ b/src/trame_server/state.py @@ -355,8 +355,12 @@ def clean(self, *_args): self._pushed_state[key] = self._pending_update.pop(key) self._suppress_change_stack.on_pending_key_removed(key) - def update(self, _dict): - """Update the current state dict with the provided one""" + def update(self, _dict=None, **inline_dict): + """ + Update the current state dict with the provided one either + using keyword arguments, a single dict argument or both. + """ + _dict = {**(_dict or {}), **inline_dict} _dict = self._translator.translate_dict(_dict) self._pending_update.update(_dict) for key in _dict: diff --git a/tests/test_state.py b/tests/test_state.py index aa4b6a3..94013d8 100644 --- a/tests/test_state.py +++ b/tests/test_state.py @@ -168,6 +168,70 @@ def test_dict_api(fake_server): assert state.initial == {"a": 1, "b": 20, "c": ["item"]} +def test_update_with_dict_or_kwargs(fake_server): + server = fake_server + state = fake_server.state + state.ready() + + # dict argument + state.update({"a": 1, "b": 2}) + assert state.a == 1 + assert state.b == 2 + assert state.is_dirty("a", "b") + state.flush() + assert server.pushed_state["a"] == 1 + assert server.pushed_state["b"] == 2 + + # keyword arguments + state.update(a=3, c=4) + assert state.a == 3 + assert state.b == 2 + assert state.c == 4 + assert state.is_dirty("a", "c") + assert not state.is_dirty("b") + state.flush() + assert server.pushed_state["a"] == 3 + assert server.pushed_state["c"] == 4 + + # dict and keyword arguments are merged (keywords win) + state.update({"a": 10, "d": 6}, a=11, e=7) + assert state.a == 11 + assert state.d == 6 + assert state.e == 7 + state.flush() + assert server.pushed_state["a"] == 11 + + # no argument is a no-op + state.update() + assert state.to_dict() == {"a": 11, "b": 2, "c": 4, "d": 6, "e": 7} + + # keyword arguments trigger change listeners + mock = MagicMock() + + @state.change("a") + def on_change(a, **_): + mock(a) + + with state: + state.update(a=5) + + mock.assert_called_once_with(5) + + +def test_update_with_kwargs_on_namespaced_state(fake_server): + server = fake_server + translator = Translator(prefix="test_") + state = State(translator=translator, commit_fn=server._push_state) + state.ready() + + state.update(a=1, b=2) + assert state.a == 1 + assert state.b == 2 + state.flush() + assert server.pushed_state["test_a"] == 1 + assert server.pushed_state["test_b"] == 2 + + @pytest.mark.asyncio async def test_change_detection(fake_server): """ From 974214ee855566c00139034eaec368155683695f Mon Sep 17 00:00:00 2001 From: Sebastien Jourdain Date: Wed, 7 Oct 2026 17:06:01 -0600 Subject: [PATCH 02/19] test: add tests to improve coverage --- src/trame_server/client.py | 6 +- tests/test_async.py | 63 +++++++ tests/test_controller.py | 84 +++++++++ tests/test_protocol.py | 171 +++++++++++++++++++ tests/test_typed_state.py | 63 +++++++ tests/test_utils_extra.py | 341 +++++++++++++++++++++++++++++++++++++ 6 files changed, 724 insertions(+), 4 deletions(-) create mode 100644 tests/test_protocol.py create mode 100644 tests/test_utils_extra.py diff --git a/src/trame_server/client.py b/src/trame_server/client.py index 63ea64e..8ed4c88 100644 --- a/src/trame_server/client.py +++ b/src/trame_server/client.py @@ -43,13 +43,11 @@ async def on_msg_complete(self, payload): # Error if "error" in payload: if future: - future.set_exception( - payload.get("error", "Server error") - ) # May need to wrap in Exception? + future.set_exception(Exception(payload.get("error", "Server error"))) else: print("Server error:", payload.get("error")) - self.in_flight_rpc.pop(msg_id) + self.in_flight_rpc.pop(msg_id, None) return # Normal processing diff --git a/tests/test_async.py b/tests/test_async.py index 5e79656..608e27c 100644 --- a/tests/test_async.py +++ b/tests/test_async.py @@ -90,3 +90,66 @@ async def run_something(): assert bg_update == "idle" await asyncio.sleep(0.1) assert bg_update == "ok" + + +@pytest.mark.asyncio +async def test_handle_task_result_logs_exception(caplog): + from trame_server.utils import asynchronous # noqa: PLC0415 + + async def failing(): + msg = "task failure" + raise ValueError(msg) + + async def forever(): + await asyncio.sleep(10) + + task = asynchronous.create_task(failing()) + await asyncio.sleep(0.01) + assert "Exception raised by task" in caplog.text + + # Cancellation is silent + caplog.clear() + task = asynchronous.create_task(forever()) + await asyncio.sleep(0) + task.cancel() + await asyncio.sleep(0.01) + assert caplog.text == "" + + +@pytest.mark.asyncio +async def test_task_decorator_schedules_coroutine(): + from trame_server.utils import asynchronous # noqa: PLC0415 + + results = [] + + @asynchronous.task + async def run(value): + results.append(value) + + run(5) + await asyncio.sleep(0.01) + assert results == [5] + + +def test_state_queue_without_auto_flush(): + import queue as std_queue # noqa: PLC0415 + + from trame_server.utils.asynchronous import QUEUE_EXIT, StateQueue # noqa: PLC0415 + + q = std_queue.Queue() + state = StateQueue(q, auto_flush=False) + assert state._unknown_private is None + with pytest.raises(AttributeError): + _ = state.__not_existing__ + + state.a = 1 + state.update({"b": 2}) + assert state.a == 1 + assert q.empty() + + state.flush() + assert q.get_nowait() == {"a": 1, "b": 2} + assert state.b == 2 + + state.exit() + assert q.get_nowait() == QUEUE_EXIT diff --git a/tests/test_controller.py b/tests/test_controller.py index c7c47fa..925be32 100644 --- a/tests/test_controller.py +++ b/tests/test_controller.py @@ -184,3 +184,87 @@ def fn_3(): assert child_controller.func() == [3, 2, 2.5] assert child_controller.func() == [3, 2] assert child_controller.func() == controller.child_func() + + +def test_trigger_on_method_and_unregister(): + from trame_server.controller import Controller # noqa: PLC0415 + + controller = Controller() + + class Handler: + def on_click(self): + return "clicked" + + def on_other(self): + return "other" + + handler = Handler() + controller.trigger("click")(handler.on_click) + controller.trigger("other")(handler.on_other) + assert Handler._trame_trigger_method_names == ["on_click", "on_other"] + assert controller.trigger_fn("click")() == "clicked" + + # Unregister by function returns its name + assert controller.trigger_unregister(handler.on_click) == "click" + assert controller.trigger_fn("click") is None + + # Unregister by name returns the function + assert controller.trigger_unregister("other") == handler.on_other + assert controller.trigger_fn("other") is None + + assert controller.trigger_unregister("missing") is False + + +def test_controller_function_edge_cases(): + from trame_server.controller import Controller # noqa: PLC0415 + + controller = Controller() + + with pytest.raises(AttributeError): + _ = controller.__not_existing__ + + # Empty function allowed + controller.maybe.enable_empty() + assert controller.maybe() is None + + # Discard main function + def main(): + return 1 + + controller.main = main + assert controller.main() == 1 + controller.main.discard(main) + assert not controller.main.exists() + + +@pytest.mark.asyncio +async def test_controller_main_with_task_only(): + from trame_server.controller import Controller # noqa: PLC0415 + + controller = Controller() + + async def task_fn(): + return "task" + + controller.fn = lambda: "main" + controller.fn.add_task(task_fn) + result = controller.fn() + assert len(result) == 1 + assert await result[0] == "task" + + +def test_controller_hot_reload(): + from trame_server.controller import Controller # noqa: PLC0415 + + controller = Controller(hot_reload=True) + assert controller.fn.hot_reload + + def hr_main_fn(x): + return x + 1 + + def hr_added_fn(x): + return x * 2 + + controller.fn = hr_main_fn + controller.fn.add(hr_added_fn) + assert controller.fn(3) == [4, 6] diff --git a/tests/test_protocol.py b/tests/test_protocol.py new file mode 100644 index 0000000..077dad0 --- /dev/null +++ b/tests/test_protocol.py @@ -0,0 +1,171 @@ +import asyncio +import io +from argparse import Namespace +from contextlib import redirect_stdout +from unittest.mock import MagicMock + +import pytest +import pytest_asyncio +from trame.app import asynchronous, get_client, get_server + +from trame_server.protocol import CoreServer + + +@pytest_asyncio.fixture +async def server(): + server = get_server("test_protocol") + server.start(exec_mode="task", port=0) + assert await server.ready + try: + yield server + finally: + await asyncio.sleep(0.2) + await server.stop() + + +@pytest_asyncio.fixture +async def client(server): + client = get_client(f"ws://localhost:{server.port}/ws") + asynchronous.create_task(client.connect(secret="wslink-secret")) + + for _ in range(20): + if client._session and client._session.client_id is not None: + break + await asyncio.sleep(0.1) + + assert client.connected == 2 + try: + yield client + finally: + await asyncio.sleep(0.1) + await client.disconnect() + + +async def _rpc(client, method, *args): + return await (await client._session.call(method, list(args))) + + +@pytest.mark.asyncio +async def test_rpc_error_keeps_client_alive(server, client): + with pytest.raises(Exception, match="Unregistered method called"): + await _rpc(client, "trame.does.not.exist") + + # The connection must still be usable after an error + @server.trigger("ping") + def ping(): + return "pong" + + assert await client.call_trigger("ping") == "pong" + + +@pytest.mark.asyncio +async def test_async_and_missing_trigger(server, client): + @server.trigger("async_add") + async def async_add(a, b): + await asyncio.sleep(0.01) + return a + b + + assert await client.call_trigger("async_add", [1, 2]) == 3 + + with io.StringIO() as buf, redirect_stdout(buf): + assert await client.call_trigger("not_registered") is None + assert "Trigger not_registered seems to be missing" in buf.getvalue() + + +@pytest.mark.asyncio +async def test_lifecycle_and_js_error(server, client): + on_custom = MagicMock() + server.controller.on_custom = on_custom + await _rpc(client, "trame.lifecycle.update", "custom") + on_custom.assert_called_once_with() + + # Unknown life cycle name is a no-op + await _rpc(client, "trame.lifecycle.update", "unknown_cycle") + + # JS error without handler is printed + with io.StringIO() as buf, redirect_stdout(buf): + await _rpc(client, "trame.error.client", "boom") + assert "JS Error => boom" in buf.getvalue() + + # JS error with handler + on_error = MagicMock() + server.controller.on_error = on_error + await _rpc(client, "trame.error.client", "boom again") + on_error.assert_called_once_with("boom again") + + +@pytest.mark.asyncio +async def test_get_state_and_force_push(server, client): + server.state.forced = 1 + server.state.flush() + await server.network_completion + await asyncio.sleep(0.1) + assert client.state.forced == 1 + + full_state = await _rpc(client, "trame.state.get") + assert full_state["state"]["forced"] == 1 + + # Client corrupts its local copy, server forces a resend + client.state._pushed_state["forced"] = 0 + server.force_state_push("forced") + await server.network_completion + await asyncio.sleep(0.1) + assert client.state.forced == 1 + + # No keys is a no-op + server.force_state_push() + + +@pytest.mark.asyncio +async def test_protocol_call(server): + assert server.protocol_call("trame.does.not.exist") is None + assert server.protocol_call("trame.state.get")["state"] is not None + + +@pytest.mark.asyncio +async def test_client_subscriptions(client): + session = client._session + received = [] + + def broken(_): + msg = "callback failure" + raise ValueError(msg) + + session.register_subscription("custom.topic", received.append) + session.register_subscription("custom.topic", broken) + + with io.StringIO() as buf, redirect_stdout(buf): + await session.on_msg_complete( + {"id": "publish:custom.topic:0", "result": {"x": 1}} + ) + assert "Subscription callback error" in buf.getvalue() + assert received == [{"x": 1}] + + # Error without matching in-flight rpc + with io.StringIO() as buf, redirect_stdout(buf): + await session.on_msg_complete({"id": "rpc:x:999", "error": "oops"}) + assert "Server error: oops" in buf.getvalue() + + # Notification without id is ignored + await session.on_msg_complete({"result": None}) + + session.unregister_subscription("custom.topic", broken) + assert session.subscriptions["custom.topic"] == [received.append] + session.unregister_subscription("custom.topic", received.append) + assert "custom.topic" not in session.subscriptions + # Unknown topic/callback is a no-op + session.unregister_subscription("custom.topic", received.append) + + +def test_configure_auth_key(tmp_path): + previous = CoreServer.authentication_token + try: + CoreServer.configure(Namespace(authKeyFile=None, authKey="from-arg")) + assert CoreServer.authentication_token == "from-arg" + + key_file = tmp_path / "key.txt" + key_file.write_text("from-file\n") + CoreServer.configure(Namespace(authKeyFile=str(key_file), authKey="unused")) + assert CoreServer.authentication_token == "from-file" + finally: + CoreServer.authentication_token = previous diff --git a/tests/test_typed_state.py b/tests/test_typed_state.py index 0b6e71d..01ffb7d 100644 --- a/tests/test_typed_state.py +++ b/tests/test_typed_state.py @@ -745,3 +745,66 @@ def test_benchmark_typed_state_serialization( def test_benchmark_typed_state_deserialization(benchmark, benchmark_typed_state): benchmark(benchmark_typed_state.get_dataclass) + + +def test_serialization_failure_equality(): + failure_a = IStateEncoderDecoder.failed_serialization("a") + failure_b = IStateEncoderDecoder.failed_serialization("b") + assert failure_a == failure_b + assert failure_a != "a" + assert hash(failure_a) == hash("a") + assert not IStateEncoderDecoder.is_serialization_success(failure_a) + + +def test_default_encoder_decimal(): + from decimal import Decimal # noqa: PLC0415 + + encoder = DefaultEncoderDecoder() + assert encoder.encode(Decimal("1.5")) == "1.5" + + +def test_collection_decoder_failures(): + decoder = CollectionEncoderDecoder() + + # Invalid inputs for each specialized decoder + assert not decoder.is_serialization_success( + decoder._decode_dict([1], dict[str, int]) + ) + assert not decoder.is_serialization_success(decoder._decode_iterable(1, list[int])) + assert not decoder.is_serialization_success(decoder._decode_union(1, int)) + assert not decoder.is_serialization_success(decoder._decode_dataclass(1, MyData)) + assert not decoder.is_serialization_success(decoder._decode_dataclass({}, int)) + + # Unknown dataclass type discriminator + _type_key = CollectionEncoderDecoder._DATACLASS_TYPE_KEY + assert decoder._get_dataclass_type({_type_key: "unknown.Type"}, MyData) is None + + # Dict that isn't a dataclass falls back to union / delegate decoding + assert decoder.decode({"a": 1}, dict | int) == {"a": 1} + + +def test_class_level_helpers_validate_inputs(state): + typed_state = TypedState(state, MyData) + + with pytest.raises(RuntimeError, match="Expected an instance of type __Proxy"): + TypedState.as_dataclass(MyData()) + + with pytest.raises(TypeError, match="Expected instance of MyData"): + TypedState.from_dataclass(typed_state.data, MyBiggerData()) + + with pytest.raises(RuntimeError, match="Sub state creation"): + typed_state.get_sub_state("not a proxy") + + +def test_can_bind_changes_with_list_of_keys(state): + typed_state = TypedState(state, MyData) + mock = MagicMock() + typed_state.bind_changes({(typed_state.name.a, typed_state.name.b): mock}) + TypedState.bind_typed_state_change( + [typed_state.name.a, typed_state.name.b], mock, state, typed_state.data + ) + + typed_state.data.a = 5 + state.flush() + assert mock.call_count == 2 + mock.assert_called_with(5, 2) diff --git a/tests/test_utils_extra.py b/tests/test_utils_extra.py new file mode 100644 index 0000000..8a5467b --- /dev/null +++ b/tests/test_utils_extra.py @@ -0,0 +1,341 @@ +import asyncio +import importlib +import io +import socket +import sys +from argparse import Namespace +from contextlib import redirect_stderr, redirect_stdout +from types import SimpleNamespace +from unittest.mock import MagicMock + +import more_itertools +import pytest + +from trame_server.ui import VirtualNodeManager +from trame_server.utils import browser, hot_reload, logger, server +from trame_server.utils.argument_parser import ArgumentParser + +# ----------------------------------------------------------------------------- +# logger +# ----------------------------------------------------------------------------- + + +@pytest.fixture +def network_log(tmp_path): + log_file = tmp_path / "network.log" + log_file.write_text("previous run") + logger.initialize_logger({"log_network": str(log_file)}) + try: + yield log_file + finally: + logger.initialize_logger({}) + + +def test_logger_disabled(tmp_path, capsys): + logger.initialize_logger({}) + logger.state_c2s({"a": 1}) + logger.error("nothing written") + assert "Error: nothing written" in capsys.readouterr().out + assert list(tmp_path.iterdir()) == [] + + +def test_logger_enabled(network_log, capsys): + # Previous log removed at initialization + assert not network_log.exists() + + logger.initial_state({"init": 1}) + logger.state_c2s({"c2s": 2}) + logger.state_s2c({"s2c": b"binary"}) + logger.action_c2s({"name": "click"}) + logger.action_s2c({"type": "action"}) + logger.error("something broke") + + content = network_log.read_text() + assert logger.StateExchangeType.STATE_INITIAL in content + assert logger.StateExchangeType.STATE_CLIENT_TO_SERVER in content + assert logger.StateExchangeType.STATE_SERVER_TO_CLIENT in content + assert logger.StateExchangeType.ACTION_CLIENT_TO_SERVER in content + assert logger.StateExchangeType.ACTION_SERVER_TO_CLIENT in content + assert "" in content + assert "ERROR: something broke" in content + assert "Error: something broke" in capsys.readouterr().out + + with pytest.raises(TypeError): + logger.state_c2s({"not_serializable": object()}) + + +# ----------------------------------------------------------------------------- +# argument_parser +# ----------------------------------------------------------------------------- + + +def test_argument_parser_env(monkeypatch): + monkeypatch.setattr(sys, "argv", ["app.py", "--ignored"]) + monkeypatch.setenv("TRAME_ARGS", "-p 8081 --server") + parser = ArgumentParser() + parser.add_argument("-p", "--port", type=int) + parser.add_argument("--server", action="store_true") + args, unknown = parser.parse_known_args() + assert args.port == 8081 + assert args.server + assert unknown == [] + + +@pytest.mark.parametrize( + "argv", + [ + ["app.py", "--trame-args=-p 8082 --server"], + ["app.py", "--", "--trame-args", "-p 8082 --server"], + ], +) +def test_argument_parser_trame_args(monkeypatch, argv): + monkeypatch.delenv("TRAME_ARGS", raising=False) + monkeypatch.setattr(sys, "argv", argv) + parser = ArgumentParser() + parser.add_argument("-p", "--port", type=int) + parser.add_argument("--server", action="store_true") + args, _ = parser.parse_known_args() + assert args.port == 8082 + assert args.server + + +# ----------------------------------------------------------------------------- +# browser / server information +# ----------------------------------------------------------------------------- + + +def _fake_server(host="localhost", port=1234): + cli = ArgumentParser() + cli.add_argument("--host", default=host) + return SimpleNamespace( + cli=cli, + port=port, + server_options=Namespace(host=host), + ) + + +@pytest.mark.asyncio +async def test_open_browser(monkeypatch): + import webbrowser # noqa: PLC0415 + + open_mock = MagicMock() + monkeypatch.setattr(webbrowser, "open", open_mock) + + with io.StringIO() as buf, redirect_stdout(buf): + browser.open_browser(_fake_server()) + assert "--server" in buf.getvalue() + + await asyncio.sleep(0.2) + open_mock.assert_called_once_with("http://localhost:1234/") + + +def test_open_browser_error(monkeypatch): + def fail(): + raise RuntimeError + + monkeypatch.setattr(asyncio, "get_event_loop", fail) + # Errors are silently ignored + browser.open_browser(_fake_server()) + + +def test_print_informations_host_fallback(monkeypatch): + calls = [] + + def gethostbyname(name): + calls.append(name) + if name == "unknown-host": + raise OSError + return "10.0.0.1" + + monkeypatch.setattr(socket, "gethostbyname", gethostbyname) + monkeypatch.setattr(socket, "gethostname", lambda: "my-machine") + + with io.StringIO() as buf, redirect_stdout(buf): + server.print_informations(_fake_server(host="unknown-host")) + output = buf.getvalue() + + assert calls == ["unknown-host", "my-machine"] + assert "Network: http://10.0.0.1:1234/" in output + + +def test_print_informations_no_network(monkeypatch): + def gethostbyname(_): + raise socket.gaierror + + monkeypatch.setattr(socket, "gethostbyname", gethostbyname) + + with io.StringIO() as buf, redirect_stdout(buf): + server.print_informations(_fake_server(host="nowhere")) + assert "Network: http://nowhere:1234/" in buf.getvalue() + + +# ----------------------------------------------------------------------------- +# ui +# ----------------------------------------------------------------------------- + + +def test_virtual_node_manager(): + manager = VirtualNodeManager("fake_server") + with pytest.raises(AttributeError): + _ = manager.__not_existing__ + + created = [] + + def constructor(trame_server): + created.append(trame_server) + return MagicMock() + + manager.set_vn_constructor(constructor) + assert manager.content is manager["content"] + assert created == ["fake_server"] + + +# ----------------------------------------------------------------------------- +# hot_reload +# ----------------------------------------------------------------------------- + +MODULE_TEMPLATE = """ +def ctrl(*_, **__): + return lambda f: f + +class Obj: + def attr(self, *_): + return lambda f: f + +obj = Obj() + +def extra(f): + f.extra = True + return f + +@obj.attr("x") +@extra +@ctrl("y") +def compute(x): + return x * {factor} + +class Holder: + def method(self): + return {factor} +""" + + +@pytest.fixture +def reloadable_module(tmp_path, monkeypatch): + path = tmp_path / "hr_sample_module.py" + + def write(factor): + path.write_text(MODULE_TEMPLATE.format(factor=factor)) + + write(1) + monkeypatch.syspath_prepend(str(tmp_path)) + module = importlib.import_module("hr_sample_module") + try: + yield module, path, write + finally: + sys.modules.pop("hr_sample_module", None) + + +def test_reload_function_and_method(reloadable_module): + module, _, write = reloadable_module + holder = module.Holder() + + assert module.compute(3) == 3 + assert holder.method() == 1 + + write(10) + + reloaded = hot_reload.reload(module.compute) + assert reloaded(3) == 30 + assert reloaded.extra + reloaded_method = hot_reload.reload(holder.method) + assert reloaded_method.__self__ is holder + assert reloaded_method() == 10 + + +def test_reload_recovers_from_errors(reloadable_module, monkeypatch): + module, path, write = reloadable_module + holder = module.Holder() + path.write_text("def broken(:\n") + + attempts = [] + + def fix_file(func): + attempts.append(func) + if len(attempts) == 1: + # Syntax valid, but function is missing + path.write_text("def something_else():\n pass\n") + else: + write(5) + + monkeypatch.setattr(hot_reload, "_handle_exception", fix_file) + assert hot_reload.reload(holder.method)() == 5 + assert len(attempts) == 2 + + +def test_reload_skips(): + assert hot_reload.reload(42) == 42 + + fn = lambda: 1 # noqa: E731 + assert hot_reload.reload(fn) is fn + + # Function living in site-packages + assert hot_reload.reload(more_itertools.first) is more_itertools.first + + # Methods of trame elements are skipped + from trame_client.widgets.core import AbstractElement # noqa: PLC0415 + + class CustomElement(AbstractElement): + def method(self): + pass + + element = object.__new__(CustomElement) + method = element.method + assert hot_reload.reload(method) is method + + +def test_reload_nested_function_closure(): + captured_value = 7 + + def nested_with_closure(): + return captured_value + + reloaded = hot_reload.reload(nested_with_closure) + assert reloaded is not nested_with_closure + assert reloaded() == 7 + + +def test_hot_reload_decorator_stripping(): + import ast # noqa: PLC0415 + + tree = ast.parse( + "@outer\n@hot_reload\n@ctrl.add('x')\n@keep\ndef fn():\n pass\n" + ) + assert hot_reload._isolate_function_def("fn", tree) + names = [hot_reload._get_decorator_name(d) for d in tree.body[0].decorator_list] + assert names == ["keep"] + + assert not hot_reload._isolate_function_def("missing", ast.parse("x = 1")) + + with pytest.raises(Exception, match="Failed to find decorator name"): + hot_reload._get_decorator_name(ast.parse("(lambda f: f)").body[0].value) + + attr_dec = ast.parse("@state.change\ndef f(): pass").body[0].decorator_list[0] + assert hot_reload._get_decorator_name(attr_dec) == "state" + + +def test_handle_exception(monkeypatch): + monkeypatch.setattr(sys, "stdin", io.StringIO("\n")) + + def failing(): + pass + + try: + msg = "expected" + raise ValueError(msg) + except ValueError: + out, err = io.StringIO(), io.StringIO() + with redirect_stdout(out), redirect_stderr(err): + hot_reload._handle_exception(failing) + assert "ValueError: expected" in err.getvalue() + assert "press return to continue" in out.getvalue() From 08e2abc49c2aaddd579f905714aa74b3529aeb08 Mon Sep 17 00:00:00 2001 From: Sebastien Jourdain Date: Wed, 7 Oct 2026 17:09:17 -0600 Subject: [PATCH 03/19] fix: remove dead code due to py evolving --- src/trame_server/utils/version.py | 15 +-------------- 1 file changed, 1 insertion(+), 14 deletions(-) diff --git a/src/trame_server/utils/version.py b/src/trame_server/utils/version.py index 6635511..1a237d4 100644 --- a/src/trame_server/utils/version.py +++ b/src/trame_server/utils/version.py @@ -1,17 +1,4 @@ -try: - # Use importlib metadata if available (python >=3.8) - from importlib.metadata import PackageNotFoundError, version -except ImportError: - # No importlib metadata. Try to use pkg_resources instead. - from pkg_resources import ( - DistributionNotFound as PackageNotFoundError, - ) - from pkg_resources import ( - get_distribution, - ) - - def version(x): - return get_distribution(x).version +from importlib.metadata import PackageNotFoundError, version def get_version(package_name): From 2fbded30c1e9428085c9c49517df61977bc09504 Mon Sep 17 00:00:00 2001 From: Sebastien Jourdain Date: Wed, 7 Oct 2026 17:14:07 -0600 Subject: [PATCH 04/19] fix(msg_pack): properly handle serialization error --- src/trame_server/client.py | 16 ++++------------ tests/test_protocol.py | 21 +++++++++++++++++++++ 2 files changed, 25 insertions(+), 12 deletions(-) diff --git a/src/trame_server/client.py b/src/trame_server/client.py index 8ed4c88..c9573c7 100644 --- a/src/trame_server/client.py +++ b/src/trame_server/client.py @@ -108,13 +108,9 @@ async def auth(self, **kwargs): "args": [kwargs], "kwargs": {}, } - self.in_flight_rpc[key] = resp - try: - packed_wrapper = msgpack.packb(wrapper) - except Exception: - del wrapper["error"]["data"] - packed_wrapper = msgpack.packb(wrapper) + packed_wrapper = msgpack.packb(wrapper) + self.in_flight_rpc[key] = resp async with self.attachment_atomic: for chunk in generate_chunks(packed_wrapper, MAX_MSG_SIZE): @@ -127,7 +123,6 @@ async def call(self, method, args=None, kwargs=None): self.msg_count += 1 key = f"rpc:{self.client_id}:{self.msg_count}" resp = self.loop.create_future() - self.in_flight_rpc[key] = resp if args is None: args = [] if kwargs is None: @@ -141,11 +136,8 @@ async def call(self, method, args=None, kwargs=None): "kwargs": kwargs, } - try: - packed_wrapper = msgpack.packb(wrapper) - except Exception: - del wrapper["error"]["data"] - packed_wrapper = msgpack.packb(wrapper) + packed_wrapper = msgpack.packb(wrapper) + self.in_flight_rpc[key] = resp async with self.attachment_atomic: for chunk in generate_chunks(packed_wrapper, MAX_MSG_SIZE): diff --git a/tests/test_protocol.py b/tests/test_protocol.py index 077dad0..3368db7 100644 --- a/tests/test_protocol.py +++ b/tests/test_protocol.py @@ -58,6 +58,27 @@ def ping(): assert await client.call_trigger("ping") == "pong" +@pytest.mark.asyncio +async def test_unserializable_args_raise_serialization_error(server, client): + @server.trigger("echo") + def echo(value): + return value + + pending = dict(client._session.in_flight_rpc) + + with pytest.raises(TypeError, match="can not serialize"): + await client.call_trigger("echo", [object()]) + + with pytest.raises(TypeError, match="can not serialize"): + await client._session.auth(secret=object()) + + # Failed requests are not left waiting for a response + assert client._session.in_flight_rpc == pending + + # Connection is still usable + assert await client.call_trigger("echo", [1]) == 1 + + @pytest.mark.asyncio async def test_async_and_missing_trigger(server, client): @server.trigger("async_add") From 858f0108b49650a3c0e3470346e40b3d18453c6d Mon Sep 17 00:00:00 2001 From: Sebastien Jourdain Date: Wed, 7 Oct 2026 17:22:40 -0600 Subject: [PATCH 05/19] fix(hot_reload): better handle non interactive env --- src/trame_server/utils/hot_reload.py | 45 ++++++++++------ tests/test_utils_extra.py | 79 +++++++++++++++++++++++++--- 2 files changed, 102 insertions(+), 22 deletions(-) diff --git a/src/trame_server/utils/hot_reload.py b/src/trame_server/utils/hot_reload.py index 2f52a9e..bdffb39 100644 --- a/src/trame_server/utils/hot_reload.py +++ b/src/trame_server/utils/hot_reload.py @@ -36,6 +36,7 @@ import inspect import site import sys +import time import traceback import types from pathlib import Path @@ -63,6 +64,11 @@ # in editable environments. SKIP_SITE_PACKAGES = True +# While saving, editors may briefly leave the file empty. +# Wait up to EMPTY_FILE_RETRIES * EMPTY_FILE_RETRY_DELAY seconds for content. +EMPTY_FILE_RETRIES = 20 +EMPTY_FILE_RETRY_DELAY = 0.05 + def hot_reload(func): """Decorator to reload the function on every call @@ -95,6 +101,10 @@ def reload(func, perform_checks=True): If perform_checks is True, then several checks will be performed beforehand to determine whether or not the function should be skipped. + + If reloading fails, the user is asked to fix the code and press return + to try again. When stdin is not interactive (or reaches EOF), the error + is reported and the original function is returned instead. """ if perform_checks: if not isinstance(func, (types.FunctionType, types.MethodType)): @@ -118,7 +128,8 @@ def reload(func, perform_checks=True): try: return _reload_func(func) except Exception: - _handle_exception(func) + if not _handle_exception(func): + return func def _reload_func(func): @@ -167,7 +178,7 @@ def _find_function_locals(func): def _recompile_function(func): - tree = _parse_func_file_until_successful(func) + tree = ast.parse(_load_file(inspect.getfile(func))) if not _isolate_function_def(func.__name__, tree): path = inspect.getfile(func) msg = f"Failed to find '{func.__qualname__}' in file '{path}'" @@ -176,20 +187,12 @@ def _recompile_function(func): return compile(tree, filename="", mode="exec") -def _parse_func_file_until_successful(func): - path = inspect.getfile(func) - while True: - source = _load_file(path) - try: - return ast.parse(source) - except SyntaxError: - _handle_exception(func) - - def _load_file(path): - src = "" - # while loop here since while saving, the file may sometimes be empty. - while src == "": + src = Path(path).read_text() + for _ in range(EMPTY_FILE_RETRIES): + if src: + break + time.sleep(EMPTY_FILE_RETRY_DELAY) src = Path(path).read_text() return src + "\n" @@ -214,12 +217,22 @@ def _isolate_function_def(funcname, tree): def _handle_exception(func): + """Report the current exception and wait for the user to fix the code. + + Return True if the reload should be attempted again, False otherwise. + """ fpath = inspect.getfile(func) exc = traceback.format_exc() exc = exc.replace('File ""', f'File "{fpath}"') sys.stderr.write(exc + "\n") + + if sys.stdin is None or not sys.stdin.isatty(): + print(f"Hot reload of '{func.__qualname__}' failed, keeping previous version") + return False + print(f"Edit '{func.__qualname__}' in '{fpath}' and press return to continue") - sys.stdin.readline() + # readline() returns an empty string on EOF (e.g. Ctrl-D) + return sys.stdin.readline() != "" def _get_decorator_name(dec_node): diff --git a/tests/test_utils_extra.py b/tests/test_utils_extra.py index 8a5467b..9675e52 100644 --- a/tests/test_utils_extra.py +++ b/tests/test_utils_extra.py @@ -267,6 +267,7 @@ def fix_file(func): path.write_text("def something_else():\n pass\n") else: write(5) + return True monkeypatch.setattr(hot_reload, "_handle_exception", fix_file) assert hot_reload.reload(holder.method)() == 5 @@ -324,18 +325,84 @@ def test_hot_reload_decorator_stripping(): assert hot_reload._get_decorator_name(attr_dec) == "state" -def test_handle_exception(monkeypatch): - monkeypatch.setattr(sys, "stdin", io.StringIO("\n")) +class InteractiveInput(io.StringIO): + def isatty(self): + return True + + +def _handle_exception_output(monkeypatch, stdin): + monkeypatch.setattr(sys, "stdin", stdin) def failing(): pass + out, err = io.StringIO(), io.StringIO() try: msg = "expected" raise ValueError(msg) except ValueError: - out, err = io.StringIO(), io.StringIO() with redirect_stdout(out), redirect_stderr(err): - hot_reload._handle_exception(failing) - assert "ValueError: expected" in err.getvalue() - assert "press return to continue" in out.getvalue() + retry = hot_reload._handle_exception(failing) + + assert "ValueError: expected" in err.getvalue() + return retry, out.getvalue() + + +def test_handle_exception_interactive(monkeypatch): + retry, output = _handle_exception_output(monkeypatch, InteractiveInput("\n")) + assert retry + assert "press return to continue" in output + + # EOF (e.g. Ctrl-D) gives up + retry, _ = _handle_exception_output(monkeypatch, InteractiveInput("")) + assert not retry + + +@pytest.mark.parametrize("stdin", [None, io.StringIO("\n")]) +def test_handle_exception_non_interactive(monkeypatch, stdin): + retry, output = _handle_exception_output(monkeypatch, stdin) + assert not retry + assert "keeping previous version" in output + assert "press return" not in output + + +def test_reload_keeps_function_when_non_interactive(reloadable_module, monkeypatch): + module, path, _ = reloadable_module + holder = module.Holder() + path.write_text("def broken(:\n") + monkeypatch.setattr(sys, "stdin", io.StringIO("")) + + out, err = io.StringIO(), io.StringIO() + with redirect_stdout(out), redirect_stderr(err): + reloaded = hot_reload.reload(holder.method) + + assert reloaded == holder.method + assert reloaded() == 1 + assert "SyntaxError" in err.getvalue() + # Reported once, not in a loop + assert out.getvalue().count("keeping previous version") == 1 + + +def test_load_file_waits_for_content(tmp_path, monkeypatch): + path = tmp_path / "saving.py" + path.write_text("") + sleeps = [] + + def fake_sleep(delay): + sleeps.append(delay) + if len(sleeps) == 3: + path.write_text("x = 1") + + monkeypatch.setattr(hot_reload.time, "sleep", fake_sleep) + assert hot_reload._load_file(path) == "x = 1\n" + assert len(sleeps) == 3 + + +def test_load_file_gives_up_on_empty_file(tmp_path, monkeypatch): + path = tmp_path / "empty.py" + path.write_text("") + sleeps = [] + monkeypatch.setattr(hot_reload.time, "sleep", sleeps.append) + + assert hot_reload._load_file(path) == "\n" + assert len(sleeps) == hot_reload.EMPTY_FILE_RETRIES From f2ecb93ecd4ff02db0c65da3f4877777688b56cf Mon Sep 17 00:00:00 2001 From: Sebastien Jourdain Date: Wed, 7 Oct 2026 17:30:59 -0600 Subject: [PATCH 06/19] fix(asynchronous): get event loop safely on python 3.14 asyncio.get_event_loop() warns on 3.12 and raises on 3.14 when no loop is running, which broke create_task() before the server starts. Add asynchronous.get_event_loop() mirroring wslink's loop selection and use it in create_task, open_browser and the server exit path. Co-Authored-By: Claude Opus 5.5 --- src/trame_server/core.py | 4 +-- src/trame_server/utils/asynchronous.py | 32 +++++++++++++++++- src/trame_server/utils/browser.py | 5 +-- tests/test_async.py | 46 ++++++++++++++++++++++++++ tests/test_utils_extra.py | 4 +-- 5 files changed, 84 insertions(+), 7 deletions(-) diff --git a/src/trame_server/core.py b/src/trame_server/core.py index 0d29f04..6afaa3d 100644 --- a/src/trame_server/core.py +++ b/src/trame_server/core.py @@ -13,7 +13,7 @@ from .protocol import CoreServer from .state import State from .ui import VirtualNodeManager -from .utils import share +from .utils import asynchronous, share from .utils.argument_parser import ArgumentParser from .utils.namespace import Translator @@ -744,7 +744,7 @@ def start( if exec_mode == "main": self._running_stage = 0 if self.controller.on_server_exited.exists(): - loop = asyncio.get_event_loop() + loop = asynchronous.get_event_loop() for exit_task in self.controller.on_server_exited( **self.state.to_dict() ): diff --git a/src/trame_server/utils/asynchronous.py b/src/trame_server/utils/asynchronous.py index e5ba47f..81894cc 100644 --- a/src/trame_server/utils/asynchronous.py +++ b/src/trame_server/utils/asynchronous.py @@ -1,5 +1,6 @@ import asyncio import logging +import warnings from . import is_dunder, is_private @@ -8,6 +9,7 @@ "create_state_queue_monitor_task", "create_task", "decorate_task", + "get_event_loop", "handle_task_result", "task", ] @@ -24,6 +26,34 @@ def handle_task_result(task: asyncio.Task) -> None: logging.exception("Exception raised by task = %r", task) +def get_event_loop(): + """ + Return the running event loop or, when called outside of one, the event + loop of the current thread (creating and setting it if needed). + + Unlike asyncio.get_event_loop(), this works on Python 3.14+ where no loop + is implicitly created anymore. This allows scheduling work before the + server is started, and mirrors how wslink selects its loop. + + :return: The event loop + :rtype: asyncio.AbstractEventLoop + """ + try: + return asyncio.get_running_loop() + except RuntimeError: + pass + + try: + with warnings.catch_warnings(): + # Python 3.12-3.13 warn when implicitly creating a loop + warnings.simplefilter("ignore", DeprecationWarning) + return asyncio.get_event_loop() + except RuntimeError: + loop = asyncio.new_event_loop() + asyncio.set_event_loop(loop) + return loop + + def create_task(coroutine, loop=None): """ Create a task from a coroutine while also attaching a done callback so any @@ -37,7 +67,7 @@ def create_task(coroutine, loop=None): :rtype: asyncio.Task """ if loop is None: - loop = asyncio.get_event_loop() + loop = get_event_loop() return decorate_task(loop.create_task(coroutine)) diff --git a/src/trame_server/utils/browser.py b/src/trame_server/utils/browser.py index 58f9ac7..af81954 100644 --- a/src/trame_server/utils/browser.py +++ b/src/trame_server/utils/browser.py @@ -2,10 +2,11 @@ def open_browser(server): args = server.cli.parse_known_args()[0] local_url = f"http://{args.host}:{server.port}/" try: - import asyncio # noqa: PLC0415 import webbrowser # noqa: PLC0415 - loop = asyncio.get_event_loop() + from .asynchronous import get_event_loop # noqa: PLC0415 + + loop = get_event_loop() loop.call_later(0.1, lambda: webbrowser.open(local_url)) print( "And to prevent your browser from opening, " diff --git a/tests/test_async.py b/tests/test_async.py index 608e27c..655c907 100644 --- a/tests/test_async.py +++ b/tests/test_async.py @@ -1,4 +1,5 @@ import asyncio +import contextlib import multiprocessing import time from concurrent.futures import ProcessPoolExecutor @@ -153,3 +154,48 @@ def test_state_queue_without_auto_flush(): state.exit() assert q.get_nowait() == QUEUE_EXIT + + +@pytest.fixture +def no_current_loop(): + from trame_server.utils import asynchronous # noqa: PLC0415 + + asyncio.set_event_loop(None) + try: + yield asynchronous + finally: + with contextlib.suppress(RuntimeError): + asynchronous.get_event_loop().close() + asyncio.set_event_loop(None) + + +def test_get_event_loop_without_loop(no_current_loop): + # Must not warn (3.12+) nor raise (3.14+) when no loop exists yet + loop = no_current_loop.get_event_loop() + assert isinstance(loop, asyncio.AbstractEventLoop) + assert not loop.is_closed() + # Same loop is reused afterwards + assert no_current_loop.get_event_loop() is loop + + +def test_get_event_loop_reuses_current_loop(no_current_loop): + loop = asyncio.new_event_loop() + asyncio.set_event_loop(loop) + assert no_current_loop.get_event_loop() is loop + + +def test_create_task_before_loop_is_running(no_current_loop): + async def work(): + return 42 + + task = no_current_loop.create_task(work()) + loop = no_current_loop.get_event_loop() + assert task.get_loop() is loop + assert loop.run_until_complete(task) == 42 + + +@pytest.mark.asyncio +async def test_get_event_loop_returns_running_loop(): + from trame_server.utils import asynchronous # noqa: PLC0415 + + assert asynchronous.get_event_loop() is asyncio.get_running_loop() diff --git a/tests/test_utils_extra.py b/tests/test_utils_extra.py index 9675e52..c10434e 100644 --- a/tests/test_utils_extra.py +++ b/tests/test_utils_extra.py @@ -12,7 +12,7 @@ import pytest from trame_server.ui import VirtualNodeManager -from trame_server.utils import browser, hot_reload, logger, server +from trame_server.utils import asynchronous, browser, hot_reload, logger, server from trame_server.utils.argument_parser import ArgumentParser # ----------------------------------------------------------------------------- @@ -133,7 +133,7 @@ def test_open_browser_error(monkeypatch): def fail(): raise RuntimeError - monkeypatch.setattr(asyncio, "get_event_loop", fail) + monkeypatch.setattr(asynchronous, "get_event_loop", fail) # Errors are silently ignored browser.open_browser(_fake_server()) From 19903a479cac3390ee4aab6303c8f5faa4a33cce Mon Sep 17 00:00:00 2001 From: Sebastien Jourdain Date: Wed, 7 Oct 2026 17:33:32 -0600 Subject: [PATCH 07/19] ci: test python 3.10/3.12/3.14 and align tooling with py310 - Run pytest on Python 3.10, 3.12 and 3.14 and include the version in the job name - Use coverage source_pkgs so a local trame_server directory can not shadow the installed package - Set ruff target-version to py310 to match requires-python Co-Authored-By: Claude Opus 5.5 --- .coveragerc | 2 +- .github/workflows/test_and_release.yml | 4 ++-- pyproject.toml | 2 +- 3 files changed, 4 insertions(+), 4 deletions(-) diff --git a/.coveragerc b/.coveragerc index a29d073..2234fa7 100644 --- a/.coveragerc +++ b/.coveragerc @@ -1,6 +1,6 @@ [run] omit = *docs*, *tests*, setup.py -source = trame_server +source_pkgs = trame_server [paths] source = diff --git a/.github/workflows/test_and_release.yml b/.github/workflows/test_and_release.yml index 152e096..260b4ed 100644 --- a/.github/workflows/test_and_release.yml +++ b/.github/workflows/test_and_release.yml @@ -27,12 +27,12 @@ jobs: pre-commit run --all-files pytest: - name: Pytest ${{ matrix.config.name }} + name: Pytest ${{ matrix.config.name }} (Python ${{ matrix.python-version }}) runs-on: ${{ matrix.config.os }} strategy: fail-fast: false matrix: - python-version: ["3.10"] + python-version: ["3.10", "3.12", "3.14"] config: - { name: "Linux", os: ubuntu-latest } - { name: "MacOSX", os: macos-latest } diff --git a/pyproject.toml b/pyproject.toml index 3cc55f0..0f5cda3 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -80,7 +80,7 @@ testpaths = [ [tool.ruff] line-length = 88 indent-width = 4 -target-version = "py37" +target-version = "py310" [tool.ruff.lint] extend-select = [ From b6ce105e158ff79b54737ad91ea0faa195dba12f Mon Sep 17 00:00:00 2001 From: Sebastien Jourdain Date: Wed, 7 Oct 2026 17:35:13 -0600 Subject: [PATCH 08/19] refactor: apply py310 ruff rules - Import Callable, Iterable and Iterator from collections.abc - Use builtin type instead of typing.Type - Make zip() strictness explicit (strict=False keeps current behavior) Co-Authored-By: Claude Opus 5.5 --- src/trame_server/state.py | 3 ++- src/trame_server/utils/typed_state.py | 17 ++++++++--------- 2 files changed, 10 insertions(+), 10 deletions(-) diff --git a/src/trame_server/state.py b/src/trame_server/state.py index 91b3b70..a5ebd57 100644 --- a/src/trame_server/state.py +++ b/src/trame_server/state.py @@ -2,8 +2,9 @@ import logging import weakref from collections import deque +from collections.abc import Iterable, Iterator from contextlib import contextmanager -from typing import Any, Iterable, Iterator +from typing import Any from .utils import asynchronous, is_dunder, is_private, share from .utils.hot_reload import reload diff --git a/src/trame_server/utils/typed_state.py b/src/trame_server/utils/typed_state.py index ad58711..923d2cd 100644 --- a/src/trame_server/utils/typed_state.py +++ b/src/trame_server/utils/typed_state.py @@ -1,5 +1,6 @@ import inspect from abc import ABC, abstractmethod +from collections.abc import Callable, Iterable from contextlib import contextmanager from dataclasses import MISSING, Field, fields, is_dataclass from datetime import date, datetime, time, timezone @@ -11,10 +12,7 @@ from types import UnionType from typing import ( Any, - Callable, Generic, - Iterable, - Type, TypeVar, Union, cast, @@ -65,6 +63,7 @@ def visit(kls): _resolve_type(argument, type_var_mapping) for argument in arguments ), + strict=False, ) ) visit(origin) @@ -436,7 +435,7 @@ class TypedState(Generic[T]): def __init__( self, state: State, - dataclass_type: Type[T], + dataclass_type: type[T], *, namespace="", encoders: list[IStateEncoderDecoder] | None = None, @@ -496,7 +495,7 @@ def set_dataclass(self, data: T) -> None: @classmethod def _create_state_proxy( cls, - dataclass_type: Type[T], + dataclass_type: type[T], state: State, *, namespace="", @@ -531,7 +530,7 @@ def handler(state_id: str, field: Field, field_type: type): ) @classmethod - def _create_state_names_proxy(cls, dataclass_type: Type[T], *, namespace="") -> T: + def _create_state_names_proxy(cls, dataclass_type: type[T], *, namespace="") -> T: """ Returns a State proxy with the same field structure as the input dataclass and for each field returning the fully qualified state id name associated with a dataclass leaf. @@ -549,7 +548,7 @@ def handler(state_id: str, _field: Field, _field_type: type): @classmethod def _build_proxy_cls( cls, - dataclass_type: Type[T], + dataclass_type: type[T], prefix: str, handler: Callable[[str, Field, type], Any], cls_suffix: str, @@ -610,14 +609,14 @@ def _build_proxy_cls( return cast(T, proxy_instance) @classmethod - def _get_proxy_dataclass_type(cls, instance: T) -> Type[T] | None: + def _get_proxy_dataclass_type(cls, instance: T) -> type[T] | None: """ :return: dataclass type attached to the input proxy instance. """ return getattr(instance, cls._STATE_PROXY_DATACLASS_TYPE, None) @classmethod - def _get_proxy_dataclass_type_or_raise(cls, instance: T) -> Type[T]: + def _get_proxy_dataclass_type_or_raise(cls, instance: T) -> type[T]: """ :return: dataclass type attached to the proxy instance :raises: RuntimeError if the input instance is not a proxy. From 249ebcffeb18898b85bd55e0e530a44e1a691379 Mon Sep 17 00:00:00 2001 From: Sebastien Jourdain Date: Wed, 7 Oct 2026 17:36:02 -0600 Subject: [PATCH 09/19] style(examples): format with py310 target Use parenthesized context managers now that ruff targets py310. Co-Authored-By: Claude Opus 5.5 --- examples/child-server/multi_server_router.py | 16 ++++++++++------ 1 file changed, 10 insertions(+), 6 deletions(-) diff --git a/examples/child-server/multi_server_router.py b/examples/child-server/multi_server_router.py index 268bfe0..89b0f6e 100644 --- a/examples/child-server/multi_server_router.py +++ b/examples/child-server/multi_server_router.py @@ -29,9 +29,11 @@ def test_controller(self) -> None: print(self.state.test) def _build_ui(self, template_name) -> None: - with VAppLayout( - self.server, template_name=template_name, full_height=True - ), v3.VContainer(), v3.VCard(title="This is the first app"): + with ( + VAppLayout(self.server, template_name=template_name, full_height=True), + v3.VContainer(), + v3.VCard(title="This is the first app"), + ): v3.VBtn("Test Trigger", click="console.log(test); trame.trigger('test');") v3.VBtn("Test Controller", click=self.ctrl.test_controller) @@ -57,9 +59,11 @@ def test_controller(self) -> None: print(self.state.test) def _build_ui(self, template_name) -> None: - with VAppLayout( - self.server, template_name=template_name, full_height=True - ), v3.VContainer(), v3.VCard(title="This is the second app"): + with ( + VAppLayout(self.server, template_name=template_name, full_height=True), + v3.VContainer(), + v3.VCard(title="This is the second app"), + ): v3.VBtn("Test Trigger", click="console.log(test); trame.trigger('test');") v3.VBtn("Test Controller", click=self.ctrl.test_controller) From 76c71137ad5a31f0f5be5c99cee7640e94ebe08e Mon Sep 17 00:00:00 2001 From: Sebastien Jourdain Date: Wed, 7 Oct 2026 17:36:51 -0600 Subject: [PATCH 10/19] fix(protocol): ignore unknown keys when clearing client state cache clear_state_client_cache() raised KeyError for keys that were never sent to the client. Co-Authored-By: Claude Opus 5.5 --- src/trame_server/protocol.py | 2 +- tests/test_protocol.py | 19 +++++++++++++++++++ 2 files changed, 20 insertions(+), 1 deletion(-) diff --git a/src/trame_server/protocol.py b/src/trame_server/protocol.py index edf2b1f..8ffad76 100644 --- a/src/trame_server/protocol.py +++ b/src/trame_server/protocol.py @@ -178,7 +178,7 @@ def push_actions(self, actions): def clear_state_client_cache(self, *keys): for k in keys: - del self._clients_state[k] + self._clients_state.pop(k, None) # --------------------------------------------------------------- # RPCs diff --git a/tests/test_protocol.py b/tests/test_protocol.py index 3368db7..a4b6fdf 100644 --- a/tests/test_protocol.py +++ b/tests/test_protocol.py @@ -190,3 +190,22 @@ def test_configure_auth_key(tmp_path): assert CoreServer.authentication_token == "from-file" finally: CoreServer.authentication_token = previous + + +@pytest.mark.asyncio +async def test_clear_state_client_cache(server, client): + server.state.cached = 1 + server.state.flush() + await server.network_completion + await asyncio.sleep(0.1) + assert client.state.cached == 1 + + # Unknown keys are ignored + server.clear_state_client_cache("never_sent", "cached") + + # Once cleared, an unchanged value is sent again + client.state._pushed_state["cached"] = 0 + server.protocol.push_state_change({"cached": 1}) + await server.network_completion + await asyncio.sleep(0.1) + assert client.state.cached == 1 From 16d58211b53fa497e8ccb28b18b9dd1df594c3fd Mon Sep 17 00:00:00 2001 From: Sebastien Jourdain Date: Wed, 7 Oct 2026 17:43:48 -0600 Subject: [PATCH 11/19] fix(client): ignore responses without a pending future A response for a cancelled request (e.g. after a timeout) raised InvalidStateError and a system message without a matching request raised AttributeError. Both stopped the listen loop and disconnected the client. In-flight entries are now always removed on response. Co-Authored-By: Claude Opus 5.5 --- src/trame_server/client.py | 25 +++++++++---------- tests/test_protocol.py | 51 ++++++++++++++++++++++++++++++++++++++ 2 files changed, 63 insertions(+), 13 deletions(-) diff --git a/src/trame_server/client.py b/src/trame_server/client.py index c9573c7..9d2642d 100644 --- a/src/trame_server/client.py +++ b/src/trame_server/client.py @@ -38,25 +38,27 @@ async def on_msg_complete(self, payload): msg_id = payload.get("id") msg_type, msg_topic, _ = msg_id.split(":") - future = self.in_flight_rpc.get(msg_id) + # Remove right away so we never leave a stale entry behind + future = self.in_flight_rpc.pop(msg_id, None) + + # Skip futures that were already resolved or cancelled (e.g. timeout) + pending = future is not None and not future.done() # Error if "error" in payload: - if future: + if pending: future.set_exception(Exception(payload.get("error", "Server error"))) - else: + elif future is None: print("Server error:", payload.get("error")) - self.in_flight_rpc.pop(msg_id, None) return # Normal processing msg_result = payload.get("result") # RPC - if msg_type == "rpc": - if future: - future.set_result(msg_result) + if msg_type == "rpc" and pending: + future.set_result(msg_result) # Publish if msg_type == "publish" and msg_topic in self.subscriptions: @@ -73,13 +75,10 @@ async def on_msg_complete(self, payload): if msg_id == WsLinkSession.AUTH_ID: self.client_id = msg_result.get("clientID") self.unchunker.set_max_message_size(msg_result.get("maxMsgSize")) - future.set_result(self.client_id) - else: - future.set_result(msg_result) + msg_result = self.client_id - # Clean pending future - if future: - self.in_flight_rpc.pop(msg_id) + if pending: + future.set_result(msg_result) async def listen(self): async for msg in self.ws: diff --git a/tests/test_protocol.py b/tests/test_protocol.py index a4b6fdf..20f4997 100644 --- a/tests/test_protocol.py +++ b/tests/test_protocol.py @@ -8,6 +8,7 @@ import pytest_asyncio from trame.app import asynchronous, get_client, get_server +from trame_server.client import WsLinkSession from trame_server.protocol import CoreServer @@ -209,3 +210,53 @@ async def test_clear_state_client_cache(server, client): await server.network_completion await asyncio.sleep(0.1) assert client.state.cached == 1 + + +@pytest.mark.asyncio +async def test_late_response_after_timeout_keeps_client_alive(server, client): + @server.trigger("slow") + async def slow(): + await asyncio.sleep(0.3) + return "late" + + @server.trigger("fast") + def fast(): + return "fast" + + # Cancelling the call also cancels its pending future + with pytest.raises(asyncio.TimeoutError): + await asyncio.wait_for(client.call_trigger("slow"), timeout=0.05) + + # Let the late response arrive + await asyncio.sleep(0.5) + + assert client._session.in_flight_rpc == {} + assert await client.call_trigger("fast") == "fast" + + +@pytest.mark.asyncio +async def test_responses_without_pending_future(client): + session = client._session + client_id = session.client_id + + # System messages that nobody is waiting for + await session.on_msg_complete({"id": "system:c0:99", "result": {"x": 1}}) + await session.on_msg_complete( + { + "id": WsLinkSession.AUTH_ID, + "result": {"clientID": client_id, "maxMsgSize": 1024}, + } + ) + assert session.client_id == client_id + + # Responses for futures that were cancelled + for payload in [ + {"id": "rpc:x:1", "result": 1}, + {"id": "rpc:x:1", "error": "oops"}, + {"id": "system:c0:1", "result": 1}, + ]: + future = session.loop.create_future() + future.cancel() + session.in_flight_rpc[payload["id"]] = future + await session.on_msg_complete(payload) + assert payload["id"] not in session.in_flight_rpc From 3a802d1c1cd210f1da8a6cb78fc03c7ebfd3e8bb Mon Sep 17 00:00:00 2001 From: Sebastien Jourdain Date: Wed, 7 Oct 2026 17:54:36 -0600 Subject: [PATCH 12/19] fix(client): report connected only once authenticated - Add ConnectionStatus IntEnum (DISCONNECTED, CONNECTING, CONNECTED) which stays compatible with the previous 0/1/2 values - connect() waits for the authentication response before reporting CONNECTED, so client_id is always available once connected - connect() raises when the connection or authentication fails and always resets its state, so a failed attempt no longer leaves the client stuck in CONNECTING - call_trigger() raises ConnectionError when not connected Co-Authored-By: Claude Opus 5.5 --- src/trame_server/client.py | 74 ++++++++++++++++++++------- tests/test_protocol.py | 102 +++++++++++++++++++++++++++++++++++-- 2 files changed, 155 insertions(+), 21 deletions(-) diff --git a/src/trame_server/client.py b/src/trame_server/client.py index 9d2642d..8dcf783 100644 --- a/src/trame_server/client.py +++ b/src/trame_server/client.py @@ -2,6 +2,7 @@ import logging import os import traceback +from enum import IntEnum import aiohttp import msgpack @@ -16,6 +17,14 @@ logger = logging.getLogger(__name__) +class ConnectionStatus(IntEnum): + """Connection status of a Client""" + + DISCONNECTED = 0 + CONNECTING = 1 + CONNECTED = 2 + + class WsLinkSession: CLIENT_ERROR = -32099 AUTH_ID = "system:c0:0" @@ -177,7 +186,7 @@ class Client: def __init__(self, url=None, config=None, translator=None, hot_reload=False): # Network - self._connected = 0 + self._connected = ConnectionStatus.DISCONNECTED self._session = None self._url = url self._config = {} if config is None else config @@ -192,29 +201,54 @@ def __init__(self, url=None, config=None, translator=None, hot_reload=False): ) async def connect(self, url=None, **kwargs): - if self._connected: + """ + Connect to the server and process messages until disconnected. + + The client is only reported as connected once authentication + succeeded. Raises if the connection or authentication fails. + """ + if self._connected != ConnectionStatus.DISCONNECTED: return - self._connected = 1 + self._connected = ConnectionStatus.CONNECTING config = {**self._config, **kwargs} if url is None: url = self._url - async with aiohttp.ClientSession() as session: - async with session.ws_connect(url) as ws: - self._session = WsLinkSession(ws) - self._state.ready() - self._session.register_subscription( - "trame.state.topic", self._on_state_update - ) - self._connected = 2 - task = asynchronous.create_task(self._session.listen()) - await self._session.auth(**config) - await task - - self._session.clear_subscriptions() - self._session = None - self._connected = 0 + try: + async with aiohttp.ClientSession() as session: + async with session.ws_connect(url) as ws: + self._session = WsLinkSession(ws) + self._state.ready() + self._session.register_subscription( + "trame.state.topic", self._on_state_update + ) + listen_task = asynchronous.create_task(self._session.listen()) + try: + await self._wait_for_auth(listen_task, config) + except BaseException: + listen_task.cancel() + raise + self._connected = ConnectionStatus.CONNECTED + await listen_task + finally: + if self._session: + self._session.clear_subscriptions() + self._session = None + self._connected = ConnectionStatus.DISCONNECTED + + async def _wait_for_auth(self, listen_task, config): + auth_response = await self._session.auth(**config) + await asyncio.wait( + {auth_response, listen_task}, return_when=asyncio.FIRST_COMPLETED + ) + if not auth_response.done(): + auth_response.cancel() + msg = "Connection closed before authentication completed" + raise ConnectionError(msg) + + # Raise if authentication failed + auth_response.result() async def disconnect(self): if self._session: @@ -266,6 +300,10 @@ def state(self): return self._state async def call_trigger(self, name, args=None, kwargs=None): + if self._connected != ConnectionStatus.CONNECTED: + msg = f"Client is not connected (status: {self._connected.name})" + raise ConnectionError(msg) + if args is None: args = [] diff --git a/tests/test_protocol.py b/tests/test_protocol.py index 20f4997..52261dc 100644 --- a/tests/test_protocol.py +++ b/tests/test_protocol.py @@ -1,14 +1,17 @@ import asyncio import io +import socket from argparse import Namespace from contextlib import redirect_stdout +from types import SimpleNamespace from unittest.mock import MagicMock +import aiohttp import pytest import pytest_asyncio from trame.app import asynchronous, get_client, get_server -from trame_server.client import WsLinkSession +from trame_server.client import ConnectionStatus, WsLinkSession from trame_server.protocol import CoreServer @@ -30,11 +33,13 @@ async def client(server): asynchronous.create_task(client.connect(secret="wslink-secret")) for _ in range(20): - if client._session and client._session.client_id is not None: + if client.connected == ConnectionStatus.CONNECTED: break await asyncio.sleep(0.1) - assert client.connected == 2 + # Only reported as connected once authenticated + assert client.connected == ConnectionStatus.CONNECTED + assert client._session.client_id is not None try: yield client finally: @@ -260,3 +265,94 @@ async def test_responses_without_pending_future(client): session.in_flight_rpc[payload["id"]] = future await session.on_msg_complete(payload) assert payload["id"] not in session.in_flight_rpc + + +@pytest.mark.asyncio +async def test_connected_status_follows_authentication(server): + client = get_client(f"ws://localhost:{server.port}/ws") + assert client.connected == ConnectionStatus.DISCONNECTED + + observed = [] + task = asynchronous.create_task(client.connect(secret="wslink-secret")) + deadline = asyncio.get_running_loop().time() + 5 + while asyncio.get_running_loop().time() < deadline: + status = client.connected + client_id = client._session.client_id if client._session else None + # Skip samples taken before the task started + started = observed or status != ConnectionStatus.DISCONNECTED + if started and (not observed or observed[-1][0] != status): + observed.append((status, client_id)) + if status == ConnectionStatus.CONNECTED: + break + # Sample on every loop iteration to observe each transition + await asyncio.sleep(0) + + statuses = [status for status, _ in observed] + assert statuses == [ConnectionStatus.CONNECTING, ConnectionStatus.CONNECTED] + # client_id is known as soon as we report being connected + assert observed[-1][1] is not None + + await client.disconnect() + await task + assert client.connected == ConnectionStatus.DISCONNECTED + + +@pytest.mark.asyncio +async def test_connect_authentication_failure(server): + client = get_client(f"ws://localhost:{server.port}/ws") + with pytest.raises(Exception, match="Authentication failed"): + await asyncio.wait_for(client.connect(secret="wrong"), timeout=5) + + assert client.connected == ConnectionStatus.DISCONNECTED + assert client._session is None + + +@pytest.mark.asyncio +async def test_connect_failure_allows_reconnect(server): + with socket.socket() as sock: + sock.bind(("localhost", 0)) + closed_port = sock.getsockname()[1] + + client = get_client(f"ws://localhost:{closed_port}/ws") + with pytest.raises(aiohttp.ClientConnectionError): + await client.connect() + assert client.connected == ConnectionStatus.DISCONNECTED + + # Not stuck in CONNECTING, so a new attempt goes through + task = asynchronous.create_task( + client.connect(f"ws://localhost:{server.port}/ws", secret="wslink-secret") + ) + for _ in range(50): + if client.connected == ConnectionStatus.CONNECTED: + break + await asyncio.sleep(0.1) + assert client.connected == ConnectionStatus.CONNECTED + + await client.disconnect() + await task + + +@pytest.mark.asyncio +async def test_connection_closed_before_authentication(): + client = get_client("ws://unused") + auth_response = asyncio.get_running_loop().create_future() + + async def auth(**_): + return auth_response + + async def closed(): + pass + + client._session = SimpleNamespace(auth=auth) + listen_task = asyncio.ensure_future(closed()) + + with pytest.raises(ConnectionError, match="closed before authentication"): + await client._wait_for_auth(listen_task, {}) + assert auth_response.cancelled() + + +@pytest.mark.asyncio +async def test_call_trigger_requires_connection(): + client = get_client("ws://unused") + with pytest.raises(ConnectionError, match=r"not connected.*DISCONNECTED"): + await client.call_trigger("anything") From b1d6cb47ae88a109cf580a65fca8e5f751b01c8b Mon Sep 17 00:00:00 2001 From: Sebastien Jourdain Date: Wed, 7 Oct 2026 18:00:32 -0600 Subject: [PATCH 13/19] refactor(client): share request sending between auth and call Co-Authored-By: Claude Opus 5.5 --- src/trame_server/client.py | 35 +++++++++++------------------------ 1 file changed, 11 insertions(+), 24 deletions(-) diff --git a/src/trame_server/client.py b/src/trame_server/client.py index 8dcf783..a3faa23 100644 --- a/src/trame_server/client.py +++ b/src/trame_server/client.py @@ -107,35 +107,20 @@ async def listen(self): await self.on_msg_complete(full_message) async def auth(self, **kwargs): - key = WsLinkSession.AUTH_ID - resp = self.loop.create_future() - wrapper = { - "wslink": "1.0", - "id": key, - "method": "wslink.hello", - "args": [kwargs], - "kwargs": {}, - } - - packed_wrapper = msgpack.packb(wrapper) - self.in_flight_rpc[key] = resp - - async with self.attachment_atomic: - for chunk in generate_chunks(packed_wrapper, MAX_MSG_SIZE): - if self.ws is not None: - await self.ws.send_bytes(chunk) - - return resp + return await self._send(WsLinkSession.AUTH_ID, "wslink.hello", [kwargs], {}) async def call(self, method, args=None, kwargs=None): self.msg_count += 1 key = f"rpc:{self.client_id}:{self.msg_count}" - resp = self.loop.create_future() - if args is None: - args = [] - if kwargs is None: - kwargs = {} + return await self._send( + key, + method, + [] if args is None else args, + {} if kwargs is None else kwargs, + ) + async def _send(self, key, method, args, kwargs): + """Send a request and return the future resolved by its response""" wrapper = { "wslink": "1.0", "id": key, @@ -144,7 +129,9 @@ async def call(self, method, args=None, kwargs=None): "kwargs": kwargs, } + # Only track the request once we know it can be serialized packed_wrapper = msgpack.packb(wrapper) + resp = self.loop.create_future() self.in_flight_rpc[key] = resp async with self.attachment_atomic: From 1b895c491f8741c70024b61a5f1df1ff762be0fd Mon Sep 17 00:00:00 2001 From: Sebastien Jourdain Date: Wed, 7 Oct 2026 18:07:20 -0600 Subject: [PATCH 14/19] fix(server): translate state names in clear_state_client_cache On a child server with a prefix, the untranslated names did not match the cached keys, so nothing was cleared and unchanged values were never sent again. Translate them like force_state_push does. Co-Authored-By: Claude Opus 5.5 --- src/trame_server/core.py | 10 +++++++++- tests/test_protocol.py | 23 +++++++++++++++++++++++ 2 files changed, 32 insertions(+), 1 deletion(-) diff --git a/src/trame_server/core.py b/src/trame_server/core.py index 6afaa3d..529188e 100644 --- a/src/trame_server/core.py +++ b/src/trame_server/core.py @@ -504,9 +504,17 @@ def get_server_state(self): } def clear_state_client_cache(self, *state_names): + """ + Forget what was last sent to the client for the given state names, + so their next push is sent even if their value did not change. + + :param *state_names: Set of key names as seen by this server. + """ protocol = self.protocol if protocol: - protocol.clear_state_client_cache(*state_names) + protocol.clear_state_client_cache( + *[self._translator.translate_key(k) for k in state_names] + ) # ------------------------------------------------------------------------- diff --git a/tests/test_protocol.py b/tests/test_protocol.py index 52261dc..ce9564e 100644 --- a/tests/test_protocol.py +++ b/tests/test_protocol.py @@ -356,3 +356,26 @@ async def test_call_trigger_requires_connection(): client = get_client("ws://unused") with pytest.raises(ConnectionError, match=r"not connected.*DISCONNECTED"): await client.call_trigger("anything") + + +@pytest.mark.asyncio +async def test_clear_state_client_cache_on_child_server(server, client): + child_server = server.create_child_server(prefix="child_") + child_server.state.value = 1 + child_server.state.flush() + await server.network_completion + await asyncio.sleep(0.1) + assert client.state.child_value == 1 + assert "child_value" in server.protocol._clients_state + + # Names are given as seen by the child server, like force_state_push + child_server.clear_state_client_cache("value") + assert "child_value" not in server.protocol._clients_state + + # Once cleared, an unchanged value is sent again + client.state._pushed_state["child_value"] = 0 + child_server.state.dirty("value") + child_server.state.flush() + await server.network_completion + await asyncio.sleep(0.1) + assert client.state.child_value == 1 From 28f1a404fcfdec9f132b04e0aa067ccae25b49e1 Mon Sep 17 00:00:00 2001 From: Sebastien Jourdain Date: Wed, 7 Oct 2026 18:15:11 -0600 Subject: [PATCH 15/19] refactor: use logging instead of print for diagnostics - client: server errors, subscription callback errors (with traceback) and websocket errors are logged as errors, close events as debug - protocol: JS errors without handler are logged as errors and missing triggers as warnings Warnings and errors still reach stderr when no logging is configured. Co-Authored-By: Claude Opus 5.5 --- src/trame_server/client.py | 20 +++++----- src/trame_server/protocol.py | 8 +++- tests/test_protocol.py | 76 ++++++++++++++++++++++++++---------- 3 files changed, 71 insertions(+), 33 deletions(-) diff --git a/src/trame_server/client.py b/src/trame_server/client.py index a3faa23..26c5b21 100644 --- a/src/trame_server/client.py +++ b/src/trame_server/client.py @@ -1,7 +1,6 @@ import asyncio import logging import os -import traceback from enum import IntEnum import aiohttp @@ -16,6 +15,10 @@ logger = logging.getLogger(__name__) +_WS_CLOSE_TYPES = frozenset( + {aiohttp.WSMsgType.CLOSE, aiohttp.WSMsgType.CLOSING, aiohttp.WSMsgType.CLOSED} +) + class ConnectionStatus(IntEnum): """Connection status of a Client""" @@ -58,7 +61,7 @@ async def on_msg_complete(self, payload): if pending: future.set_exception(Exception(payload.get("error", "Server error"))) elif future is None: - print("Server error:", payload.get("error")) + logger.error("Server error: %s", payload.get("error")) return @@ -76,8 +79,7 @@ async def on_msg_complete(self, payload): try: fn(event) except Exception: - print("Subscription callback error") - traceback.print_exc() + logger.exception("Subscription callback error (%s)", msg_topic) # System if msg_type == "system": @@ -91,14 +93,10 @@ async def on_msg_complete(self, payload): async def listen(self): async for msg in self.ws: - if msg.type == aiohttp.WSMsgType.CLOSE: - print("CLOSE") - elif msg.type == aiohttp.WSMsgType.CLOSING: - print("CLOSING") - elif msg.type == aiohttp.WSMsgType.CLOSED: - print("CLOSED") + if msg.type in _WS_CLOSE_TYPES: + logger.debug("WebSocket %s", msg.type.name) elif msg.type == aiohttp.WSMsgType.ERROR: - print("ERROR") + logger.error("WebSocket error: %s", self.ws.exception()) elif msg.type == aiohttp.WSMsgType.TEXT: logger.critical("wslink is not expecting text message:\n> %s", msg.data) if msg.type == aiohttp.WSMsgType.BINARY: diff --git a/src/trame_server/protocol.py b/src/trame_server/protocol.py index 8ffad76..8205e4d 100644 --- a/src/trame_server/protocol.py +++ b/src/trame_server/protocol.py @@ -1,4 +1,5 @@ import inspect +import logging import os from pathlib import Path @@ -10,6 +11,9 @@ from trame_server.state import TRAME_NON_INIT_VALUE from trame_server.utils import clean_state, logger +# 'logger' is the network exchange logger (--log-network) +log = logging.getLogger(__name__) + class CoreServer(ServerProtocol): authentication_token = "wslink-secret" @@ -209,7 +213,7 @@ def js_error(self, message): if self.server.controller.on_error.exists(): self.server.controller.on_error(message) else: - print(f" JS Error => {message}") + log.error("JS Error => %s", message) # --------------------------------------------------------------- @@ -233,7 +237,7 @@ async def trigger(self, name, args, kwargs): if inspect.isawaitable(result): result = await result return result - print(f"Trigger {name} seems to be missing") + log.warning("Trigger %s seems to be missing", name) return None diff --git a/tests/test_protocol.py b/tests/test_protocol.py index ce9564e..bd1fd0d 100644 --- a/tests/test_protocol.py +++ b/tests/test_protocol.py @@ -1,8 +1,7 @@ import asyncio -import io +import logging import socket from argparse import Namespace -from contextlib import redirect_stdout from types import SimpleNamespace from unittest.mock import MagicMock @@ -47,6 +46,14 @@ async def client(server): await client.disconnect() +def _logged(caplog, level, message): + """Return the record matching level and message, if any""" + for record in caplog.records: + if record.levelname == level and record.getMessage() == message: + return record + return None + + async def _rpc(client, method, *args): return await (await client._session.call(method, list(args))) @@ -86,7 +93,7 @@ def echo(value): @pytest.mark.asyncio -async def test_async_and_missing_trigger(server, client): +async def test_async_and_missing_trigger(server, client, caplog): @server.trigger("async_add") async def async_add(a, b): await asyncio.sleep(0.01) @@ -94,13 +101,12 @@ async def async_add(a, b): assert await client.call_trigger("async_add", [1, 2]) == 3 - with io.StringIO() as buf, redirect_stdout(buf): - assert await client.call_trigger("not_registered") is None - assert "Trigger not_registered seems to be missing" in buf.getvalue() + assert await client.call_trigger("not_registered") is None + assert _logged(caplog, "WARNING", "Trigger not_registered seems to be missing") @pytest.mark.asyncio -async def test_lifecycle_and_js_error(server, client): +async def test_lifecycle_and_js_error(server, client, caplog): on_custom = MagicMock() server.controller.on_custom = on_custom await _rpc(client, "trame.lifecycle.update", "custom") @@ -109,10 +115,9 @@ async def test_lifecycle_and_js_error(server, client): # Unknown life cycle name is a no-op await _rpc(client, "trame.lifecycle.update", "unknown_cycle") - # JS error without handler is printed - with io.StringIO() as buf, redirect_stdout(buf): - await _rpc(client, "trame.error.client", "boom") - assert "JS Error => boom" in buf.getvalue() + # JS error without handler is logged + await _rpc(client, "trame.error.client", "boom") + assert _logged(caplog, "ERROR", "JS Error => boom") # JS error with handler on_error = MagicMock() @@ -150,7 +155,7 @@ async def test_protocol_call(server): @pytest.mark.asyncio -async def test_client_subscriptions(client): +async def test_client_subscriptions(client, caplog): session = client._session received = [] @@ -161,17 +166,14 @@ def broken(_): session.register_subscription("custom.topic", received.append) session.register_subscription("custom.topic", broken) - with io.StringIO() as buf, redirect_stdout(buf): - await session.on_msg_complete( - {"id": "publish:custom.topic:0", "result": {"x": 1}} - ) - assert "Subscription callback error" in buf.getvalue() + await session.on_msg_complete({"id": "publish:custom.topic:0", "result": {"x": 1}}) assert received == [{"x": 1}] + record = _logged(caplog, "ERROR", "Subscription callback error (custom.topic)") + assert record.exc_info[0] is ValueError # Error without matching in-flight rpc - with io.StringIO() as buf, redirect_stdout(buf): - await session.on_msg_complete({"id": "rpc:x:999", "error": "oops"}) - assert "Server error: oops" in buf.getvalue() + await session.on_msg_complete({"id": "rpc:x:999", "error": "oops"}) + assert _logged(caplog, "ERROR", "Server error: oops") # Notification without id is ignored await session.on_msg_complete({"result": None}) @@ -379,3 +381,37 @@ async def test_clear_state_client_cache_on_child_server(server, client): await server.network_completion await asyncio.sleep(0.1) assert client.state.child_value == 1 + + +class FakeWebSocket: + def __init__(self, *msg_types): + self._messages = [SimpleNamespace(type=t, data="text") for t in msg_types] + + def __aiter__(self): + return self._iterate() + + async def _iterate(self): + for msg in self._messages: + yield msg + + def exception(self): + return RuntimeError("connection reset") + + +@pytest.mark.asyncio +async def test_listen_logs_websocket_events(caplog): + caplog.set_level(logging.DEBUG, logger="trame_server.client") + session = WsLinkSession( + FakeWebSocket( + aiohttp.WSMsgType.TEXT, + aiohttp.WSMsgType.ERROR, + aiohttp.WSMsgType.CLOSING, + aiohttp.WSMsgType.CLOSED, + ) + ) + await session.listen() + + assert _logged(caplog, "CRITICAL", "wslink is not expecting text message:\n> text") + assert _logged(caplog, "ERROR", "WebSocket error: connection reset") + assert _logged(caplog, "DEBUG", "WebSocket CLOSING") + assert _logged(caplog, "DEBUG", "WebSocket CLOSED") From 18cca2085a4a678a07491fe5a7d0a55420c969c7 Mon Sep 17 00:00:00 2001 From: Sebastien Jourdain Date: Thu, 8 Oct 2026 10:45:32 -0600 Subject: [PATCH 16/19] test: stop test_server_start_sync from leaking environment TRAME_ARGS="--banner --no-http" leaked into every later test, so their servers ran without static routes, and trame_net.log was written to the repository root. Use monkeypatch and tmp_path instead. Co-Authored-By: Claude Opus 5.5 --- tests/test_server.py | 7 +++---- 1 file changed, 3 insertions(+), 4 deletions(-) diff --git a/tests/test_server.py b/tests/test_server.py index 46f80b6..d165083 100644 --- a/tests/test_server.py +++ b/tests/test_server.py @@ -1,5 +1,4 @@ import asyncio -import os import subprocess import sys from pathlib import Path @@ -206,9 +205,9 @@ def register_protocol(protocol): await child_server.stop() -def test_server_start_sync(): - os.environ["TRAME_ARGS"] = "--banner --no-http" - os.environ["TRAME_LOG_NETWORK"] = "trame_net.log" +def test_server_start_sync(monkeypatch, tmp_path): + monkeypatch.setenv("TRAME_ARGS", "--banner --no-http") + monkeypatch.setenv("TRAME_LOG_NETWORK", str(tmp_path / "trame_net.log")) server = get_server("test_server_start_sync") server.serve.update( { From 0623638913b3b2791975cc0630c08e5eed6f729e Mon Sep 17 00:00:00 2001 From: Sebastien Jourdain Date: Thu, 8 Oct 2026 10:45:35 -0600 Subject: [PATCH 17/19] fix(protocol): rely on wslink for on_server_bind route precedence The aiohttp router reordering hack no longer worked with aiohttp>=3.10, which resolves routes through an index the hack never updated, so a "/" route added in on_server_bind was shadowed by wslink's redirect. wslink 2.5.8 registers its static routes on start, so routes added in on_server_bind take precedence on any aiohttp version. Co-Authored-By: Claude Opus 5.5 --- pyproject.toml | 2 +- src/trame_server/protocol.py | 8 ++----- tests/test_server.py | 42 ++++++++++++++++++++++++++++++++++++ 3 files changed, 45 insertions(+), 7 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index 0f5cda3..6339aff 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -6,7 +6,7 @@ authors = [ {name = "Kitware Inc."}, ] dependencies = [ - "wslink>=2.5.7,<3", + "wslink>=2.5.8,<3", "trame-common>=1.2.7", "more-itertools", ] diff --git a/src/trame_server/protocol.py b/src/trame_server/protocol.py index 8205e4d..4c3743f 100644 --- a/src/trame_server/protocol.py +++ b/src/trame_server/protocol.py @@ -94,13 +94,9 @@ def initialize(self): # Called by wslink def set_server(self, _server): self.server._server = _server if self.server.controller.on_server_bind.exists(): - # Hack - use internal of aiohttp to rework routes order - # FIXME: WON'T WORK with a different backend - server_routes = _server.app.router._resources - wslink_routes = list(server_routes) - server_routes.clear() + # wslink registers its static routes on start, so routes added + # here take precedence over them self.server.controller.on_server_bind(_server) - server_routes.extend(wslink_routes) def port_callback(self, port_used): self.server._running_port = port_used diff --git a/tests/test_server.py b/tests/test_server.py index d165083..c88d077 100644 --- a/tests/test_server.py +++ b/tests/test_server.py @@ -3,7 +3,9 @@ import sys from pathlib import Path +import aiohttp import pytest +from aiohttp import web from trame.app import get_server from trame.modules import www from wslink import register as export_rpc @@ -245,3 +247,43 @@ async def test_server_task_cancel_doesnt_hang_ready_future(): task = server.start(exec_mode="task", port=0) task.cancel() await asyncio.wait_for(server.ready, timeout=1) + + +def _text_handler(text): + async def handler(_request): + return web.Response(text=text) + + return handler + + +@pytest.mark.asyncio +async def test_on_server_bind_routes_take_precedence(): + server = get_server("test_on_server_bind_routes") + + @server.controller.add("on_server_bind") + def add_routes(wslink_server): + wslink_server.app.router.add_routes( + [ + web.get("/", _text_handler("custom root")), + web.get("/index.html", _text_handler("custom index")), + web.get("/api/hello", _text_handler("hello")), + ] + ) + + server.start(exec_mode="task", port=0) + assert await server.ready + + try: + async with aiohttp.ClientSession() as session: + for path, expected in [ + ("/", "custom root"), + ("/index.html", "custom index"), + ("/api/hello", "hello"), + ]: + url = f"http://localhost:{server.port}{path}" + async with session.get(url, allow_redirects=False) as response: + assert response.status == 200, path + assert await response.text() == expected + finally: + await asyncio.sleep(0.1) + await server.stop() From 8f68834c3e391bc2375b79d35f341dfb7b68d382 Mon Sep 17 00:00:00 2001 From: Sebastien Jourdain Date: Thu, 8 Oct 2026 10:56:06 -0600 Subject: [PATCH 18/19] test(protocol): avoid relying on client internals Use a raw WsLinkSession to observe what the server pushes instead of corrupting the client state cache or inspecting private attributes. Co-Authored-By: Claude Opus 5.5 --- tests/test_protocol.py | 195 +++++++++++++++++++++++------------------ 1 file changed, 110 insertions(+), 85 deletions(-) diff --git a/tests/test_protocol.py b/tests/test_protocol.py index bd1fd0d..7a29803 100644 --- a/tests/test_protocol.py +++ b/tests/test_protocol.py @@ -8,6 +8,7 @@ import aiohttp import pytest import pytest_asyncio +from aiohttp import web from trame.app import asynchronous, get_client, get_server from trame_server.client import ConnectionStatus, WsLinkSession @@ -38,7 +39,6 @@ async def client(server): # Only reported as connected once authenticated assert client.connected == ConnectionStatus.CONNECTED - assert client._session.client_id is not None try: yield client finally: @@ -46,6 +46,35 @@ async def client(server): await client.disconnect() +@pytest_asyncio.fixture +async def session(server): + """Raw wslink session recording the state pushed by the server""" + async with ( + aiohttp.ClientSession() as http, + http.ws_connect(f"ws://localhost:{server.port}/ws") as ws, + ): + session = WsLinkSession(ws) + session.pushed_states = [] + session.register_subscription("trame.state.topic", session.pushed_states.append) + listen_task = asynchronous.create_task(session.listen()) + await (await session.auth(secret="wslink-secret")) + try: + yield session + finally: + await session.close() + await listen_task + + +async def _rpc(session, method, *args): + return await (await session.call(method, list(args))) + + +async def _flush(server, state=None): + (state or server.state).flush() + await server.network_completion + await asyncio.sleep(0.1) + + def _logged(caplog, level, message): """Return the record matching level and message, if any""" for record in caplog.records: @@ -54,39 +83,33 @@ def _logged(caplog, level, message): return None -async def _rpc(client, method, *args): - return await (await client._session.call(method, list(args))) - - @pytest.mark.asyncio -async def test_rpc_error_keeps_client_alive(server, client): +async def test_rpc_error_keeps_session_alive(server, session): with pytest.raises(Exception, match="Unregistered method called"): - await _rpc(client, "trame.does.not.exist") + await _rpc(session, "trame.does.not.exist") # The connection must still be usable after an error @server.trigger("ping") def ping(): return "pong" - assert await client.call_trigger("ping") == "pong" + assert await _rpc(session, "trame.trigger", "ping", [], {}) == "pong" @pytest.mark.asyncio -async def test_unserializable_args_raise_serialization_error(server, client): +async def test_unserializable_args_raise_serialization_error(server, client, session): @server.trigger("echo") def echo(value): return value - pending = dict(client._session.in_flight_rpc) - with pytest.raises(TypeError, match="can not serialize"): await client.call_trigger("echo", [object()]) with pytest.raises(TypeError, match="can not serialize"): - await client._session.auth(secret=object()) + await session.auth(secret=object()) # Failed requests are not left waiting for a response - assert client._session.in_flight_rpc == pending + assert session.in_flight_rpc == {} # Connection is still usable assert await client.call_trigger("echo", [1]) == 1 @@ -106,46 +129,51 @@ async def async_add(a, b): @pytest.mark.asyncio -async def test_lifecycle_and_js_error(server, client, caplog): +async def test_lifecycle_and_js_error(server, session, caplog): on_custom = MagicMock() server.controller.on_custom = on_custom - await _rpc(client, "trame.lifecycle.update", "custom") + await _rpc(session, "trame.lifecycle.update", "custom") on_custom.assert_called_once_with() # Unknown life cycle name is a no-op - await _rpc(client, "trame.lifecycle.update", "unknown_cycle") + await _rpc(session, "trame.lifecycle.update", "unknown_cycle") # JS error without handler is logged - await _rpc(client, "trame.error.client", "boom") + await _rpc(session, "trame.error.client", "boom") assert _logged(caplog, "ERROR", "JS Error => boom") # JS error with handler on_error = MagicMock() server.controller.on_error = on_error - await _rpc(client, "trame.error.client", "boom again") + await _rpc(session, "trame.error.client", "boom again") on_error.assert_called_once_with("boom again") @pytest.mark.asyncio -async def test_get_state_and_force_push(server, client): +async def test_get_state_and_force_push(server, session): server.state.forced = 1 - server.state.flush() - await server.network_completion - await asyncio.sleep(0.1) - assert client.state.forced == 1 + await _flush(server) + assert session.pushed_states == [{"forced": 1}] - full_state = await _rpc(client, "trame.state.get") + full_state = await _rpc(session, "trame.state.get") assert full_state["state"]["forced"] == 1 - # Client corrupts its local copy, server forces a resend - client.state._pushed_state["forced"] = 0 + # Unchanged value is not sent again... + session.pushed_states.clear() + server.state.dirty("forced") + await _flush(server) + assert session.pushed_states == [] + + # ...unless the server forces a resend server.force_state_push("forced") - await server.network_completion - await asyncio.sleep(0.1) - assert client.state.forced == 1 + await _flush(server) + assert session.pushed_states == [{"forced": 1}] # No keys is a no-op + session.pushed_states.clear() server.force_state_push() + await _flush(server) + assert session.pushed_states == [] @pytest.mark.asyncio @@ -155,8 +183,8 @@ async def test_protocol_call(server): @pytest.mark.asyncio -async def test_client_subscriptions(client, caplog): - session = client._session +async def test_session_subscriptions(caplog): + session = WsLinkSession(None) received = [] def broken(_): @@ -201,22 +229,18 @@ def test_configure_auth_key(tmp_path): @pytest.mark.asyncio -async def test_clear_state_client_cache(server, client): +async def test_clear_state_client_cache(server, session): server.state.cached = 1 - server.state.flush() - await server.network_completion - await asyncio.sleep(0.1) - assert client.state.cached == 1 + await _flush(server) + assert session.pushed_states == [{"cached": 1}] # Unknown keys are ignored server.clear_state_client_cache("never_sent", "cached") # Once cleared, an unchanged value is sent again - client.state._pushed_state["cached"] = 0 - server.protocol.push_state_change({"cached": 1}) - await server.network_completion - await asyncio.sleep(0.1) - assert client.state.cached == 1 + server.state.dirty("cached") + await _flush(server) + assert session.pushed_states == [{"cached": 1}, {"cached": 1}] @pytest.mark.asyncio @@ -237,24 +261,22 @@ def fast(): # Let the late response arrive await asyncio.sleep(0.5) - assert client._session.in_flight_rpc == {} assert await client.call_trigger("fast") == "fast" @pytest.mark.asyncio -async def test_responses_without_pending_future(client): - session = client._session - client_id = session.client_id +async def test_responses_without_pending_future(): + session = WsLinkSession(None) # System messages that nobody is waiting for await session.on_msg_complete({"id": "system:c0:99", "result": {"x": 1}}) await session.on_msg_complete( { "id": WsLinkSession.AUTH_ID, - "result": {"clientID": client_id, "maxMsgSize": 1024}, + "result": {"clientID": "c1", "maxMsgSize": 1024}, } ) - assert session.client_id == client_id + assert session.client_id == "c1" # Responses for futures that were cancelled for payload in [ @@ -274,25 +296,28 @@ async def test_connected_status_follows_authentication(server): client = get_client(f"ws://localhost:{server.port}/ws") assert client.connected == ConnectionStatus.DISCONNECTED - observed = [] + statuses = [] task = asynchronous.create_task(client.connect(secret="wslink-secret")) deadline = asyncio.get_running_loop().time() + 5 while asyncio.get_running_loop().time() < deadline: status = client.connected - client_id = client._session.client_id if client._session else None # Skip samples taken before the task started - started = observed or status != ConnectionStatus.DISCONNECTED - if started and (not observed or observed[-1][0] != status): - observed.append((status, client_id)) + started = statuses or status != ConnectionStatus.DISCONNECTED + if started and (not statuses or statuses[-1] != status): + statuses.append(status) if status == ConnectionStatus.CONNECTED: break # Sample on every loop iteration to observe each transition await asyncio.sleep(0) - statuses = [status for status, _ in observed] assert statuses == [ConnectionStatus.CONNECTING, ConnectionStatus.CONNECTED] - # client_id is known as soon as we report being connected - assert observed[-1][1] is not None + + # Usable as soon as we report being connected + @server.trigger("ping") + def ping(): + return "pong" + + assert await client.call_trigger("ping") == "pong" await client.disconnect() await task @@ -306,7 +331,6 @@ async def test_connect_authentication_failure(server): await asyncio.wait_for(client.connect(secret="wrong"), timeout=5) assert client.connected == ConnectionStatus.DISCONNECTED - assert client._session is None @pytest.mark.asyncio @@ -336,21 +360,26 @@ async def test_connect_failure_allows_reconnect(server): @pytest.mark.asyncio async def test_connection_closed_before_authentication(): - client = get_client("ws://unused") - auth_response = asyncio.get_running_loop().create_future() - - async def auth(**_): - return auth_response - - async def closed(): - pass + async def close_right_away(request): + ws = web.WebSocketResponse() + await ws.prepare(request) + await ws.close() + return ws + + app = web.Application() + app.router.add_get("/ws", close_right_away) + runner = web.AppRunner(app) + await runner.setup() + await web.TCPSite(runner, "localhost", 0).start() + port = runner.addresses[0][1] - client._session = SimpleNamespace(auth=auth) - listen_task = asyncio.ensure_future(closed()) - - with pytest.raises(ConnectionError, match="closed before authentication"): - await client._wait_for_auth(listen_task, {}) - assert auth_response.cancelled() + try: + client = get_client(f"ws://localhost:{port}/ws") + with pytest.raises(ConnectionError, match="closed before authentication"): + await asyncio.wait_for(client.connect(), timeout=5) + assert client.connected == ConnectionStatus.DISCONNECTED + finally: + await runner.cleanup() @pytest.mark.asyncio @@ -361,26 +390,22 @@ async def test_call_trigger_requires_connection(): @pytest.mark.asyncio -async def test_clear_state_client_cache_on_child_server(server, client): +async def test_clear_state_client_cache_on_child_server(server, session): child_server = server.create_child_server(prefix="child_") child_server.state.value = 1 - child_server.state.flush() - await server.network_completion - await asyncio.sleep(0.1) - assert client.state.child_value == 1 - assert "child_value" in server.protocol._clients_state + await _flush(server, child_server.state) + assert session.pushed_states == [{"child_value": 1}] - # Names are given as seen by the child server, like force_state_push - child_server.clear_state_client_cache("value") - assert "child_value" not in server.protocol._clients_state + # Unchanged value is not sent again... + child_server.state.dirty("value") + await _flush(server, child_server.state) + assert session.pushed_states == [{"child_value": 1}] - # Once cleared, an unchanged value is sent again - client.state._pushed_state["child_value"] = 0 + # ...unless cleared, using the names as seen by the child server + child_server.clear_state_client_cache("value") child_server.state.dirty("value") - child_server.state.flush() - await server.network_completion - await asyncio.sleep(0.1) - assert client.state.child_value == 1 + await _flush(server, child_server.state) + assert session.pushed_states == [{"child_value": 1}, {"child_value": 1}] class FakeWebSocket: From 0273d33a758078a8b76cac9a9005f2c0f7509a00 Mon Sep 17 00:00:00 2001 From: Sebastien Jourdain Date: Thu, 8 Oct 2026 11:39:40 -0600 Subject: [PATCH 19/19] ci: document python versions used for testing Co-Authored-By: Claude Opus 5.5 --- .github/workflows/test_and_release.yml | 1 + 1 file changed, 1 insertion(+) diff --git a/.github/workflows/test_and_release.yml b/.github/workflows/test_and_release.yml index 260b4ed..9beddd1 100644 --- a/.github/workflows/test_and_release.yml +++ b/.github/workflows/test_and_release.yml @@ -32,6 +32,7 @@ jobs: strategy: fail-fast: false matrix: + # min supported, commonly used (e.g. ParaView), latest python-version: ["3.10", "3.12", "3.14"] config: - { name: "Linux", os: ubuntu-latest }