From 377e6fb32323923cbcb85431bb6f83aead4e5f67 Mon Sep 17 00:00:00 2001 From: Hector van der Aa Date: Fri, 18 Sep 2026 21:27:52 +0200 Subject: [PATCH] Added playback mode ready for offline processing CoreConfig how has a playback mode option which can be set to reprocess or read. Setting it to reprocess means that all derived signals are reprocessed during playback, in this case the PlaybackEngine only plays back the signals whose source is set to "origin" Setting it to read means that all derived signals are playedback from the recorded buffer, this whill be the preferred mode once offline processing is set up which will allow for more complex data processing offline once a recording is completed, which would require the read playback mode as it wouldn't support realtime processing as reprocess does --- pyproject.toml | 2 + src/dynalab_core/__init__.py | 25 ++++- src/dynalab_core/config.py | 4 + src/dynalab_core/playback.py | 35 ++++-- test/manual/core.py | 2 +- test/manual/core_tui.py | 2 +- test/test_core.py | 134 ---------------------- test/test_derive.py | 128 --------------------- test/test_dlpak.py | 47 -------- test/test_errors.py | 150 ------------------------- test/test_json_server.py | 212 ----------------------------------- 11 files changed, 52 insertions(+), 689 deletions(-) delete mode 100644 test/test_core.py delete mode 100644 test/test_derive.py delete mode 100644 test/test_dlpak.py delete mode 100644 test/test_errors.py delete mode 100644 test/test_json_server.py diff --git a/pyproject.toml b/pyproject.toml index 67ea91f..5564505 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -8,6 +8,8 @@ authors = [ ] requires-python = ">=3.13" dependencies = [ + "psutil>=7.2.2", + "psutils>=3.3.17", "pydantic>=2.13.4", ] diff --git a/src/dynalab_core/__init__.py b/src/dynalab_core/__init__.py index ebbfe76..db8d46a 100644 --- a/src/dynalab_core/__init__.py +++ b/src/dynalab_core/__init__.py @@ -61,7 +61,9 @@ class Core: ) self._input_worker_thread.start() - self._playback_engine = PlaybackEngine(self._data_input_queue) + self._playback_engine = PlaybackEngine( + self._core_config, self._data_input_queue + ) self._derive_registry = DeriveRegistry( self._data_input_queue, self._get_live_value_descriptor @@ -332,21 +334,30 @@ class Core: signal: SignalDescriptor | None = None if self._mode.value == "realtime": signal = self._connector_registry.get_signal_descriptor(signal_id) + if signal is None: + signal = self._derive_registry.get_signal_descriptor(signal_id) + elif self._mode.value == "playback": signal = self._playback_engine.get_signal_descriptor(signal_id) - if signal is None: - signal = self._derive_registry.get_signal_descriptor(signal_id) + if signal is None and self._core_config.playback_mode == "reprocess": + signal = self._derive_registry.get_signal_descriptor(signal_id) return signal def get_all_signal_descriptors(self) -> list[SignalDescriptor]: root_signals: list[SignalDescriptor] = [] + derive_signals: list[SignalDescriptor] = [] if self._mode.value == "realtime": root_signals = self._connector_registry.get_all_signal_descriptors() elif self._mode.value == "playback": root_signals = self._playback_engine.get_all_signal_descriptors() - derive_signals = self._derive_registry.get_all_signal_descriptors() + + if ( + self._core_config.playback_mode == "reprocess" + or self._mode.value == "realtime" + ): + derive_signals = self._derive_registry.get_all_signal_descriptors() return root_signals + derive_signals def get_live_value(self, signal_id: UUID) -> float | None: @@ -467,7 +478,11 @@ class Core: message.timestamp, message.signal_id, message.value ) - self._derive_registry.put_data(message) + if ( + self._mode.value == "realtime" + or self._core_config.playback_mode == "reprocess" + ): + self._derive_registry.put_data(message) except Exception as error: log.exception( "Core input worker failed", diff --git a/src/dynalab_core/config.py b/src/dynalab_core/config.py index f170aac..9fb4ed0 100644 --- a/src/dynalab_core/config.py +++ b/src/dynalab_core/config.py @@ -2,13 +2,17 @@ # Copyright (C) 2026 Association Exergie # SPDX-License-Identifier: GPL-3.0-or-later +from typing import Literal from pydantic import BaseModel, Field +PlaybackMode = Literal["reprocess", "read"] + class CoreConfig(BaseModel): host: str = "127.0.0.1" port: int = Field(default=58763, ge=0, le=65535) thread_debug: bool = False + playback_mode: PlaybackMode = "reprocess" def bind_str(self) -> str: return f"{self.host}:{self.port}" diff --git a/src/dynalab_core/playback.py b/src/dynalab_core/playback.py index 5859290..eb07683 100644 --- a/src/dynalab_core/playback.py +++ b/src/dynalab_core/playback.py @@ -9,6 +9,7 @@ import threading import time from uuid import UUID +from dynalab_core.config import CoreConfig from dynalab_core.dlpak import DLPak from dynalab_core.errors import PlaybackError from dynalab_core.protocols.packets import ProtocolMessage @@ -19,7 +20,10 @@ log = logging.getLogger(__name__) class PlaybackEngine: - def __init__(self, data_queue: Queue[ProtocolMessage]) -> None: + def __init__( + self, core_config: CoreConfig, data_queue: Queue[ProtocolMessage] + ) -> None: + self._core_config = core_config self._data: DLPak | None = None self._data_queue: Queue[ProtocolMessage] = data_queue self._playing: threading.Event = threading.Event() @@ -65,16 +69,24 @@ class PlaybackEngine: if self._data is not None and self._data._manifest is not None: for signal in self._data._manifest.signals: if signal.id == uuid: + if ( + signal.origin == "derived" + and self._core_config.playback_mode == "reprocess" + ): + return None return signal return None def get_all_signal_descriptors(self) -> list[SignalDescriptor]: if self._data is not None and self._data._manifest is not None: - return [ - signal - for signal in self._data._manifest.signals - if signal.origin != "derived" - ] + if self._core_config.playback_mode == "reprocess": + return [ + signal + for signal in self._data._manifest.signals + if signal.origin != "derived" + ] + elif self._core_config.playback_mode == "read": + return self._data._manifest.signals return list[SignalDescriptor]() @@ -90,11 +102,12 @@ class PlaybackEngine: code="incomplete_data", ) self._data = data - self._ignored_signals = { - signal.id - for signal in self._data._manifest.signals - if signal.origin == "derived" - } + if self._core_config.playback_mode == "reprocess": + self._ignored_signals = { + signal.id + for signal in self._data._manifest.signals + if signal.origin == "derived" + } def _update_internal_timestamp(self) -> None: self._internal_timestamp = time.monotonic_ns() - self._start_timestamp diff --git a/test/manual/core.py b/test/manual/core.py index 8f9de07..c3148fa 100644 --- a/test/manual/core.py +++ b/test/manual/core.py @@ -25,7 +25,7 @@ processing_str = Path( signal_id_1: UUID = UUID("3f32cea3-d872-4c16-a0a2-54b57171aeb2") signal_id_2: UUID = UUID("6578ac37-99d1-410c-b3f7-d049339919b2") -config = CoreConfig(port=8765) +config = CoreConfig(port=8765, thread_debug=True) log.info("Configured manual core on %s", config.bind_str()) dl_core = Core(config) dl_core.start() diff --git a/test/manual/core_tui.py b/test/manual/core_tui.py index 1c031a0..010955e 100644 --- a/test/manual/core_tui.py +++ b/test/manual/core_tui.py @@ -273,7 +273,7 @@ def main() -> None: dynalab_logger.setLevel(logging.DEBUG) dynalab_logger.propagate = False dynalab_logger.addHandler(log_handler) - config = CoreConfig(host=HOST, port=PORT, thread_debug=False) + config = CoreConfig(host=HOST, port=PORT, playback_mode="read") core = Core(config) processing_source = ( Path(__file__).with_name("process.py").read_text(encoding="utf-8") diff --git a/test/test_core.py b/test/test_core.py deleted file mode 100644 index 5b9306e..0000000 --- a/test/test_core.py +++ /dev/null @@ -1,134 +0,0 @@ -import logging -from pathlib import Path -import threading -from uuid import uuid4 - -import pytest -from dynalab_core import Core -from dynalab_core.config import CoreConfig -from dynalab_core.dlunit import DLUnit -from dynalab_core.errors import CoreError -from dynalab_core.protocols.endpoint import ConnectorEndpoint -from dynalab_core.protocols.packets.data import ValueDescriptor -from dynalab_core.protocols.packets.handshake import ConnectorHello, SignalDescriptor -from test.common import find_available_port - - -def test_core_cannot_start_twice(caplog: pytest.LogCaptureFixture) -> None: - port = find_available_port(8765) - core = Core(CoreConfig(port=port)) - - with caplog.at_level(logging.DEBUG, logger="dynalab_core"): - core.start() - - try: - with pytest.raises(CoreError, match="can only be started once") as raised: - core.start() - finally: - core.stop() - - events = [getattr(record, "event", None) for record in caplog.records] - assert events.count("core.starting") == 1 - assert "core.start_rejected" in events - assert "core.stopped" in events - - rejection = next( - record - for record in caplog.records - if getattr(record, "event", None) == "core.start_rejected" - ) - assert rejection.levelno == logging.WARNING - assert rejection.core_state == "started" - assert raised.value.code == "invalid_state" - - -def test_core_input_worker_logs_failure( - caplog: pytest.LogCaptureFixture, monkeypatch: pytest.MonkeyPatch -) -> None: - core = Core(CoreConfig(port=find_available_port(8765))) - - def fail_routing(message: ValueDescriptor) -> None: - raise RuntimeError("routing failed") - - monkeypatch.setattr(core._derive_registry, "put_data", fail_routing) - - with caplog.at_level(logging.DEBUG, logger="dynalab_core"): - core.start() - try: - core._data_input_queue.put_nowait( - ValueDescriptor(signal_id=uuid4(), value=1.0, timestamp=1) - ) - assert core._input_worker_stopped_event.wait(1.0) - finally: - core.stop() - - failures = [ - record - for record in caplog.records - if getattr(record, "event", None) == "core.input_worker_failed" - ] - assert len(failures) == 1 - assert failures[0].exception_type == "RuntimeError" - assert failures[0].message_type == "ValueDescriptor" - assert failures[0].exc_info is not None - - -def test_core_mode_is_shared_with_endpoints() -> None: - core = Core(CoreConfig(port=find_available_port(8765))) - endpoint = ConnectorEndpoint( - ConnectorHello( - connector_uuid=uuid4(), - connector_name="test", - connector_version="1.0", - signals=[], - ), - threading.Event(), - threading.Event(), - core._data_input_queue, - core._connector_registry._core_mode, - ) - - try: - assert endpoint._core_mode is core._mode - assert core.get_mode() == "realtime" - - core.set_mode("playback") - - assert endpoint._core_mode.value == "playback" - assert core.get_mode() == "playback" - with pytest.raises(CoreError) as raised: - core.load_playback_engine() - assert raised.value.code == "no_playback_data" - finally: - core.stop() - - -def test_core_binds_in_memory_dlunit() -> None: - core = Core(CoreConfig(port=find_available_port(8765))) - input_signal = SignalDescriptor(id=uuid4(), name="Input", type="number") - return_signal = SignalDescriptor( - id=uuid4(), name="Output", type="number", origin="derived" - ) - dlunit = DLUnit() - dlunit.set_manifest(uuid4(), [input_signal], return_signal) - dlunit.set_data( - """ -from dynalab_core.protocols.packets.data import ValueDescriptor -from dynalab_core.protocols.packets.handshake import SignalDescriptor - -def identity(value: ValueDescriptor, output: SignalDescriptor) -> ValueDescriptor: - return ValueDescriptor(signal_id=output.id, value=value.value, timestamp=value.timestamp) -""" - ) - - try: - unit_id = core.bind_derive_unit(dlunit) - assert core.get_signal_descriptor(return_signal.id) == return_signal - - with pytest.raises(CoreError, match="either a DLUnit or a file") as raised: - core.bind_derive_unit(dlunit, file=Path("unit.dlunit")) - - assert raised.value.code == "bind_derive_failed" - core.unbind_derive_unit(unit_id) - finally: - core.stop() diff --git a/test/test_derive.py b/test/test_derive.py deleted file mode 100644 index 3a65780..0000000 --- a/test/test_derive.py +++ /dev/null @@ -1,128 +0,0 @@ -import logging -from queue import Empty, Queue -import time -from uuid import UUID, uuid4 - -import pytest -from dynalab_core.derive import DeriveRegistry -from dynalab_core.dlunit import DLUnit -from dynalab_core.protocols.packets.data import ValueDescriptor -from dynalab_core.protocols.packets.handshake import SignalDescriptor - - -PROCESSING_FUNCTION = """ -from dynalab_core.protocols.packets.data import ValueDescriptor -from dynalab_core.protocols.packets.handshake import SignalDescriptor - -def add(left: ValueDescriptor, right: ValueDescriptor, output: SignalDescriptor) -> ValueDescriptor: - return ValueDescriptor( - signal_id=output.id, - value=left.value + right.value, - timestamp=left.timestamp, - ) -""" - -FAILING_PROCESSING_FUNCTION = """ -from dynalab_core.protocols.packets.data import ValueDescriptor -from dynalab_core.protocols.packets.handshake import SignalDescriptor - -def fail(value: ValueDescriptor, output: SignalDescriptor) -> ValueDescriptor: - raise RuntimeError("derive failed") -""" - - -def make_dlunit( - source: str, - input_signals: list[SignalDescriptor], - return_signal: SignalDescriptor, -) -> DLUnit: - dlunit = DLUnit() - dlunit.set_manifest(uuid4(), input_signals, return_signal) - dlunit.set_data(source) - return dlunit - - -def test_derive_registry_routes_with_live_values() -> None: - output_queue: Queue[ValueDescriptor] = Queue() - left_signal = SignalDescriptor(id=uuid4(), name="Left", type="number") - right_signal = SignalDescriptor(id=uuid4(), name="Right", type="number") - output_signal = SignalDescriptor(id=uuid4(), name="Total", type="number") - live_values: dict[UUID, ValueDescriptor] = {} - registry = DeriveRegistry(output_queue, live_values.get) - - try: - unit = registry.register( - make_dlunit( - PROCESSING_FUNCTION, - [left_signal, right_signal], - output_signal, - ) - ) - right_value = ValueDescriptor(signal_id=right_signal.id, value=2.0, timestamp=1) - live_values[right_signal.id] = right_value - left_value = ValueDescriptor(signal_id=left_signal.id, value=3.0, timestamp=2) - - registry.put_data(left_value) - - deadline = time.monotonic() + 1.0 - while True: - try: - result = output_queue.get_nowait() - break - except Empty: - if time.monotonic() >= deadline: - raise AssertionError("Derive unit did not produce a value") - time.sleep(0.01) - - registry.unregister(unit.uuid()) - assert registry.get(unit.uuid()) is None - finally: - registry.stop() - - assert result.signal_id == output_signal.id - assert result.value == 5.0 - assert result.timestamp == left_value.timestamp - - -def test_derive_registry_ignores_unknown_unit_on_unregister() -> None: - registry = DeriveRegistry(Queue(), lambda signal_id: None) - - try: - registry.unregister(uuid4()) - finally: - registry.stop() - - -def test_derive_worker_logs_failure_and_stops( - caplog: pytest.LogCaptureFixture, -) -> None: - input_signal = SignalDescriptor(id=uuid4(), name="Input", type="number") - output_signal = SignalDescriptor(id=uuid4(), name="Output", type="number") - registry = DeriveRegistry(Queue(), lambda signal_id: None) - - with caplog.at_level(logging.DEBUG, logger="dynalab_core"): - try: - unit = registry.register( - make_dlunit( - FAILING_PROCESSING_FUNCTION, - [input_signal], - output_signal, - ) - ) - unit.put_data( - [ValueDescriptor(signal_id=input_signal.id, value=1.0, timestamp=1)] - ) - assert unit._stopped_event.wait(1.0) - assert registry.get_all_signal_descriptors() == [] - finally: - registry.stop() - - failures = [ - record - for record in caplog.records - if getattr(record, "event", None) == "derive.worker_failed" - ] - assert len(failures) == 1 - assert failures[0].derive_uuid == str(unit.uuid()) - assert failures[0].exception_type == "RuntimeError" - assert failures[0].exc_info is not None diff --git a/test/test_dlpak.py b/test/test_dlpak.py deleted file mode 100644 index 6a1b678..0000000 --- a/test/test_dlpak.py +++ /dev/null @@ -1,47 +0,0 @@ -import logging -from datetime import datetime, timezone -from uuid import uuid4 - -import pytest -from dynalab_core.buffer import ValueBuffer -from dynalab_core.dlpak import DLPak -from dynalab_core.protocols.packets.handshake import SignalDescriptor - - -def test_dlpak_logs_written_archive(caplog: pytest.LogCaptureFixture, tmp_path) -> None: - signal = SignalDescriptor(id=uuid4(), name="Signal", type="number") - buffer = ValueBuffer() - buffer.append(1, signal.id, 2.0) - package = DLPak() - package.set_data(buffer) - package.set_manifest(datetime.now(timezone.utc), [signal]) - - with caplog.at_level(logging.INFO, logger="dynalab_core"): - package.write(tmp_path, "recording") - - record = next( - record - for record in caplog.records - if getattr(record, "event", None) == "dlpak.written" - ) - assert record.output_filename == "recording.dlpak" - assert record.sample_count == 1 - assert record.signal_count == 1 - - -def test_dlpak_read_restores_manifest_and_data(tmp_path) -> None: - timestamp = datetime.now(timezone.utc) - signal = SignalDescriptor(id=uuid4(), name="Signal", type="number") - buffer = ValueBuffer() - buffer.append(1, signal.id, 2.0) - package = DLPak() - package.set_data(buffer) - package.set_manifest(timestamp, [signal]) - package.write(tmp_path, "recording") - - loaded_package = DLPak() - loaded_package.read(tmp_path / "recording.dlpak") - - assert loaded_package._manifest == package._manifest - assert loaded_package._data is not None - assert loaded_package._data.export_csv() == buffer.export_csv() diff --git a/test/test_errors.py b/test/test_errors.py deleted file mode 100644 index 4ce88fc..0000000 --- a/test/test_errors.py +++ /dev/null @@ -1,150 +0,0 @@ -from datetime import datetime, timezone -from queue import Queue -from uuid import uuid4 -from zipfile import BadZipFile - -import pytest - -from dynalab_core.buffer import ValueBuffer -from dynalab_core.derive import DeriveRegistry, DeriveUnit -from dynalab_core.dlpak import DLPak -from dynalab_core.dlunit import DLUnit -from dynalab_core.errors import ( - CoreError, - DLPakError, - DeriveError, - DynaLabError, - PlaybackError, -) -from dynalab_core.playback import PlaybackEngine -from dynalab_core.protocols.packets.data import ValueDescriptor -from dynalab_core.protocols.packets.handshake import SignalDescriptor - - -def _identity( - value: ValueDescriptor, output: SignalDescriptor -) -> ValueDescriptor: - return ValueDescriptor( - signal_id=output.id, - value=value.value, - timestamp=value.timestamp, - ) - - -def test_subsystem_errors_share_public_base_and_code() -> None: - error = PlaybackError("Playback data is unavailable", code="no_data") - - assert isinstance(error, DynaLabError) - assert error.code == "no_data" - assert str(error) == "Playback data is unavailable" - - -def test_core_rejects_invalid_mode_and_missing_recording() -> None: - from dynalab_core import Core - - core = Core.__new__(Core) - core._processing_buffer = None - - with pytest.raises(CoreError, match="Unknown core mode") as mode: - core.set_mode("invalid") # type: ignore[arg-type] - with pytest.raises(CoreError, match="no recording") as recording: - core.write(".", "unused") - - assert mode.value.code == "invalid_mode" - assert recording.value.code == "no_recording" - - -def test_dlpak_requires_data_before_manifest() -> None: - package = DLPak() - - with pytest.raises(DLPakError, match="before setting its data") as raised: - package.set_manifest(datetime.now(timezone.utc), []) - - assert raised.value.code == "no_data" - - -def test_dlpak_wraps_archive_read_failures(tmp_path) -> None: - invalid_archive = tmp_path / "invalid.dlpak" - invalid_archive.write_text("not a zip archive") - - with pytest.raises(DLPakError, match="Could not read") as raised: - DLPak().read(invalid_archive) - - assert raised.value.code == "read_failed" - assert isinstance(raised.value.__cause__, BadZipFile) - - -def test_derive_unit_reports_argument_mismatch() -> None: - input_signal = SignalDescriptor(id=uuid4(), name="Input", type="number") - output_signal = SignalDescriptor( - id=uuid4(), name="Output", type="number", origin="derived" - ) - unit = DeriveUnit(uuid4(), _identity, [input_signal], output_signal, Queue()) - - try: - with pytest.raises(DeriveError, match="expected 1 input") as raised: - unit.process_offline([]) - finally: - unit.stop() - - assert raised.value.code == "argument_mismatch" - - -def test_derive_registry_wraps_invalid_source() -> None: - registry = DeriveRegistry(Queue(), lambda signal_id: None) - output_signal = SignalDescriptor( - id=uuid4(), name="Output", type="number", origin="derived" - ) - - try: - with pytest.raises(DeriveError, match="Could not load") as raised: - dlunit = DLUnit() - dlunit.set_manifest(uuid4(), [], output_signal) - dlunit.set_data("def broken(") - registry.register(dlunit) - finally: - registry.stop() - - assert raised.value.code == "invalid_source" - assert isinstance(raised.value.__cause__, SyntaxError) - - -def test_derive_unit_rejects_configured_input_mismatch() -> None: - output_signal = SignalDescriptor( - id=uuid4(), name="Output", type="number", origin="derived" - ) - - with pytest.raises(DeriveError, match="input signal") as raised: - DeriveUnit(uuid4(), _identity, [], output_signal, Queue()) - - assert raised.value.code == "input_signal_mismatch" - - -def test_playback_rejects_incomplete_data_and_negative_seek() -> None: - engine = PlaybackEngine(Queue()) - - try: - with pytest.raises(PlaybackError, match="incomplete") as incomplete: - engine.set_data(DLPak()) - with pytest.raises(PlaybackError, match="cannot be negative") as timestamp: - engine.seek(-1) - finally: - engine.stop() - - assert incomplete.value.code == "incomplete_data" - assert timestamp.value.code == "invalid_timestamp" - - -def test_playback_accepts_complete_dlpak() -> None: - signal = SignalDescriptor(id=uuid4(), name="Input", type="number") - data = ValueBuffer() - data.append(1, signal.id, 2.0) - package = DLPak() - package.set_data(data) - package.set_manifest(datetime.now(timezone.utc), [signal]) - engine = PlaybackEngine(Queue()) - - try: - engine.set_data(package) - finally: - engine.stop() diff --git a/test/test_json_server.py b/test/test_json_server.py deleted file mode 100644 index 074ab4e..0000000 --- a/test/test_json_server.py +++ /dev/null @@ -1,212 +0,0 @@ -import asyncio -import logging -import socket -import time -from uuid import uuid4 - -import pytest -from dynalab_core import Core -from dynalab_core.config import CoreConfig -from dynalab_core.errors import JsonServerError -from dynalab_core.protocols.constants import PROTOCOL_VERSION -from dynalab_core.protocols.common import VersionDescriptor -from dynalab_core.protocols.packets import JsonConnectorHello -from dynalab_core.protocols.packets.handshake import ( - ConnectorHello, - JsonSignalDescriptor, - SignalDescriptor, -) -from test.common import find_available_port - - -def _wait_for_event( - caplog: pytest.LogCaptureFixture, event: str, timeout: float = 1.0 -) -> logging.LogRecord: - deadline = time.monotonic() + timeout - while time.monotonic() < deadline: - for record in caplog.records: - if getattr(record, "event", None) == event: - return record - time.sleep(0.01) - raise AssertionError(f"Log event {event!r} was not emitted") - - -def test_json_server_raises_timeout_error( - caplog: pytest.LogCaptureFixture, -) -> None: - port = find_available_port(8765) - core = Core(CoreConfig(port=port)) - core._json_server._debug_timeout_test = True - core._json_server._timeout = 0.01 - - with caplog.at_level(logging.DEBUG, logger="dynalab_core"): - try: - with pytest.raises(JsonServerError, match="did not start") as raised: - core.start() - finally: - core.stop() - - timeout_record = next( - record - for record in caplog.records - if getattr(record, "event", None) == "json_server.start_timeout" - ) - assert timeout_record.levelno == logging.ERROR - assert timeout_record.port == port - assert raised.value.code == "start_timeout" - - -def test_json_server_raises_startup_error( - caplog: pytest.LogCaptureFixture, -) -> None: - - port = find_available_port(8765) - core1 = Core(CoreConfig(port=port)) - core2 = Core(CoreConfig(port=port)) - - with caplog.at_level(logging.DEBUG, logger="dynalab_core"): - core1.start() - - try: - with pytest.raises(JsonServerError, match="failed to start") as raised: - core2.start() - finally: - core1.stop() - core2.stop() - - failure_record = next( - record - for record in caplog.records - if getattr(record, "event", None) == "json_server.start_failed" - ) - assert failure_record.levelno == logging.ERROR - assert failure_record.port == port - assert failure_record.exc_info is not None - assert raised.value.code == "startup_failed" - assert isinstance(raised.value.__cause__, OSError) - - -def test_json_server_logs_invalid_handshake( - caplog: pytest.LogCaptureFixture, -) -> None: - port = find_available_port(8765) - core = Core(CoreConfig(port=port)) - - with caplog.at_level(logging.DEBUG, logger="dynalab_core"): - core.start() - try: - with socket.create_connection(("127.0.0.1", port), timeout=1) as peer: - server_hello = peer.makefile("rb").readline() - assert server_hello - peer.sendall(b'{"secret":"must-not-appear-in-logs"}\n') - - rejection = _wait_for_event(caplog, "connector.handshake_rejected") - finally: - core.stop() - - assert rejection.levelno == logging.WARNING - assert rejection.reason == "invalid_frame" - assert rejection.connection_id - assert rejection.peer_address - assert "must-not-appear-in-logs" not in caplog.text - - -def test_json_server_reports_unexpected_thread_startup_failure( - caplog: pytest.LogCaptureFixture, monkeypatch: pytest.MonkeyPatch -) -> None: - async def fail_start_server(*args: object, **kwargs: object) -> None: - raise RuntimeError("unexpected startup failure") - - monkeypatch.setattr(asyncio, "start_server", fail_start_server) - core = Core(CoreConfig(port=find_available_port(8765))) - - with caplog.at_level(logging.DEBUG, logger="dynalab_core"): - try: - with pytest.raises(JsonServerError) as raised: - core.start() - finally: - core.stop() - - failures = [ - record - for record in caplog.records - if getattr(record, "event", None) == "json_server.thread_failed" - ] - assert len(failures) == 1 - assert failures[0].startup_complete is False - assert failures[0].exception_type == "RuntimeError" - assert failures[0].exc_info is not None - assert raised.value.code == "startup_failed" - assert isinstance(raised.value.__cause__, RuntimeError) - - -def test_rejected_handshake_is_not_logged_as_connected( - caplog: pytest.LogCaptureFixture, -) -> None: - port = find_available_port(8765) - core = Core(CoreConfig(port=port)) - connector_uuid = uuid4() - connector_hello = JsonConnectorHello( - connector_uuid=connector_uuid, - protocol_version=VersionDescriptor(type="alpha", major=999, minor=0, patch=0), - connector_name="Incompatible connector", - connector_version="0.1.0-test", - signals=[JsonSignalDescriptor(id=uuid4(), name="Dummy signal", type="number")], - ) - - with caplog.at_level(logging.DEBUG, logger="dynalab_core"): - core.start() - try: - with socket.create_connection(("127.0.0.1", port), timeout=1) as peer: - peer_file = peer.makefile("rb") - assert peer_file.readline() - peer.sendall(connector_hello.model_dump_json().encode("utf-8") + b"\n") - assert b'"type":"handshake_rejected"' in peer_file.readline() - rejection = _wait_for_event(caplog, "endpoint.handshake_rejected") - finally: - core.stop() - - assert rejection.reason == "protocol_version_mismatch" - assert rejection.connector_uuid == str(connector_uuid) - assert not any( - getattr(record, "event", None) == "connector.connected" - and getattr(record, "connector_uuid", None) == str(connector_uuid) - for record in caplog.records - ) - - -def test_json_server_waits_for_connection_handlers_on_stop( - caplog: pytest.LogCaptureFixture, -) -> None: - port = find_available_port(8765) - core = Core(CoreConfig(port=port)) - connector_uuid = uuid4() - connector_hello = JsonConnectorHello( - connector_uuid=connector_uuid, - protocol_version=PROTOCOL_VERSION, - connector_name="Test connector", - connector_version="0.1.0-test", - signals=[JsonSignalDescriptor(id=uuid4(), name="Dummy signal", type="number")], - ) - - with caplog.at_level(logging.DEBUG, logger="dynalab_core"): - core.start() - try: - with socket.create_connection(("127.0.0.1", port), timeout=1) as peer: - peer_file = peer.makefile("rb") - assert peer_file.readline() - peer.sendall(connector_hello.model_dump_json().encode("utf-8") + b"\n") - assert b'"type":"handshake_accepted"' in peer_file.readline() - - core.stop() - finally: - if core._state != "stopped": - core.stop() - - events = [getattr(record, "event", None) for record in caplog.records] - assert "json_server.stop_timeout" not in events - assert "connector.connected" in events - assert events.index("connector.registered") < events.index("connector.connected") - assert core._json_server._stopped_event.is_set() - assert not core._json_server._handler_tasks - assert core._connector_registry.get(connector_uuid) is None