Jean-Pierre MartineauandClaude Opus 5.5
11b5274b31
Each test docstring states the behaviour that the test proves, in the simple present. No assertion message, test id, or other string literal changed. Four statements now agree with the code: the Trace highlight stays on the last row, a menu count, and two others. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
183 lines
6.8 KiB
Python
183 lines
6.8 KiB
Python
# SPDX-License-Identifier: Apache-2.0
|
|
"""Regression tests for channel reads that are serialized and checked against the index.
|
|
|
|
The ``get_channel`` of the meshcore library waits for "the next CHANNEL_INFO event". It does
|
|
not match this event to the slot that it asked for, and the dispatcher sends the event to
|
|
each waiter that is in flight. Thus two concurrent reads both resolve on the first response.
|
|
One caller silently gets the channel of the other caller, and this files the messages of that
|
|
channel in the wrong place. :class:`MeshCoreDevice` prevents this. It serializes the reads and
|
|
checks that the response is for the slot that it requested.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
|
|
import pytest
|
|
|
|
from meshterm.core.connection import DeviceCommandError, MeshCoreDevice
|
|
|
|
|
|
class _Event:
|
|
"""A minimal stand-in for a meshcore CHANNEL_INFO event."""
|
|
|
|
def __init__(self, payload: dict) -> None:
|
|
self.payload = payload
|
|
|
|
def is_error(self) -> bool:
|
|
return False
|
|
|
|
|
|
class _CrosstalkCommands:
|
|
"""A fake ``commands`` that causes the race of unmatched responses if reads overlap.
|
|
|
|
Each call records itself as in flight and gives control to other tasks. If a second read
|
|
starts while the first read is not complete, both reads return the payload of the read
|
|
that finishes first. This is the behaviour of the library that sends messages to the wrong
|
|
place. Serialized callers never overlap, so each caller gets its own channel.
|
|
"""
|
|
|
|
def __init__(self) -> None:
|
|
self.in_flight: list[int] = []
|
|
self.peak_concurrency = 0
|
|
|
|
async def get_channel(self, index: int) -> _Event:
|
|
self.in_flight.append(index)
|
|
self.peak_concurrency = max(self.peak_concurrency, len(self.in_flight))
|
|
try:
|
|
await asyncio.sleep(0) # a real await point, where an interleave can occur
|
|
# Model "the first response resolves everyone". The crossed answer is the
|
|
# oldest request in flight, which is not always this request.
|
|
served = self.in_flight[0]
|
|
return _Event(
|
|
{
|
|
"channel_idx": served,
|
|
"channel_name": f"chan{served}",
|
|
"channel_secret": bytes([served]) * 16,
|
|
}
|
|
)
|
|
finally:
|
|
self.in_flight.remove(index)
|
|
|
|
|
|
class _MismatchCommands:
|
|
"""A fake ``commands`` that always answers with the wrong slot index."""
|
|
|
|
async def get_channel(self, index: int) -> _Event:
|
|
wrong = index + 1
|
|
return _Event(
|
|
{
|
|
"channel_idx": wrong,
|
|
"channel_name": f"chan{wrong}",
|
|
"channel_secret": bytes([wrong]) * 16,
|
|
}
|
|
)
|
|
|
|
|
|
def _device_with(commands: object) -> MeshCoreDevice:
|
|
device = MeshCoreDevice("COM-TEST")
|
|
device._mc = type("_MC", (), {"commands": commands})()
|
|
return device
|
|
|
|
|
|
async def test_get_channel_serializes_concurrent_reads() -> None:
|
|
"""Concurrent reads never overlap, so each read returns its own slot (no cross-talk)."""
|
|
commands = _CrosstalkCommands()
|
|
device = _device_with(commands)
|
|
|
|
results = await asyncio.gather(*(device.get_channel(i) for i in range(8)))
|
|
|
|
assert commands.peak_concurrency == 1
|
|
for i, payload in enumerate(results):
|
|
assert payload is not None
|
|
assert payload["channel_idx"] == i
|
|
assert payload["channel_name"] == f"chan{i}"
|
|
|
|
|
|
async def test_get_channel_rejects_response_for_a_different_slot() -> None:
|
|
"""The device refuses a response with an index that does not match the request.
|
|
|
|
It never returns the response.
|
|
"""
|
|
device = _device_with(_MismatchCommands())
|
|
|
|
with pytest.raises(DeviceCommandError, match="returned slot"):
|
|
await device.get_channel(0)
|
|
|
|
|
|
class _Refusal:
|
|
"""A meshcore ERROR event: a refusal with an error code of the companion, or a timeout."""
|
|
|
|
def __init__(self, payload: dict) -> None:
|
|
self.payload = payload
|
|
|
|
def is_error(self) -> bool:
|
|
return True
|
|
|
|
|
|
class _ScriptedCommands:
|
|
"""A fake ``commands`` that answers each slot from a queue of scripted replies.
|
|
|
|
``replies[i]`` is a list of what the next reads of slot ``i`` get: a channel name, or an
|
|
error payload. A slot with no script is past the end. The firmware answers NOT_FOUND for it.
|
|
"""
|
|
|
|
def __init__(self, replies: dict[int, list]) -> None:
|
|
self.replies = {i: list(r) for i, r in replies.items()}
|
|
self.reads: list[int] = []
|
|
|
|
async def get_channel(self, index: int): # noqa: ANN201
|
|
self.reads.append(index)
|
|
queue = self.replies.get(index)
|
|
reply = queue.pop(0) if queue else {"error_code": 2}
|
|
if isinstance(reply, dict):
|
|
return _Refusal(reply)
|
|
return _Event({"channel_idx": index, "channel_name": reply, "channel_secret": bytes(16)})
|
|
|
|
|
|
#: The refusal of the clock sync, as a concurrent channel read receives it.
|
|
_MALFORMED = {"error_code": 6}
|
|
|
|
|
|
async def test_a_borrowed_refusal_does_not_end_the_slot_probe() -> None:
|
|
"""The error of another command, delivered to a slot read, does not end the list of slots.
|
|
|
|
The library gives the first ERROR that arrives to each waiter. Thus, at connect, the
|
|
channel prewarm can take the "malformed" error of the clock sync as the end of the list
|
|
at slot 2. Then it caches a list that does not have the channels after slot 2.
|
|
"""
|
|
from meshterm.core.channel_probe import probe_channel_slots
|
|
|
|
commands = _ScriptedCommands({0: ["Public"], 1: ["ops"], 2: [_MALFORMED, "#montreal"]})
|
|
slots, complete = await probe_channel_slots(_device_with(commands))
|
|
assert [s.name for s in slots] == ["Public", "ops", "#montreal"]
|
|
assert complete
|
|
assert commands.reads.count(2) == 2
|
|
|
|
|
|
async def test_a_refusal_that_repeats_is_a_failed_read_not_a_layout() -> None:
|
|
"""Two refusals in a row that are not NOT_FOUND leave the probe unfinished.
|
|
|
|
MeshTerm does not cache the result.
|
|
"""
|
|
from meshterm.core.channel_probe import probe_channel_slots
|
|
|
|
commands = _ScriptedCommands({0: ["Public"], 1: [_MALFORMED, _MALFORMED], 2: ["ops"]})
|
|
slots, complete = await probe_channel_slots(_device_with(commands))
|
|
assert [s.name for s in slots] == ["Public"]
|
|
assert not complete
|
|
|
|
|
|
async def test_no_answer_is_a_failed_read_and_not_found_is_the_end() -> None:
|
|
"""A timeout is never the end of the slots. The NOT_FOUND of the firmware is the end."""
|
|
from meshterm.core.channel_probe import probe_channel_slots
|
|
|
|
timeout = {"reason": "no_event_received"}
|
|
slots, complete = await probe_channel_slots(
|
|
_device_with(_ScriptedCommands({0: ["Public"], 1: [timeout]}))
|
|
)
|
|
assert [s.name for s in slots] == ["Public"] and not complete
|
|
|
|
slots, complete = await probe_channel_slots(_device_with(_ScriptedCommands({0: ["Public"]})))
|
|
assert [s.name for s in slots] == ["Public"] and complete
|