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_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(): 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 + ) + == [] + )