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