github/esphome
2
0
Fork 0
mirror of https://github.com/esphome/esphome.git synced 2026-10-06 10:56:43 +00:00
Code Issues Packages Projects Releases Wiki Activity

[core] Fold the modbus write and bits tests into the shared mesh fixture (#18946)

Browse Source
This commit is contained in:
J. Nick Koston
2026-09-17 11:23:29 +12:00
committed by GitHub
parent f8012bc467
commit abfe350119
4 changed files with 340 additions and 610 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/integration/test_uart_mock_modbus.py
+66 -74
View File
@@ -19,23 +19,40 @@ from __future__ import annotations
import asyncio
from collections.abc import Callable
from dataclasses import dataclass
from aioesphomeapi import ButtonInfo, NumberInfo, SwitchInfo, TextSensorState
import pytest
from .state_utils import SensorTracker, find_entity, wait_for_state
from .state_utils import SensorTracker, find_entity, require_entity, wait_for_state
from .types import APIClientConnectedFactory, RunCompiledFunction
@dataclass
class RegisterTestCase:
"""Test parameters for a single modbus register write/read round-trip."""
def _swap16(value: int) -> int:
"""Byte-swapped view of a 16-bit register as the raw U_WORD wire value."""
return ((value & 0xFF) << 8) | (value >> 8)
initial_value: object
write_number_name: str
write_value: float
post_write_value: object
# Raw U_WORD view of reg_u_word_s's initial 0x1234
MESH_RAW_U_WORD_S = _swap16(4660)
# Initial values of the mesh fixture's address 1 registers; the
# server_controller test reads them and the write test uses them as baseline.
MESH_INITIAL_VALUES: dict[str, object] = {
"reg_u_word": 99,
"reg_u_word_s": 4660,
"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(2.5),
}
# ---------------------------------------------------------------------------
@@ -310,23 +327,7 @@ async def test_uart_mock_modbus_server_controller(
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),
}
expected_values = MESH_INITIAL_VALUES | {"reg_u_word_s_raw": MESH_RAW_U_WORD_S}
tracker = SensorTracker(list(expected_values.keys()))
futures = tracker.expect_all(expected_values)
@@ -334,14 +335,12 @@ async def test_uart_mock_modbus_server_controller(
run_compiled(yaml_config, line_callback=line_callback),
api_client_connected() as client,
):
# The controller polls from boot, so the first values can already be in
# the states the device sends on connect; matching them there saves
# waiting for the next poll
await tracker.setup_and_start_scenario(client, match_initial_states=True)
await tracker.await_all(futures)
_assert_no_modbus_errors(error_log_lines, warning_log_lines)
@pytest.mark.shared_yaml("uart_mock_modbus_mesh")
@pytest.mark.asyncio
async def test_uart_mock_modbus_server_controller_write(
yaml_config: str,
@@ -357,51 +356,47 @@ async def test_uart_mock_modbus_server_controller_write(
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),
),
# Per read-back sensor: the number entity to write through and the value;
# floats read back within tolerance, everything else exactly
register_writes: dict[str, tuple[str, int | float]] = {
"reg_u_word": ("write_u_word", 42),
"reg_u_word_s": ("write_u_word_s", 17185),
"reg_s_word": ("write_s_word", -42),
"reg_s_word_s": ("write_s_word_s", -257),
"reg_u_dword": ("write_u_dword", 2002),
"reg_s_dword": ("write_s_dword", -2002),
"reg_u_dword_r": ("write_u_dword_r", 4004),
"reg_s_dword_r": ("write_s_dword_r", -4004),
"reg_u_qword": ("write_u_qword", 6006),
"reg_s_qword": ("write_s_qword", -6006),
"reg_u_qword_r": ("write_u_qword_r", 8008),
"reg_s_qword_r": ("write_s_qword_r", -8008),
"reg_fp32": ("write_fp32", 6.28),
"reg_fp32_r": ("write_fp32_r", 9.42),
}
tracker = SensorTracker(list(register_test_cases.keys()))
tracker = SensorTracker([*register_writes, "reg_u_word_s_raw"])
# The raw U_WORD view of 0x02 pins the byte swap on the write path: the
# round trip through write_u_word_s applies the swap an even number of
# times, so only the raw sensor can catch a symmetrically dropped swap.
# Phase 1: expect initial baseline values
initial_futures = tracker.expect_all(
{name: case.initial_value for name, case in register_test_cases.items()}
MESH_INITIAL_VALUES | {"reg_u_word_s_raw": MESH_RAW_U_WORD_S}
)
# 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()}
{
name: pytest.approx(value, abs=0.01) if isinstance(value, float) else value
for name, (_, value) in register_writes.items()
}
| {"reg_u_word_s_raw": _swap16(register_writes["reg_u_word_s"][1])}
)
async with (
run_compiled(yaml_config, line_callback=line_callback),
api_client_connected() as client,
):
# The controller polls from boot, so the baseline can already be in the
# states the device sends on connect; matching it there saves waiting for
# the next poll
entities = await tracker.setup_and_start_scenario(
client, match_initial_states=True
)
@@ -410,19 +405,22 @@ async def test_uart_mock_modbus_server_controller_write(
# 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)
# Issue write commands for all register types; exact object_id match,
# since several write_* names are prefixes of a sibling
numbers = {
e.object_id.lower(): e for e in entities if isinstance(e, NumberInfo)
}
for number_name, value in register_writes.values():
entity = numbers.get(number_name)
assert entity is not None, f"{number_name} number entity not found"
client.number_command(entity.key, 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.shared_yaml("uart_mock_modbus_mesh")
@pytest.mark.asyncio
async def test_uart_mock_modbus_server_controller_bits(
yaml_config: str,
@@ -468,8 +466,6 @@ async def test_uart_mock_modbus_server_controller_bits(
run_compiled(yaml_config, line_callback=line_callback),
api_client_connected() as client,
):
# The controller polls from boot and binary sensors drop repeats, so the
# baseline can arrive only in the states the device sends on connect
entities = await tracker.setup_and_start_scenario(
client, match_initial_states=True
)
@@ -480,8 +476,7 @@ async def test_uart_mock_modbus_server_controller_bits(
# Flip both writable bits: 0x02 false -> true, 0x03 true -> false
for switch_name, value in (("write_bit_2", True), ("write_bit_3", False)):
entity = find_entity(entities, switch_name, SwitchInfo)
assert entity is not None, f"{switch_name} switch entity not found"
entity = require_entity(entities, switch_name, SwitchInfo)
client.switch_command(entity.key, value)
# Wait for both read views to reflect the written values
@@ -508,9 +503,6 @@ async def test_uart_mock_modbus_server_controller_multiple(
run_compiled(yaml_config, line_callback=line_callback),
api_client_connected() as client,
):
# The controller polls from boot, so the first values can already be in
# the states the device sends on connect; matching them there saves
# waiting for the next poll
await tracker.setup_and_start_scenario(client, match_initial_states=True)
await tracker.await_all(futures)
_assert_no_modbus_errors(error_log_lines, warning_log_lines)