Add async read-only proxy client

custom_components/arkteos/client.py

@@ -0,0 +1,288 @@
+"""Client TCP asynchrone, strictement en lecture seule, pour le proxy Arkteos."""
+
+from __future__ import annotations
+
+import asyncio
+import inspect
+from collections.abc import Awaitable, Callable, Sequence
+from contextlib import suppress
+from typing import Final, TypeAlias
+
+from .frame_extractor import FrameExtractor
+from .parser import FilterResult, FrameError, ParsedFrame, filter_values, parse_frame
+
+
+DEFAULT_PROXY_PORT: Final = 9641
+DEFAULT_RECONNECT_DELAYS: Final[tuple[float, ...]] = (1.0, 2.0, 5.0, 10.0, 30.0)
+
+DataCallback: TypeAlias = Callable[
+    [str, dict[str, int | float | str], dict[str, int | float]], object
+]
+AvailabilityCallback: TypeAlias = Callable[[bool], object]
+SleepCallable: TypeAlias = Callable[[float], Awaitable[None]]
+
+
+class ArkteosClient:
+    """Lit le proxy Arkteos, extrait les trames et diffuse les données filtrées."""
+
+    def __init__(
+        self,
+        host: str,
+        port: int = DEFAULT_PROXY_PORT,
+        *,
+        read_size: int = 1024,
+        frame_timeout: float = 30.0,
+        reconnect_delays: Sequence[float] = DEFAULT_RECONNECT_DELAYS,
+        max_buffer_size: int = 4096,
+        sleep: SleepCallable = asyncio.sleep,
+    ) -> None:
+        if not host:
+            raise ValueError("host ne peut pas être vide")
+        if not 1 <= port <= 65535:
+            raise ValueError("port doit être compris entre 1 et 65535")
+        if read_size < 1:
+            raise ValueError("read_size doit être positif")
+        if frame_timeout <= 0:
+            raise ValueError("frame_timeout doit être positif")
+        if not reconnect_delays or any(delay < 0 for delay in reconnect_delays):
+            raise ValueError("reconnect_delays doit contenir des délais positifs ou nuls")
+
+        self.host = host
+        self.port = port
+        self.read_size = read_size
+        self.frame_timeout = frame_timeout
+        self.reconnect_delays = tuple(reconnect_delays)
+
+        self.running = False
+        self.connected = False
+        self.available = False
+        self.last_valid_frame: ParsedFrame | None = None
+        self.last_frame_type: str | None = None
+        self.latest_frigo_data: dict[str, int | float | str] | None = None
+        self.latest_regulation_data: dict[str, int | float | str] | None = None
+        self.latest_metadata: dict[str, int | float | str] | None = None
+        self.frames_received = 0
+        self.frames_rejected = 0
+        self.bytes_received = 0
+        self.connection_attempts = 0
+        self.reconnect_count = 0
+        self.last_error: Exception | None = None
+
+        self._extractor = FrameExtractor(max_buffer_size=max_buffer_size)
+        self._sleep = sleep
+        self._task: asyncio.Task[None] | None = None
+        self._writer: asyncio.StreamWriter | None = None
+        self._connected_event = asyncio.Event()
+        self._data_callbacks: list[DataCallback] = []
+        self._availability_callbacks: list[AvailabilityCallback] = []
+
+    async def start(self) -> None:
+        """Démarre la tâche de connexion si le client n'est pas déjà actif."""
+
+        if self.running:
+            return
+
+        self.running = True
+        self._task = asyncio.create_task(self._run(), name="arkteos-client")
+
+    async def stop(self) -> None:
+        """Arrête le client, ferme le flux et attend toutes ses tâches."""
+
+        self.running = False
+        self._set_connected(False)
+        await self._set_available(False)
+        await self._close_writer()
+
+        task = self._task
+        self._task = None
+        if task is not None and task is not asyncio.current_task() and not task.done():
+            task.cancel()
+            with suppress(asyncio.CancelledError):
+                await task
+
+    async def wait_until_connected(self, timeout: float | None = None) -> bool:
+        """Attend une connexion établie, avec délai optionnel."""
+
+        if self.connected:
+            return True
+        try:
+            if timeout is None:
+                await self._connected_event.wait()
+            else:
+                await asyncio.wait_for(self._connected_event.wait(), timeout)
+        except TimeoutError:
+            return False
+        return self.connected
+
+    def add_data_callback(self, callback: DataCallback) -> None:
+        """Ajoute un callback de données sans le dupliquer."""
+
+        if callback not in self._data_callbacks:
+            self._data_callbacks.append(callback)
+
+    def remove_data_callback(self, callback: DataCallback) -> None:
+        """Retire un callback de données s'il est enregistré."""
+
+        with suppress(ValueError):
+            self._data_callbacks.remove(callback)
+
+    def add_availability_callback(self, callback: AvailabilityCallback) -> None:
+        """Ajoute un callback de disponibilité sans le dupliquer."""
+
+        if callback not in self._availability_callbacks:
+            self._availability_callbacks.append(callback)
+
+    def remove_availability_callback(self, callback: AvailabilityCallback) -> None:
+        """Retire un callback de disponibilité s'il est enregistré."""
+
+        with suppress(ValueError):
+            self._availability_callbacks.remove(callback)
+
+    async def _run(self) -> None:
+        delay_index = 0
+        try:
+            while self.running:
+                self.connection_attempts += 1
+                try:
+                    reader, writer = await asyncio.open_connection(self.host, self.port)
+                except asyncio.CancelledError:
+                    raise
+                except Exception as error:
+                    self.last_error = error
+                    self._set_connected(False)
+                    await self._set_available(False)
+                    if not await self._wait_before_reconnect(delay_index):
+                        break
+                    delay_index = min(delay_index + 1, len(self.reconnect_delays) - 1)
+                    continue
+
+                self._writer = writer
+                self._set_connected(True)
+                received_valid_frame = await self._read_connection(reader)
+                if received_valid_frame:
+                    delay_index = 0
+
+                self._set_connected(False)
+                await self._set_available(False)
+                await self._close_writer()
+                if not self.running:
+                    break
+                if not await self._wait_before_reconnect(delay_index):
+                    break
+                delay_index = min(delay_index + 1, len(self.reconnect_delays) - 1)
+        except asyncio.CancelledError:
+            raise
+        finally:
+            self._set_connected(False)
+            await self._set_available(False)
+            await self._close_writer()
+
+    async def _wait_before_reconnect(self, delay_index: int) -> bool:
+        if not self.running:
+            return False
+        self.reconnect_count += 1
+        await self._sleep(self.reconnect_delays[delay_index])
+        return self.running
+
+    async def _read_connection(self, reader: asyncio.StreamReader) -> bool:
+        received_valid_frame = False
+        loop = asyncio.get_running_loop()
+        deadline = loop.time() + self.frame_timeout
+
+        while self.running:
+            remaining = deadline - loop.time()
+            if remaining <= 0:
+                await self._set_available(False)
+                deadline = loop.time() + self.frame_timeout
+                remaining = self.frame_timeout
+
+            try:
+                data = await asyncio.wait_for(reader.read(self.read_size), remaining)
+            except asyncio.CancelledError:
+                raise
+            except TimeoutError:
+                await self._set_available(False)
+                deadline = loop.time() + self.frame_timeout
+                continue
+            except Exception as error:
+                self.last_error = error
+                return received_valid_frame
+
+            if not data:
+                return received_valid_frame
+
+            self.bytes_received += len(data)
+            try:
+                raw_frames = self._extractor.feed(data)
+            except Exception as error:
+                self.frames_rejected += 1
+                self.last_error = error
+                continue
+
+            for raw_frame in raw_frames:
+                try:
+                    frame = parse_frame(raw_frame)
+                    filtered = filter_values(frame)
+                except FrameError as error:
+                    self.frames_rejected += 1
+                    self.last_error = error
+                    continue
+
+                received_valid_frame = True
+                deadline = loop.time() + self.frame_timeout
+                await self._accept_frame(frame, filtered)
+
+        return received_valid_frame
+
+    async def _accept_frame(self, frame: ParsedFrame, filtered: FilterResult) -> None:
+        accepted = dict(filtered.accepted)
+        rejected = dict(filtered.rejected)
+        frame_type = accepted["frame_type"]
+        assert isinstance(frame_type, str)
+
+        self.frames_received += 1
+        self.last_valid_frame = frame
+        self.last_frame_type = frame_type
+        if frame_type == "frigo":
+            self.latest_frigo_data = accepted
+        elif frame_type == "regulation":
+            self.latest_regulation_data = accepted
+        else:
+            self.latest_metadata = accepted
+
+        await self._set_available(True)
+        for callback in tuple(self._data_callbacks):
+            await self._invoke_callback(callback, frame_type, dict(accepted), dict(rejected))
+
+    async def _set_available(self, value: bool) -> None:
+        if self.available == value:
+            return
+        self.available = value
+        for callback in tuple(self._availability_callbacks):
+            await self._invoke_callback(callback, value)
+
+    async def _invoke_callback(self, callback: Callable[..., object], *args: object) -> None:
+        try:
+            result = callback(*args)
+            if inspect.isawaitable(result):
+                await result
+        except Exception as error:
+            self.last_error = error
+
+    def _set_connected(self, value: bool) -> None:
+        self.connected = value
+        if value:
+            self._connected_event.set()
+        else:
+            self._connected_event.clear()
+
+    async def _close_writer(self) -> None:
+        writer = self._writer
+        self._writer = None
+        if writer is None:
+            return
+        writer.close()
+        try:
+            await writer.wait_closed()
+        except Exception as error:
+            self.last_error = error

