test(S2): coverage back above main's, before the badge counts S2

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

tests/s2/test_frbc_time_base.py

@@ -694,19 +694,22 @@ 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),
         initial_fill_level=45.0,
         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():

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