Mirrors_Framework/esphome
1
0
Fork 0
mirror of https://github.com/esphome/esphome.git synced 2026-08-18 03:59:08 +08:00
Code Issues Actions 9 Packages Projects Releases Wiki Activity
Files
c7d6b4aaa4643d97ea21e097872d2f03178adc3f
esphome/tests/integration/test_uart_mock_modbus.py
T
Bonne EgglestonJ. Nick KostonClaude
 f5ee72753d [modbus_client] Add typed read/write actions (#18078) 
Co-authored-by: J. Nick Koston <nick@koston.org>
Co-authored-by: Claude <noreply@anthropic.com>
2026-08-07 14:18:20 -05:00

759 lines
30 KiB
Python
Raw Blame History

"""Integration test for modbus component with virtual UART.
Tests:
test_uart_mock_modbus :
1. Read a single register and parse successfully (basic_register)
2. Read multiple registers from SDM meter and parse successfully (sdm_voltage), with some intermediate delay to simulate UART buffer time.
test_uart_mock_modbus_no_threshold :
Test modbus with no rx_full_threshold set (simulating USB UART / non-hardware UART).
Verifies the 50ms fallback timeout handles chunked data with USB packet gaps.
test_uart_mock_modbus_fairness :
Two controllers sharing one client bus, both polling far faster than the bus
can service. Verifies the hub schedules them fairly (request counts within 1).
"""
from __future__ import annotations
import asyncio
from collections.abc import Callable
from dataclasses import dataclass
from aioesphomeapi import ButtonInfo, NumberInfo
import pytest
from .state_utils import SensorTracker, find_entity
from .types import APIClientConnectedFactory, RunCompiledFunction
@dataclass
class RegisterTestCase:
"""Test parameters for a single modbus register write/read round-trip."""
initial_value: object
write_number_name: str
write_value: float
post_write_value: object
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _make_modbus_line_callback() -> tuple[Callable[[str], None], list[str], list[str]]:
"""Return a (callback, error_lines, warning_lines) tuple for tracking modbus log output.
Only captures bus-level modbus messages ([modbus:]), not modbus_controller
scheduling noise (e.g. "Duplicate modbus command found").
"""
error_log_lines: list[str] = []
warning_log_lines: list[str] = []
def line_callback(line: str) -> None:
if "[E][modbus:" in line:
error_log_lines.append(line)
if "[W][modbus:" in line:
warning_log_lines.append(line)
return line_callback, error_log_lines, warning_log_lines
def _assert_no_modbus_errors(
error_log_lines: list[str], warning_log_lines: list[str]
) -> None:
assert len(error_log_lines) == 0, (
"Expect no errors logged by the modbus mock, but got:\n"
+ "\n".join(error_log_lines)
)
assert len(warning_log_lines) == 0, (
"Expect no warnings logged by the modbus mock, but got:\n"
+ "\n".join(warning_log_lines)
)
# ---------------------------------------------------------------------------
# Tests
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_uart_mock_modbus(
yaml_config: str,
run_compiled: RunCompiledFunction,
api_client_connected: APIClientConnectedFactory,
) -> None:
"""Test basic modbus data parsing."""
tracker = SensorTracker(
[
"basic_register",
"delayed_response",
"late_response",
"no_response",
"exception_response",
]
)
basic_register_changed = tracker.expect("basic_register", 259.0)
delayed_response_changed = tracker.expect("delayed_response", 255.0)
# late_response / no_response / exception_response: expect *any* value
# (these should never fire, so we use a permissive match via expect_any)
late_response_changed = tracker.expect_any("late_response")
no_response_changed = tracker.expect_any("no_response")
exception_response_changed = tracker.expect_any("exception_response")
async with (
run_compiled(yaml_config),
api_client_connected() as client,
):
await tracker.setup_and_start_scenario(client)
await tracker.await_change(delayed_response_changed, "delayed_response")
await tracker.await_change(basic_register_changed, "basic_register")
# Run all "must not change" checks concurrently — each waits the full
# timeout, so sequential execution would multiply the wall time.
await asyncio.gather(
tracker.await_must_not_change(late_response_changed, "late_response"),
tracker.await_must_not_change(no_response_changed, "no_response"),
tracker.await_must_not_change(
exception_response_changed, "exception_response"
),
)
@pytest.mark.asyncio
async def test_uart_mock_modbus_timing(
yaml_config: str,
run_compiled: RunCompiledFunction,
api_client_connected: APIClientConnectedFactory,
) -> None:
"""Test modbus timing with multi-register SDM meter response."""
line_callback, error_log_lines, warning_log_lines = _make_modbus_line_callback()
tracker = SensorTracker(["sdm_voltage"])
voltage_changed = tracker.expect_any("sdm_voltage")
async with (
run_compiled(yaml_config, line_callback=line_callback),
api_client_connected() as client,
):
await tracker.setup_and_start_scenario(client)
await tracker.await_change(voltage_changed, "sdm_voltage")
_assert_no_modbus_errors(error_log_lines, warning_log_lines)
@pytest.mark.asyncio
async def test_uart_mock_modbus_no_threshold(
yaml_config: str,
run_compiled: RunCompiledFunction,
api_client_connected: APIClientConnectedFactory,
) -> None:
"""Test modbus with no rx_full_threshold (simulating USB UART).
Without the 50ms fallback timeout, the chunked response with a 40ms gap
between USB packets would cause a false timeout and CRC failure cascade.
Bus-level warnings (CRC/parse failures, buffer clears) are NOT expected during
chunked reassembly, if timeouts are set properly — these warnings indicate undersized timeouts.
"""
line_callback, error_log_lines, warning_log_lines = _make_modbus_line_callback()
tracker = SensorTracker(["sdm_voltage"])
voltage_changed = tracker.expect_any("sdm_voltage")
async with (
run_compiled(yaml_config, line_callback=line_callback),
api_client_connected() as client,
):
await tracker.setup_and_start_scenario(client)
await tracker.await_change(voltage_changed, "sdm_voltage")
_assert_no_modbus_errors(error_log_lines, warning_log_lines)
@pytest.mark.asyncio
async def test_uart_mock_modbus_server(
yaml_config: str,
run_compiled: RunCompiledFunction,
api_client_connected: APIClientConnectedFactory,
) -> None:
"""Test modbus server parsing with peer traffic on a shared bus."""
line_callback, error_log_lines, warning_log_lines = _make_modbus_line_callback()
tracker = SensorTracker(
["basic_read", "read_after_peer_response", "read_after_peer_timeout"]
)
futures = tracker.expect_all(
{
"basic_read": 1,
"read_after_peer_response": 1,
"read_after_peer_timeout": 1,
}
)
async with (
run_compiled(yaml_config, line_callback=line_callback),
api_client_connected() as client,
):
await tracker.setup_and_start_scenario(client)
await tracker.await_all(futures)
_assert_no_modbus_errors(error_log_lines, warning_log_lines)
@pytest.mark.asyncio
async def test_uart_mock_modbus_server_read_write(
yaml_config: str,
run_compiled: RunCompiledFunction,
api_client_connected: APIClientConnectedFactory,
) -> None:
"""Test modbus server FC 0x17 (read/write multiple registers).
Injects raw 0x17 request frames and checks the round-trip through the
server's read_lambda/write_lambda, independent of how the hub dispatches
0x17 internally:
* one request writes reg 0x01 then reads regs 0x01+0x02 -- reg 0x01 reads
back the just-written value (the write happens before the read per
Modbus 6.17), and the second register is returned by the same
multi-register read;
* a second request writes and reads a different register block.
"""
line_callback, error_log_lines, warning_log_lines = _make_modbus_line_callback()
tracker = SensorTracker(
["rw_write_1", "rw_read_1", "rw_read_2", "rw_write_3", "rw_read_3"]
)
futures = tracker.expect_all(
{
"rw_write_1": 4660, # 0x1234 written to reg 0x0001
"rw_read_1": 4660, # reg 0x0001 reads back the just-written value
"rw_read_2": 170, # 0x00AA read from reg 0x0002 in the same request
"rw_write_3": 22136, # 0x5678 written to reg 0x0003
"rw_read_3": 22136, # reg 0x0003 reads back the just-written value
}
)
async with (
run_compiled(yaml_config, line_callback=line_callback),
api_client_connected() as client,
):
await tracker.setup_and_start_scenario(client)
await tracker.await_all(futures)
_assert_no_modbus_errors(error_log_lines, warning_log_lines)
@pytest.mark.asyncio
async def test_uart_mock_modbus_server_read_write_invalid(
yaml_config: str,
run_compiled: RunCompiledFunction,
api_client_connected: APIClientConnectedFactory,
) -> None:
"""Test modbus server FC 0x17 invalid-frame handling.
Injects a well-formed (valid CRC) 0x17 request whose write byte count (2)
does not match 2x the write quantity (2 registers need 4 bytes), so the hub
must reject it with ILLEGAL_DATA_VALUE before touching any register. A valid
read is injected right after as a processing marker.
The invalid frame is verified via bus-level signals rather than the reply
frame on the wire: the mock UART cannot observe the server's TX reliably on
the host platform (the server's transmission is gated by a millis()-based tx
delay), so instead we assert the request is rejected exactly once and never
applied to a register.
"""
line_callback, error_log_lines, warning_log_lines = _make_modbus_line_callback()
tracker = SensorTracker(["write_seen", "probe"])
probe_seen = tracker.expect("probe", 1)
async with (
run_compiled(yaml_config, line_callback=line_callback),
api_client_connected() as client,
):
await tracker.setup_and_start_scenario(client)
# The probe read is injected after the malformed frame, so once it fires
# the malformed frame has already been processed.
await tracker.await_change(probe_seen, "probe")
# Exactly one bus-level rejection for the malformed frame (no cascade)...
invalid_warnings = [
line for line in warning_log_lines if "Invalid number of registers" in line
]
assert len(invalid_warnings) == 1, (
"Expected exactly one invalid-frame rejection, got warnings:\n"
+ "\n".join(warning_log_lines)
)
assert len(error_log_lines) == 0, (
"Expected no modbus errors, but got:\n" + "\n".join(error_log_lines)
)
# ...and the rejected write is never applied to the target register.
assert not tracker.sensor_states["write_seen"], (
f"malformed 0x17 must not write, but write_seen fired: {tracker.sensor_states['write_seen']}"
)
@pytest.mark.asyncio
async def test_uart_mock_modbus_server_controller(
yaml_config: str,
run_compiled: RunCompiledFunction,
api_client_connected: APIClientConnectedFactory,
) -> None:
"""Test server/controller functionality for all read register types."""
line_callback, error_log_lines, warning_log_lines = _make_modbus_line_callback()
expected_values = {
"reg_u_word": 99,
"reg_u_word_s": 4660,
"reg_u_word_s_raw": 13330,
"reg_s_word": -99,
"reg_s_word_s": -2,
"reg_u_dword": 16909060,
"reg_s_dword": -16909060,
"reg_u_dword_r": pytest.approx(67305985),
"reg_s_dword_r": pytest.approx(-67305985),
"reg_u_qword": pytest.approx(72623859790382856),
"reg_s_qword": pytest.approx(-72623859790382856),
"reg_u_qword_r": pytest.approx(578437695752307201),
"reg_s_qword_r": pytest.approx(-578437695752307201),
"reg_fp32": pytest.approx(3.14),
"reg_fp32_r": pytest.approx(3.14),
}
tracker = SensorTracker(list(expected_values.keys()))
futures = tracker.expect_all(expected_values)
async with (
run_compiled(yaml_config, line_callback=line_callback),
api_client_connected() as client,
):
await tracker.setup_and_start_scenario(client)
await tracker.await_all(futures)
_assert_no_modbus_errors(error_log_lines, warning_log_lines)
@pytest.mark.asyncio
async def test_uart_mock_modbus_server_controller_write(
yaml_config: str,
run_compiled: RunCompiledFunction,
api_client_connected: APIClientConnectedFactory,
) -> None:
"""Test server/controller write functionality for all register value types.
Verifies that writing to modbus server registers via the controller updates
the server's stored values, which are then read back correctly on the next poll.
All 14 value types are tested: U/S_WORD, U/S_WORD_S, U/S_DWORD(_R), U/S_QWORD(_R), FP32(_R).
"""
line_callback, error_log_lines, warning_log_lines = _make_modbus_line_callback()
register_test_cases: dict[str, RegisterTestCase] = {
"reg_u_word": RegisterTestCase(11, "write_u_word", 42, 42),
"reg_u_word_s": RegisterTestCase(4660, "write_u_word_s", 17185, 17185),
"reg_s_word": RegisterTestCase(-11, "write_s_word", -42, -42),
"reg_s_word_s": RegisterTestCase(-2, "write_s_word_s", -257, -257),
"reg_u_dword": RegisterTestCase(1001, "write_u_dword", 2002, 2002),
"reg_s_dword": RegisterTestCase(-1001, "write_s_dword", -2002, -2002),
"reg_u_dword_r": RegisterTestCase(3003, "write_u_dword_r", 4004, 4004),
"reg_s_dword_r": RegisterTestCase(-3003, "write_s_dword_r", -4004, -4004),
"reg_u_qword": RegisterTestCase(5005, "write_u_qword", 6006, 6006),
"reg_s_qword": RegisterTestCase(-5005, "write_s_qword", -6006, -6006),
"reg_u_qword_r": RegisterTestCase(7007, "write_u_qword_r", 8008, 8008),
"reg_s_qword_r": RegisterTestCase(-7007, "write_s_qword_r", -8008, -8008),
"reg_fp32": RegisterTestCase(
pytest.approx(1.5, abs=0.01),
"write_fp32",
3.14,
pytest.approx(3.14, abs=0.01),
),
"reg_fp32_r": RegisterTestCase(
pytest.approx(2.5, abs=0.01),
"write_fp32_r",
6.28,
pytest.approx(6.28, abs=0.01),
),
}
tracker = SensorTracker(list(register_test_cases.keys()))
# Phase 1: expect initial baseline values
initial_futures = tracker.expect_all(
{name: case.initial_value for name, case in register_test_cases.items()}
)
# Phase 2: expect post-write values (registered now so on_state can match them)
written_futures = tracker.expect_all(
{name: case.post_write_value for name, case in register_test_cases.items()}
)
async with (
run_compiled(yaml_config, line_callback=line_callback),
api_client_connected() as client,
):
entities = await tracker.setup_and_start_scenario(client)
# Wait for initial baseline values to confirm the controller <-> server
# connection is working before issuing writes
await tracker.await_all(initial_futures, timeout=4.0)
# Issue write commands for all register types
for case in register_test_cases.values():
entity = find_entity(entities, case.write_number_name, NumberInfo)
assert entity is not None, (
f"{case.write_number_name} number entity not found"
)
client.number_command(entity.key, case.write_value)
# Wait for sensors to reflect the written values (round-trip write+read)
await tracker.await_all(written_futures, timeout=4.0)
_assert_no_modbus_errors(error_log_lines, warning_log_lines)
@pytest.mark.asyncio
async def test_uart_mock_modbus_server_controller_multiple(
yaml_config: str,
run_compiled: RunCompiledFunction,
api_client_connected: APIClientConnectedFactory,
) -> None:
"""Test server/controller functionality with multiple servers."""
line_callback, error_log_lines, warning_log_lines = _make_modbus_line_callback()
expected_values = {"reg_u_word": 919, "reg_u_word_2": 929}
tracker = SensorTracker(list(expected_values.keys()))
futures = tracker.expect_all(expected_values)
async with (
run_compiled(yaml_config, line_callback=line_callback),
api_client_connected() as client,
):
await tracker.setup_and_start_scenario(client)
await tracker.await_all(futures)
_assert_no_modbus_errors(error_log_lines, warning_log_lines)
@pytest.mark.asyncio
async def test_uart_mock_modbus_client_typed(
yaml_config: str,
run_compiled: RunCompiledFunction,
api_client_connected: APIClientConnectedFactory,
) -> None:
"""Test the typed modbus_client actions end to end (each action its own hub device).
Start Scenario fires three typed actions: write_single_register puts 777 in server register 0x10 (the
ack fires on_response -> ack_flag); read_holding_registers reads it back,
with the reply decoded by the shared device dispatch into host-order words (values[0] -> typed_value);
a read of unserved register 0x99 resolves via on_error with the device's exception code
(ILLEGAL_DATA_ADDRESS = 2 -> error_code); a coil read of the register-only server resolves via
on_error with ILLEGAL_FUNCTION (= 1 -> coil_error_code), proving the bit-read request and typed error
delivery. A multi-register write (fc 0x10) lands on registers 0x11/0x12 with the read-back of 0x12
chained inside its ack handler (-> multi_value = 222); a multi-coil write draws ILLEGAL_FUNCTION from
the register-only server (-> multi_coil_error = 1). A read whose count lambda returns 0 at runtime
builds an empty (rejected) PDU, is refused at the hub door, and resolves via on_not_sent
(-> not_sent_flag).
"""
tracker = SensorTracker(
[
"typed_value",
"ack_flag",
"error_code",
"coil_error_code",
"multi_value",
"multi_coil_error",
"not_sent_flag",
]
)
futures = tracker.expect_all(
{
"typed_value": 777,
"ack_flag": 1,
"error_code": 2,
"coil_error_code": 1,
"multi_value": 222,
"multi_coil_error": 1,
"not_sent_flag": 1,
}
)
async with (
run_compiled(yaml_config),
api_client_connected() as client,
):
await tracker.setup_and_start_scenario(client)
await tracker.await_all(futures)
@pytest.mark.asyncio
async def test_uart_mock_modbus_client_inline(
yaml_config: str,
run_compiled: RunCompiledFunction,
api_client_connected: APIClientConnectedFactory,
) -> None:
"""Test modbus_client.send actions: each action is its own hub device.
Start Scenario fires: a read of served address 1 decoded in its inline on_response -> inline_value; a
read of address 2, which no server answers, resolving via on_no_response -> timeout_flag. A parallel
script fires the same write action twice while its first frame is pending; the hub drops the duplicate
write, and the second firing resolves via its own on_not_sent -> skipped_flag. This exercises
per-action reply routing, the no-reply path, and the one-outcome guarantee under the hub's write
dedup.
"""
tracker = SensorTracker(["inline_value", "timeout_flag", "skipped_flag"])
futures = tracker.expect_all(
{"inline_value": 1234, "timeout_flag": 1, "skipped_flag": 1}
)
async with (
run_compiled(yaml_config),
api_client_connected() as client,
):
await tracker.setup_and_start_scenario(client)
await tracker.await_all(futures, timeout=5.0)
@pytest.mark.asyncio
async def test_uart_mock_modbus_grouping(
yaml_config: str,
run_compiled: RunCompiledFunction,
api_client_connected: APIClientConnectedFactory,
) -> None:
"""Pins how sensors are grouped into polled ranges across the combinations that matter.
Each block in the fixture covers one relationship between neighbouring sensors - sharing a wide
register, contiguous, separated by a gap, differing polling rates, coils, and a pinned range - so
that the frames on the wire and the byte each sensor decodes from are locked down.
"""
line_callback, error_log_lines, warning_log_lines = _make_modbus_line_callback()
# Values are those the component produced before the range rework, captured from it directly.
expected_values = {
# one register returning 4 bytes, read as two halves
"reuse_lo": 273,
"reuse_hi": 546,
# contiguous registers, mixed widths
"ext_word": 4660,
"ext_next": 22136,
"ext_dword": pytest.approx(2596069120),
# a wide register pushes its neighbour past the bytes it actually returned
"wide_first": 2730,
"wide_next": 3003,
# a gap keeps them apart
"gap_low": 320,
"gap_high": 325,
# contiguous, second one polling more slowly
"rate_first": 336,
"rate_slow": 337,
# a wide value, one of its halves, and the register after it
"shared_dword": pytest.approx(2759468),
"shared_high": 6956,
"shared_after": 781,
# a wide register hidden behind a wider plain sibling, and the sensor after them
"masked_wide": 4369,
"masked_pair": pytest.approx(286335522),
"masked_after": 13107,
# pinned range, and the contiguous sensor after it
"forced_first": 352,
"forced_next": 353,
}
tracker = SensorTracker(list(expected_values.keys()))
futures = tracker.expect_all(expected_values)
async with (
run_compiled(yaml_config, line_callback=line_callback),
api_client_connected() as client,
):
await tracker.setup_and_start_scenario(client)
await tracker.await_all(futures)
# Every frame sent must match one the mock answers, so an unexpected read (a range that split,
# merged or changed length) shows up here as an unanswered request. This is what pins the coil
# grouping too, since binary sensors carry no numeric state to compare.
_assert_no_modbus_errors(error_log_lines, warning_log_lines)
@pytest.mark.asyncio
async def test_uart_mock_modbus_shared_address(
yaml_config: str,
run_compiled: RunCompiledFunction,
api_client_connected: APIClientConnectedFactory,
) -> None:
"""Sensors sharing and overlapping one register range must all decode from a single read.
A U_WORD and a U_DWORD share start address 0x9001 (non-mergeable, so the range widens to 2
registers) and a third U_WORD at 0x9002 falls inside the widened range. A regression guard for the
range-grouping rewrite: without the same-address fallback the shared sensors land in duplicate
ranges and one never publishes; without the in-range join the 0x9002 sensor splits into a second
overlapping frame that the mock (which expects exactly one read) never answers.
A force_new_range sensor at 0x30 plus a plain sensor at 0x10 pin the covered branch's lower-bound
check: the forced sensor sorts first, and without the bound the lower-address sensor is absorbed
into the forced range with a wrapped byte offset and never polls its own register.
A U_QWORD at 0x100 with plain sensors at 0x101 and 0x103 pins that non-merging sensors inside a
wide sensor's span keep polling separately, and that the sensor at the span's tail address does not
anchor a re-use join on a mid-range predecessor (which would make it decode that sensor's bytes).
A sensor at 0x201 carrying skip_updates sits inside a widened shared-address range at 0x200 but
keeps its own range, so polling rates stay independent; folding it in would also make it decode
0x201 out of the shared response (2) instead of its own poll (777).
"""
line_callback, error_log_lines, warning_log_lines = _make_modbus_line_callback()
# 0x9001 = 0x0397 (919); 0x9001..0x9002 = 0x03970291 (60228241, approx: not exact in float32);
# 0x9002 = 0x0291 (657); 0x30 = 0x0111 (273); 0x10 = 0x0222 (546)
expected_values = {
"shared_word": 919,
"shared_dword": pytest.approx(60228241),
"covered_word": 657,
"forced_high": 273,
"plain_low": 546,
"wide_qword": 100,
"inside_wide": 321,
"tail_of_wide": 421,
"rate_word": 321,
"rate_dword": pytest.approx(21037058),
"own_rate": 777,
}
tracker = SensorTracker(list(expected_values.keys()))
futures = tracker.expect_all(expected_values)
async with (
run_compiled(yaml_config, line_callback=line_callback),
api_client_connected() as client,
):
await tracker.setup_and_start_scenario(client)
await tracker.await_all(futures)
_assert_no_modbus_errors(error_log_lines, warning_log_lines)
@pytest.mark.asyncio
async def test_uart_mock_modbus_custom_command(
yaml_config: str,
run_compiled: RunCompiledFunction,
api_client_connected: APIClientConnectedFactory,
) -> None:
"""Test a custom_command sensor polling a register served by the mock server.
The custom_command is a raw frame (device address + PDU); the hub appends the CRC and
routes the response back to the polling command, whose sensor lambda parses the payload.
Guards the custom polling wiring: the command must reference the sensor's custom_data and
decode the real function code, or nothing is ever transmitted. A plain read on the same
register anchors the bus.
"""
line_callback, error_log_lines, warning_log_lines = _make_modbus_line_callback()
expected_values = {"plain_read": 259, "custom_read": 259}
tracker = SensorTracker(list(expected_values.keys()))
futures = tracker.expect_all(expected_values)
async with (
run_compiled(yaml_config, line_callback=line_callback),
api_client_connected() as client,
):
await tracker.setup_and_start_scenario(client)
await tracker.await_all(futures)
_assert_no_modbus_errors(error_log_lines, warning_log_lines)
@pytest.mark.asyncio
async def test_uart_mock_modbus_offline(
yaml_config: str,
run_compiled: RunCompiledFunction,
api_client_connected: APIClientConnectedFactory,
) -> None:
"""A silent device drives the controller offline; answering again recovers it.
The mock answers nothing at first, so the controller burns through max_cmd_retries
(1 retry after the first timeout) and fires on_offline. While offline it keeps
retrying every offline_skip_updates+1 cycles. The test then flips the mock to
answering; the next retry gets a response, on_online fires, and the register value
publishes. This pins the pooled non-response counter, can_send() gating, the
offline retry cadence, and recovery - none of which the responding-path tests touch.
The fixture gives offline_skip_updates and the sensor's skip_updates the same period
on purpose: offline probing must follow the offline cadence alone, since requiring
both cadences to coincide leaves phase combinations where no probe ever goes out.
"""
tracker = SensorTracker(["link_state", "reg"])
offline_future = tracker.expect("link_state", 0)
async with (
run_compiled(yaml_config),
api_client_connected() as client,
):
entities = await tracker.setup_and_start_scenario(client)
# The unanswered poll and its retry each time out (~100ms), then on_offline fires.
await tracker.await_change(offline_future, "link_state", timeout=5.0)
# Register the recovery expectations before waking the device so no update is missed.
online_future = tracker.expect("link_state", 1)
value_future = tracker.expect("reg", 259)
serve_btn = find_entity(entities, "serve", ButtonInfo)
assert serve_btn is not None, "Serve button not found"
client.button_command(serve_btn.key)
# The next offline-cadence retry gets an answer: back online, value published.
await tracker.await_change(online_future, "link_state", timeout=5.0)
await tracker.await_change(value_future, "reg", timeout=5.0)
@pytest.mark.asyncio
async def test_uart_mock_modbus_fairness(
yaml_config: str,
run_compiled: RunCompiledFunction,
api_client_connected: APIClientConnectedFactory,
) -> None:
"""Two controllers sharing one bus should get a fair share of it.
Both controllers poll different devices (addresses 1 and 2) on the same
client hub, far faster than the bus can service, so they continually
contend for it. The on_tx hook in the fixture counts the requests issued
for each address. With fair scheduling in the modbus hub, neither
controller should starve the other: the two request counts must end up
within 1 of each other.
"""
tracker = SensorTracker(["requests_1", "requests_2"])
async with (
run_compiled(yaml_config),
api_client_connected() as client,
):
entities = await tracker.setup_and_start_scenario(client)
# Let both controllers hammer the bus for a while.
await asyncio.sleep(2.0)
# Stop polling so the counters settle to a final, stable value (state
# coalescing means intermediate values may be skipped, but the final
# value is always delivered once changes stop).
stop_btn = find_entity(entities, "stop_scenario", ButtonInfo)
assert stop_btn is not None, "Stop Scenario button not found"
client.button_command(stop_btn.key)
await asyncio.sleep(0.5)
assert tracker.sensor_states["requests_1"], "controller 1 issued no requests"
assert tracker.sensor_states["requests_2"], "controller 2 issued no requests"
count_1 = tracker.sensor_states["requests_1"][-1]
count_2 = tracker.sensor_states["requests_2"][-1]
# Both must have polled repeatedly, otherwise "fairness" is meaningless.
assert count_1 >= 5 and count_2 >= 5, (
f"expected both controllers to poll repeatedly, "
f"got controller 1={count_1}, controller 2={count_2}"
)
# Fair scheduling: the bus alternates between the two pending requests,
# so the counts can differ by at most one in-flight request.
assert abs(count_1 - count_2) <= 1, (
f"controllers did not get a fair share of the bus: "
f"controller 1 issued {count_1}, controller 2 issued {count_2}"
)
Reference in New Issue View Git Blame Copy Permalink