tests/test_client.py

@@ -0,0 +1,523 @@
+"""Tests sans réseau réel du client TCP asynchrone Arkteos."""
+
+import asyncio
+from collections import deque
+from pathlib import Path
+from typing import Deque
+
+import pytest
+
+from custom_components.arkteos.client import ArkteosClient, DEFAULT_PROXY_PORT
+
+
+FIXTURES_PATH = Path(__file__).with_name("fixtures")
+
+
+class FakeWriter:
+    def __init__(self) -> None:
+        self.closed = False
+        self.wait_closed_calls = 0
+
+    def close(self) -> None:
+        self.closed = True
+
+    async def wait_closed(self) -> None:
+        self.wait_closed_calls += 1
+
+
+class ControlledReader:
+    def __init__(self, chunks: list[bytes] | None = None) -> None:
+        self._chunks: Deque[bytes] = deque(chunks or [])
+        self._data_ready = asyncio.Event()
+        if self._chunks:
+            self._data_ready.set()
+
+    def push(self, data: bytes) -> None:
+        self._chunks.append(data)
+        self._data_ready.set()
+
+    async def read(self, _size: int) -> bytes:
+        while not self._chunks:
+            await self._data_ready.wait()
+            self._data_ready.clear()
+        return self._chunks.popleft()
+
+
+def _load_frames() -> dict[str, bytes]:
+    return {
+        "metadata": (FIXTURES_PATH / "metadata_95.bin").read_bytes(),
+        "frigo": (FIXTURES_PATH / "frigo_163.bin").read_bytes(),
+        "regulation": (FIXTURES_PATH / "regulation_227.bin").read_bytes(),
+    }
+
+
+async def _wait_for(predicate: object, attempts: int = 200) -> None:
+    for _ in range(attempts):
+        if callable(predicate) and predicate():
+            return
+        await asyncio.sleep(0.001)
+    raise AssertionError("condition asynchrone non atteinte")
+
+
+def _connection_factory(
+    monkeypatch: pytest.MonkeyPatch,
+    outcomes: list[tuple[ControlledReader, FakeWriter] | Exception],
+) -> None:
+    pending = deque(outcomes)
+
+    async def open_connection(_host: str, _port: int) -> tuple[ControlledReader, FakeWriter]:
+        if not pending:
+            raise ConnectionRefusedError("aucune connexion factice disponible")
+        outcome = pending.popleft()
+        if isinstance(outcome, Exception):
+            raise outcome
+        return outcome
+
+    monkeypatch.setattr(asyncio, "open_connection", open_connection)
+
+
+def test_default_port() -> None:
+    assert ArkteosClient("proxy.example").port == DEFAULT_PROXY_PORT
+
+
+def test_successful_connection(monkeypatch: pytest.MonkeyPatch) -> None:
+    async def scenario() -> None:
+        reader = ControlledReader()
+        writer = FakeWriter()
+        _connection_factory(monkeypatch, [(reader, writer)])
+        client = ArkteosClient("proxy.example")
+        await client.start()
+        assert await client.wait_until_connected(0.1)
+        await client.stop()
+        assert writer.closed
+
+    asyncio.run(scenario())
+
+
+def test_connection_refused(monkeypatch: pytest.MonkeyPatch) -> None:
+    async def scenario() -> None:
+        _connection_factory(monkeypatch, [ConnectionRefusedError("refus")])
+
+        async def pause(_delay: float) -> None:
+            await asyncio.sleep(0)
+
+        client = ArkteosClient("proxy.example", reconnect_delays=(1,), sleep=pause)
+        await client.start()
+        await _wait_for(lambda: client.connection_attempts >= 1)
+        await client.stop()
+        assert not client.connected
+        assert isinstance(client.last_error, ConnectionRefusedError)
+
+    asyncio.run(scenario())
+
+
+def test_progressive_reconnection(monkeypatch: pytest.MonkeyPatch) -> None:
+    async def scenario() -> None:
+        _connection_factory(
+            monkeypatch,
+            [ConnectionRefusedError("1"), ConnectionRefusedError("2"), ConnectionRefusedError("3")],
+        )
+        delays: list[float] = []
+
+        async def pause(delay: float) -> None:
+            delays.append(delay)
+            await asyncio.sleep(0)
+
+        client = ArkteosClient(
+            "proxy.example", reconnect_delays=(1, 2, 5), sleep=pause
+        )
+        await client.start()
+        await _wait_for(lambda: len(delays) >= 3)
+        await client.stop()
+        assert delays[:3] == [1, 2, 5]
+        assert client.reconnect_count >= 3
+
+    asyncio.run(scenario())
+
+
+def test_complete_frame_read(monkeypatch: pytest.MonkeyPatch) -> None:
+    async def scenario() -> None:
+        metadata = _load_frames()["metadata"]
+        reader = ControlledReader([metadata])
+        _connection_factory(monkeypatch, [(reader, FakeWriter())])
+        client = ArkteosClient("proxy.example")
+        await client.start()
+        await _wait_for(lambda: client.frames_received == 1)
+        assert client.last_frame_type == "metadata"
+        await client.stop()
+
+    asyncio.run(scenario())
+
+
+def test_fragmented_read(monkeypatch: pytest.MonkeyPatch) -> None:
+    async def scenario() -> None:
+        frigo = _load_frames()["frigo"]
+        reader = ControlledReader([frigo[:70], frigo[70:]])
+        _connection_factory(monkeypatch, [(reader, FakeWriter())])
+        client = ArkteosClient("proxy.example")
+        await client.start()
+        await _wait_for(lambda: client.frames_received == 1)
+        assert client.latest_frigo_data is not None
+        await client.stop()
+
+    asyncio.run(scenario())
+
+
+def test_multiple_frames_in_one_read(monkeypatch: pytest.MonkeyPatch) -> None:
+    async def scenario() -> None:
+        frames = _load_frames()
+        reader = ControlledReader([frames["metadata"] + frames["frigo"]])
+        _connection_factory(monkeypatch, [(reader, FakeWriter())])
+        client = ArkteosClient("proxy.example")
+        await client.start()
+        await _wait_for(lambda: client.frames_received == 2)
+        await client.stop()
+
+    asyncio.run(scenario())
+
+
+def test_metadata_data(monkeypatch: pytest.MonkeyPatch) -> None:
+    async def scenario() -> None:
+        reader = ControlledReader([_load_frames()["metadata"]])
+        _connection_factory(monkeypatch, [(reader, FakeWriter())])
+        client = ArkteosClient("proxy.example")
+        await client.start()
+        await _wait_for(lambda: client.latest_metadata is not None)
+        assert client.latest_metadata == {"frame_type": "metadata"}
+        await client.stop()
+
+    asyncio.run(scenario())
+
+
+def test_frigo_data(monkeypatch: pytest.MonkeyPatch) -> None:
+    async def scenario() -> None:
+        reader = ControlledReader([_load_frames()["frigo"]])
+        _connection_factory(monkeypatch, [(reader, FakeWriter())])
+        client = ArkteosClient("proxy.example")
+        await client.start()
+        await _wait_for(lambda: client.latest_frigo_data is not None)
+        assert client.latest_frigo_data["frame_type"] == "frigo"
+        await client.stop()
+
+    asyncio.run(scenario())
+
+
+def test_regulation_data(monkeypatch: pytest.MonkeyPatch) -> None:
+    async def scenario() -> None:
+        reader = ControlledReader([_load_frames()["regulation"]])
+        _connection_factory(monkeypatch, [(reader, FakeWriter())])
+        client = ArkteosClient("proxy.example")
+        await client.start()
+        await _wait_for(lambda: client.latest_regulation_data is not None)
+        assert client.latest_regulation_data["frame_type"] == "regulation"
+        await client.stop()
+
+    asyncio.run(scenario())
+
+
+def test_invalid_frame_is_rejected_without_stopping(monkeypatch: pytest.MonkeyPatch) -> None:
+    class InvalidExtractor:
+        def feed(self, _data: bytes) -> list[bytes]:
+            return [b"invalid"]
+
+    async def scenario() -> None:
+        reader = ControlledReader([b"source"])
+        _connection_factory(monkeypatch, [(reader, FakeWriter())])
+        client = ArkteosClient("proxy.example")
+        client._extractor = InvalidExtractor()  # type: ignore[assignment]
+        await client.start()
+        await _wait_for(lambda: client.frames_rejected == 1)
+        assert client.running
+        await client.stop()
+
+    asyncio.run(scenario())
+
+
+def test_bytes_received_counter(monkeypatch: pytest.MonkeyPatch) -> None:
+    async def scenario() -> None:
+        metadata = _load_frames()["metadata"]
+        reader = ControlledReader([metadata])
+        _connection_factory(monkeypatch, [(reader, FakeWriter())])
+        client = ArkteosClient("proxy.example")
+        await client.start()
+        await _wait_for(lambda: client.bytes_received == len(metadata))
+        await client.stop()
+
+    asyncio.run(scenario())
+
+
+def test_frames_received_counter(monkeypatch: pytest.MonkeyPatch) -> None:
+    async def scenario() -> None:
+        frames = _load_frames()
+        reader = ControlledReader([frames["frigo"] + frames["regulation"]])
+        _connection_factory(monkeypatch, [(reader, FakeWriter())])
+        client = ArkteosClient("proxy.example")
+        await client.start()
+        await _wait_for(lambda: client.frames_received == 2)
+        await client.stop()
+
+    asyncio.run(scenario())
+
+
+def test_frames_rejected_counter(monkeypatch: pytest.MonkeyPatch) -> None:
+    class RaisingExtractor:
+        def feed(self, _data: bytes) -> list[bytes]:
+            raise ValueError("rejet")
+
+    async def scenario() -> None:
+        reader = ControlledReader([b"source"])
+        _connection_factory(monkeypatch, [(reader, FakeWriter())])
+        client = ArkteosClient("proxy.example")
+        client._extractor = RaisingExtractor()  # type: ignore[assignment]
+        await client.start()
+        await _wait_for(lambda: client.frames_rejected == 1)
+        await client.stop()
+
+    asyncio.run(scenario())
+
+
+def test_latest_data_updates(monkeypatch: pytest.MonkeyPatch) -> None:
+    async def scenario() -> None:
+        frames = _load_frames()
+        reader = ControlledReader(
+            [frames["metadata"] + frames["frigo"] + frames["regulation"]]
+        )
+        _connection_factory(monkeypatch, [(reader, FakeWriter())])
+        client = ArkteosClient("proxy.example")
+        await client.start()
+        await _wait_for(lambda: client.frames_received == 3)
+        assert client.last_frame_type == "regulation"
+        assert client.latest_metadata is not None
+        assert client.latest_frigo_data is not None
+        assert client.latest_regulation_data is not None
+        await client.stop()
+
+    asyncio.run(scenario())
+
+
+def test_synchronous_data_callback(monkeypatch: pytest.MonkeyPatch) -> None:
+    async def scenario() -> None:
+        received: list[str] = []
+        reader = ControlledReader([_load_frames()["frigo"]])
+        _connection_factory(monkeypatch, [(reader, FakeWriter())])
+        client = ArkteosClient("proxy.example")
+        client.add_data_callback(lambda frame_type, _accepted, _rejected: received.append(frame_type))
+        await client.start()
+        await _wait_for(lambda: received == ["frigo"])
+        await client.stop()
+
+    asyncio.run(scenario())
+
+
+def test_asynchronous_data_callback(monkeypatch: pytest.MonkeyPatch) -> None:
+    async def scenario() -> None:
+        received: list[str] = []
+
+        async def callback(frame_type: str, _accepted: dict[str, object], _rejected: dict[str, object]) -> None:
+            received.append(frame_type)
+
+        reader = ControlledReader([_load_frames()["regulation"]])
+        _connection_factory(monkeypatch, [(reader, FakeWriter())])
+        client = ArkteosClient("proxy.example")
+        client.add_data_callback(callback)
+        await client.start()
+        await _wait_for(lambda: received == ["regulation"])
+        await client.stop()
+
+    asyncio.run(scenario())
+
+
+def test_callback_exception_does_not_stop_client(monkeypatch: pytest.MonkeyPatch) -> None:
+    async def scenario() -> None:
+        def callback(_frame_type: str, _accepted: object, _rejected: object) -> None:
+            raise RuntimeError("callback")
+
+        reader = ControlledReader([_load_frames()["metadata"]])
+        _connection_factory(monkeypatch, [(reader, FakeWriter())])
+        client = ArkteosClient("proxy.example")
+        client.add_data_callback(callback)
+        await client.start()
+        await _wait_for(lambda: client.frames_received == 1)
+        assert client.running
+        assert isinstance(client.last_error, RuntimeError)
+        await client.stop()
+
+    asyncio.run(scenario())
+
+
+def test_availability_true_after_valid_frame(monkeypatch: pytest.MonkeyPatch) -> None:
+    async def scenario() -> None:
+        states: list[bool] = []
+        reader = ControlledReader([_load_frames()["metadata"]])
+        _connection_factory(monkeypatch, [(reader, FakeWriter())])
+        client = ArkteosClient("proxy.example")
+        client.add_availability_callback(states.append)
+        await client.start()
+        await _wait_for(lambda: client.available)
+        assert states == [True]
+        await client.stop()
+
+    asyncio.run(scenario())
+
+
+def test_availability_false_after_timeout(monkeypatch: pytest.MonkeyPatch) -> None:
+    async def scenario() -> None:
+        states: list[bool] = []
+        reader = ControlledReader([_load_frames()["metadata"]])
+        _connection_factory(monkeypatch, [(reader, FakeWriter())])
+        client = ArkteosClient("proxy.example", frame_timeout=0.01)
+        client.add_availability_callback(states.append)
+        await client.start()
+        await _wait_for(lambda: states == [True, False])
+        assert not client.available
+        await client.stop()
+
+    asyncio.run(scenario())
+
+
+def test_availability_restored_after_new_frame(monkeypatch: pytest.MonkeyPatch) -> None:
+    async def scenario() -> None:
+        states: list[bool] = []
+        frames = _load_frames()
+        reader = ControlledReader([frames["metadata"]])
+        _connection_factory(monkeypatch, [(reader, FakeWriter())])
+        client = ArkteosClient("proxy.example", frame_timeout=0.01)
+        client.add_availability_callback(states.append)
+        await client.start()
+        await _wait_for(lambda: states == [True, False])
+        reader.push(frames["frigo"])
+        await _wait_for(lambda: states == [True, False, True])
+        await client.stop()
+
+    asyncio.run(scenario())
+
+
+def test_remote_disconnection(monkeypatch: pytest.MonkeyPatch) -> None:
+    async def scenario() -> None:
+        writer = FakeWriter()
+        reader = ControlledReader([b""])
+        _connection_factory(monkeypatch, [(reader, writer), ConnectionRefusedError("fin")])
+
+        async def pause(_delay: float) -> None:
+            await asyncio.sleep(0)
+
+        client = ArkteosClient("proxy.example", reconnect_delays=(1,), sleep=pause)
+        await client.start()
+        await _wait_for(lambda: writer.closed)
+        assert not client.connected
+        await client.stop()
+
+    asyncio.run(scenario())
+
+
+def test_writer_is_closed(monkeypatch: pytest.MonkeyPatch) -> None:
+    async def scenario() -> None:
+        writer = FakeWriter()
+        _connection_factory(monkeypatch, [(ControlledReader(), writer)])
+        client = ArkteosClient("proxy.example")
+        await client.start()
+        await client.wait_until_connected(0.1)
+        await client.stop()
+        assert writer.closed
+        assert writer.wait_closed_calls == 1
+
+    asyncio.run(scenario())
+
+
+def test_stop_during_reconnect_wait(monkeypatch: pytest.MonkeyPatch) -> None:
+    async def scenario() -> None:
+        _connection_factory(monkeypatch, [ConnectionRefusedError("refus")])
+        sleeping = asyncio.Event()
+
+        async def pause(_delay: float) -> None:
+            sleeping.set()
+            await asyncio.Event().wait()
+
+        client = ArkteosClient("proxy.example", sleep=pause)
+        await client.start()
+        await _wait_for(sleeping.is_set)
+        await client.stop()
+        assert not client.running
+
+    asyncio.run(scenario())
+
+
+def test_stop_during_read(monkeypatch: pytest.MonkeyPatch) -> None:
+    async def scenario() -> None:
+        writer = FakeWriter()
+        _connection_factory(monkeypatch, [(ControlledReader(), writer)])
+        client = ArkteosClient("proxy.example")
+        await client.start()
+        await client.wait_until_connected(0.1)
+        await client.stop()
+        assert writer.closed
+        assert not client.connected
+
+    asyncio.run(scenario())
+
+
+def test_double_start_does_not_create_two_tasks(monkeypatch: pytest.MonkeyPatch) -> None:
+    async def scenario() -> None:
+        _connection_factory(monkeypatch, [(ControlledReader(), FakeWriter())])
+        client = ArkteosClient("proxy.example")
+        await client.start()
+        task = client._task
+        await client.start()
+        assert client._task is task
+        await client.stop()
+
+    asyncio.run(scenario())
+
+
+def test_double_stop_is_safe(monkeypatch: pytest.MonkeyPatch) -> None:
+    async def scenario() -> None:
+        _connection_factory(monkeypatch, [(ControlledReader(), FakeWriter())])
+        client = ArkteosClient("proxy.example")
+        await client.start()
+        await client.stop()
+        await client.stop()
+        assert not client.running
+
+    asyncio.run(scenario())
+
+
+def test_client_has_no_socket_write_operations(monkeypatch: pytest.MonkeyPatch) -> None:
+    async def scenario() -> None:
+        writer = FakeWriter()
+        _connection_factory(monkeypatch, [(ControlledReader([_load_frames()["metadata"]]), writer)])
+        client = ArkteosClient("proxy.example")
+        await client.start()
+        await _wait_for(lambda: client.frames_received == 1)
+        await client.stop()
+        assert not hasattr(writer, "write")
+
+    asyncio.run(scenario())
+
+
+def test_no_client_task_remains_after_stop(monkeypatch: pytest.MonkeyPatch) -> None:
+    async def scenario() -> None:
+        _connection_factory(monkeypatch, [(ControlledReader(), FakeWriter())])
+        client = ArkteosClient("proxy.example")
+        await client.start()
+        task = client._task
+        await client.stop()
+        assert client._task is None
+        assert task is not None and task.done()
+
+    asyncio.run(scenario())
+
+
+def test_all_real_fixtures(monkeypatch: pytest.MonkeyPatch) -> None:
+    async def scenario() -> None:
+        frames = _load_frames()
+        reader = ControlledReader(
+            [frames["metadata"] + frames["frigo"] + frames["regulation"]]
+        )
+        _connection_factory(monkeypatch, [(reader, FakeWriter())])
+        client = ArkteosClient("proxy.example")
+        await client.start()
+        await _wait_for(lambda: client.frames_received == 3)
+        assert client.last_frame_type == "regulation"
+        await client.stop()
+
+    asyncio.run(scenario())