524 lines
18 KiB
Python
524 lines
18 KiB
Python
"""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())
|