From 00a6136bc568e0d6598984f7ffea257fbd936ba1 Mon Sep 17 00:00:00 2001 From: "F.N. Claessen" Date: Sun, 8 Nov 2026 15:46:20 +0100 Subject: [PATCH 1/2] test(S2): cover what the handler base, the CEM and the example server leave untested Raise combined line coverage from 95.17% to 95.93%, above main's 95.68% before the S2 package counted: an invalid message answered as INVALID_DATA, messages given as JSON, failure callbacks, activating the control type in force or one the resource manager lacks, re-registering a control type, an empty schedule, and the example server's websocket producer and consumer (send, reset, bad JSON, close, connection error). Each test fails when the code it covers is broken (6 of 6 mutants killed). Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01CcpawsJs74cSwFLXkz7ZwW Signed-off-by: F.N. Claessen --- tests/s2/test_example_server_websocket.py | 139 +++++++++++++++++++ tests/s2/test_handler_base.py | 159 ++++++++++++++++++++++ 2 files changed, 298 insertions(+) create mode 100644 tests/s2/test_example_server_websocket.py create mode 100644 tests/s2/test_handler_base.py diff --git a/tests/s2/test_example_server_websocket.py b/tests/s2/test_example_server_websocket.py new file mode 100644 index 00000000..06aa76c0 --- /dev/null +++ b/tests/s2/test_example_server_websocket.py @@ -0,0 +1,139 @@ +"""How the example server passes messages between a websocket and its CEM.""" + +from __future__ import annotations + +from types import SimpleNamespace + +import aiohttp +import pytest + +from flexmeasures_client.s2.script import websockets_server as server + + +class FakeCEM: + """Just enough of a CEM: a queue of messages to send, and a record of what it handled.""" + + def __init__(self, to_send=(), handle_raises=None): + self._to_send = list(to_send) + self.handled = [] + self.closed = False + self._handle_raises = handle_raises + self._logger = server.LOGGER + + def is_closed(self): + return self.closed or not self._to_send + + async def get_message(self): + return self._to_send.pop(0) + + async def handle_message(self, message): + if self._handle_raises: + raise self._handle_raises + self.handled.append(message) + + async def close(self): + self.closed = True + + +class FakeWebSocket: + def __init__(self, incoming=(), send_fails_after=None): + self._incoming = list(incoming) + self.sent = [] + self.closed = False + self._send_fails_after = send_fails_after + + async def send_json(self, message): + if ( + self._send_fails_after is not None + and len(self.sent) >= self._send_fails_after + ): + raise aiohttp.ClientConnectionResetError("gone") + self.sent.append(message) + + async def close(self): + self.closed = True + + def exception(self): + return RuntimeError("broken pipe") + + def __aiter__(self): + return self + + async def __anext__(self): + if not self._incoming: + raise StopAsyncIteration + return self._incoming.pop(0) + + +def text(data: str): + return SimpleNamespace(type=aiohttp.WSMsgType.TEXT, data=data) + + +@pytest.mark.asyncio +async def test_the_producer_sends_what_the_cem_queues(): + cem = FakeCEM(to_send=[{"n": 1}, {"n": 2}]) + ws = FakeWebSocket() + + await server.websocket_producer(ws, cem) + + assert ws.sent == [{"n": 1}, {"n": 2}] + + +@pytest.mark.asyncio +async def test_the_producer_stops_when_the_connection_is_reset(): + cem = FakeCEM(to_send=[{"n": 1}, {"n": 2}, {"n": 3}]) + ws = FakeWebSocket(send_fails_after=1) + + await server.websocket_producer(ws, cem) + + assert ws.sent == [{"n": 1}] + assert cem._to_send == [ + {"n": 3} + ], "nothing more is taken from the CEM once the connection is gone" + + +@pytest.mark.asyncio +async def test_the_consumer_hands_each_message_to_the_cem(): + cem = FakeCEM() + ws = FakeWebSocket(incoming=[text('{"message_type": "Handshake"}')]) + + await server.websocket_consumer(ws, cem) + + assert cem.handled == [{"message_type": "Handshake"}] + + +@pytest.mark.asyncio +async def test_the_consumer_logs_a_message_that_is_not_json_and_carries_on(caplog): + cem = FakeCEM() + ws = FakeWebSocket( + incoming=[text("not json"), text('{"message_type": "Handshake"}')] + ) + + await server.websocket_consumer(ws, cem) + + assert cem.handled == [{"message_type": "Handshake"}] + assert "handle_message failed for 'not json'" in caplog.text + + +@pytest.mark.asyncio +async def test_close_closes_the_cem_and_the_connection(): + cem = FakeCEM() + ws = FakeWebSocket(incoming=[text("close")]) + + await server.websocket_consumer(ws, cem) + + assert cem.closed and ws.closed + assert cem.handled == [] + + +@pytest.mark.asyncio +async def test_a_connection_error_closes_the_cem(caplog): + cem = FakeCEM() + ws = FakeWebSocket( + incoming=[SimpleNamespace(type=aiohttp.WSMsgType.ERROR, data=None)] + ) + + await server.websocket_consumer(ws, cem) + + assert cem.closed + assert "ws connection closed with exception broken pipe" in caplog.text diff --git a/tests/s2/test_handler_base.py b/tests/s2/test_handler_base.py new file mode 100644 index 00000000..3e9af23f --- /dev/null +++ b/tests/s2/test_handler_base.py @@ -0,0 +1,159 @@ +"""What every S2 handler does with the messages it is given, whatever its control type.""" + +from __future__ import annotations + +import json + +import pandas as pd +import pytest +from s2python.common import ControlType, Handshake, ReceptionStatusValues + +from flexmeasures_client.s2 import Handler, register +from flexmeasures_client.s2.cem import CEM +from flexmeasures_client.s2.control_types.FRBC import FRBCTest +from flexmeasures_client.s2.utils import get_reception_status + + +class HandshakeHandler(Handler): + """A handler for one message type, which counts what reaches it.""" + + def __init__(self): + super().__init__() + self.handled = [] + + @register(Handshake) + def handle_handshake(self, message: Handshake): + self.handled.append(message) + return get_reception_status(message) + + +@pytest.mark.asyncio +async def test_an_invalid_message_is_answered_as_invalid_data(rm_handshake): + handler = HandshakeHandler() + message = rm_handshake.to_dict() + message["supported_protocol_versions"] = "not a list" + + answer = await handler.handle_message(message) + + assert answer.status == ReceptionStatusValues.INVALID_DATA + assert str(answer.subject_message_id) == str(message["message_id"]) + assert answer.diagnostic_label, "the answer says what was invalid" + assert handler.handled == [], "the handler itself never sees an invalid message" + assert answer in handler.outgoing_messages.values(), "the answer is kept as sent" + + +@pytest.mark.asyncio +async def test_a_valid_message_reaches_its_handler(rm_handshake): + """The positive control for the test above: the same message, unaltered, is handled.""" + handler = HandshakeHandler() + + answer = await handler.handle_message(rm_handshake.to_dict()) + + assert answer.status == ReceptionStatusValues.OK + assert len(handler.handled) == 1 + + +@pytest.mark.asyncio +async def test_a_message_can_be_given_as_json(rm_handshake): + handler = HandshakeHandler() + + answer = await handler.handle_message(rm_handshake.to_json()) + + assert answer.status == ReceptionStatusValues.OK + assert len(handler.handled) == 1 + + +def test_a_failure_callback_is_stored_with_its_arguments(): + handler = HandshakeHandler() + called = [] + + handler.register_failure_callbacks( + "some-id", lambda **kwargs: called.append(kwargs), why="test" + ) + handler.failure_callbacks["some-id"]() + + assert called == [{"why": "test"}] + assert handler.success_callbacks == {} + + +def test_a_message_given_as_json_is_recognised_as_supported(rm_handshake): + handler = HandshakeHandler() + + assert handler.supports_message(rm_handshake.to_json()) + assert not handler.supports_message( + json.dumps({"message_type": "FRBC.StorageStatus"}) + ) + + +async def _cem_with_details(resource_manager_details, rm_handshake) -> CEM: + cem = CEM(fm_client=None) + cem.register_control_type(FRBCTest()) + await cem.handle_message(rm_handshake) + await cem.handle_message(resource_manager_details) + while not cem._sending_queue.empty(): + await cem.get_message() + return cem + + +@pytest.mark.asyncio +async def test_activating_the_control_type_in_force_sends_nothing( + resource_manager_details, rm_handshake +): + cem = await _cem_with_details(resource_manager_details, rm_handshake) + assert cem.control_type == ControlType.NO_SELECTION + + answer = await cem.activate_control_type(ControlType.NO_SELECTION) + + assert answer is None + assert cem._sending_queue.empty() + + +@pytest.mark.asyncio +async def test_a_control_type_the_resource_manager_lacks_is_not_activated( + resource_manager_details, rm_handshake +): + cem = await _cem_with_details(resource_manager_details, rm_handshake) + + answer = await cem.activate_control_type(ControlType.POWER_ENVELOPE_BASED_CONTROL) + + assert answer is None + assert cem._sending_queue.empty() + assert cem.control_type == ControlType.NO_SELECTION + + +def test_registering_a_control_type_again_replaces_its_handler(caplog): + cem = CEM(fm_client=None) + first, second = FRBCTest(), FRBCTest() + cem.register_control_type(first) + + cem.register_control_type(second) + + assert cem._control_types_handlers[ControlType.FILL_RATE_BASED_CONTROL] is second + assert "already registered" in caplog.text + + +def test_an_empty_schedule_gives_no_instructions(frbc_system_description): + from flexmeasures_client.s2.control_types.FRBC.utils import ( + fm_schedule_to_instructions, + ) + + no_values = { + "start": "2026-01-01T00:00:00+00:00", + "duration": "PT0H", + "values": [], + "unit": "W", + } + no_frame = pd.DataFrame({"schedule": []}, index=pd.DatetimeIndex([], tz="UTC")) + + assert ( + fm_schedule_to_instructions( + no_values, frbc_system_description, initial_fill_level=0.0 + ) + == [] + ) + assert ( + fm_schedule_to_instructions( + no_frame, frbc_system_description, initial_fill_level=0.0 + ) + == [] + ) From 481602427545e45042c28b852cb8fda929eebee5 Mon Sep 17 00:00:00 2001 From: "F.N. Claessen" Date: Sun, 8 Nov 2026 15:46:22 +0100 Subject: [PATCH 2/2] test(FRBC): time the long-schedule translation in CPU time It measured wall-clock time against 0.5 s, and failed in 2 of 12 runs on a busy machine; the S2 job now also runs under coverage. In CPU time it takes 0.11 s, and 0.22 s under coverage, still well apart from the 1.7 s it guards against. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01CcpawsJs74cSwFLXkz7ZwW Signed-off-by: F.N. Claessen --- tests/s2/test_frbc_time_base.py | 9 ++++++--- 1 file changed, 6 insertions(+), 3 deletions(-) diff --git a/tests/s2/test_frbc_time_base.py b/tests/s2/test_frbc_time_base.py index eb940673..89c22530 100644 --- a/tests/s2/test_frbc_time_base.py +++ b/tests/s2/test_frbc_time_base.py @@ -694,11 +694,14 @@ def test_a_handler_needs_a_positive_resolution_and_duration(resolution, duration def test_a_long_schedule_against_a_detailed_forecast_is_translated_quickly(): - """48 hours, at quarter hours, against the largest forecast S2 allows: well under the 1.7 s that stepping each minute took.""" + """48 hours, at quarter hours, against the largest forecast S2 allows: well under the 1.7 s that stepping each minute took. + + Timed in CPU time, so that other processes on a busy machine do not count against it. + """ import time detailed = _usage(*[(timedelta(minutes=10), 1e-4 * (i % 3)) for i in range(288)]) - started = time.perf_counter() + started = time.process_time() fm_schedule_to_instructions( _schedule([500.0, 0.0] * 96), _system_description(LOW, HIGH), @@ -706,7 +709,7 @@ def test_a_long_schedule_against_a_detailed_forecast_is_translated_quickly(): usage_forecast=detailed, leakage_behaviour=_leakage((0, 50, 1e-4), (50, 100, 2e-4)), ) - assert time.perf_counter() - started < 0.5 + assert time.process_time() - started < 0.5 def test_a_level_held_at_its_lower_bound_rises_from_there_once_filling_wins():