github/esphome
2
0
Fork 0
mirror of https://github.com/esphome/esphome.git synced 2026-09-11 23:37:34 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add integration test for non-hardware UART and extract constants

Browse Source 
- Add test fixture simulating USB UART (no rx_full_threshold set)
  with a 20ms gap between response chunks
- Make rx_full_threshold optional in uart_mock
- Extract MODBUS_BITS_PER_CHAR and MS_PER_SEC constexprs
This commit is contained in:
J. Nick Koston
2026-03-07 17:41:56 -10:00
parent d13e87b5b5
commit c4b26fd3f5
4 changed files with 157 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files
esphome/components/modbus/modbus.cpp
+11 -5
View File
@@ -11,6 +11,11 @@ static const char *const TAG = "modbus";
// Maximum bytes to log for Modbus frames (truncated if larger)
static constexpr size_t MODBUS_MAX_LOG_BYTES = 64;
// Bits per character on the wire: 1 start + 8 data + 1 parity + 1 stop
static constexpr uint32_t MODBUS_BITS_PER_CHAR = 11;
// Milliseconds per second
static constexpr uint32_t MS_PER_SEC = 1000;
void Modbus::setup() {
if (this->flow_control_pin_ != nullptr) {
this->flow_control_pin_->setup();
@@ -19,16 +24,17 @@ void Modbus::setup() {
this->frame_delay_ms_ =
std::max(2, // 1750us minimum per spec - rounded up to 2ms.
// 3.5 characters * 11 bits per character * 1000ms/sec / (bits/sec) (Standard modbus frame delay)
(uint16_t) (3.5 * 11 * 1000 / this->parent_->get_baud_rate()) + 1);
(uint16_t) (3.5 * MODBUS_BITS_PER_CHAR * MS_PER_SEC / this->parent_->get_baud_rate()) + 1);
// When rx_full_threshold is configured (non-zero), the UART has a hardware FIFO with a
// meaningful threshold (e.g., ESP32 native UART), so we can calculate a precise delay.
// Otherwise (e.g., USB UART), use 50ms to handle data arriving in chunks.
static constexpr uint16_t DEFAULT_LONG_RX_BUFFER_DELAY_MS = 50;
size_t rx_threshold = this->parent_->get_rx_full_threshold();
this->long_rx_buffer_delay_ms_ = rx_threshold != uart::UARTComponent::RX_FULL_THRESHOLD_UNSET
? (rx_threshold * 11 * 1000 / this->parent_->get_baud_rate()) + 1
: DEFAULT_LONG_RX_BUFFER_DELAY_MS;
this->long_rx_buffer_delay_ms_ =
rx_threshold != uart::UARTComponent::RX_FULL_THRESHOLD_UNSET
? (rx_threshold * MODBUS_BITS_PER_CHAR * MS_PER_SEC / this->parent_->get_baud_rate()) + 1
: DEFAULT_LONG_RX_BUFFER_DELAY_MS;
}
void Modbus::loop() {
@@ -296,7 +302,7 @@ void Modbus::send_next_frame_() {
this->last_send_tx_offset_ = 0;
} else {
this->write_array(frame.data.get(), frame.size);
this->last_send_tx_offset_ = frame.size * 11 * 1000 / this->parent_->get_baud_rate() + 1;
this->last_send_tx_offset_ = frame.size * MODBUS_BITS_PER_CHAR * MS_PER_SEC / this->parent_->get_baud_rate() + 1;
}
#if ESPHOME_LOG_LEVEL >= ESPHOME_LOG_LEVEL_VERBOSE
tests/integration/fixtures/external_components/uart_mock/__init__.py
+3 -2
View File
@@ -87,7 +87,7 @@ CONFIG_SCHEMA = cv.Schema(
cv.GenerateID(): cv.declare_id(MockUartComponent),
cv.Required(CONF_BAUD_RATE): cv.int_range(min=1),
cv.Optional(CONF_RX_BUFFER_SIZE, default=256): cv.validate_bytes,
cv.Optional(CONF_RX_FULL_THRESHOLD, default=10): cv.int_range(min=1, max=120),
cv.Optional(CONF_RX_FULL_THRESHOLD): cv.int_range(min=1, max=120),
cv.Optional(CONF_RX_TIMEOUT, default=2): cv.int_range(min=0, max=92),
cv.Optional(CONF_STOP_BITS, default=1): cv.one_of(1, 2, int=True),
cv.Optional(CONF_DATA_BITS, default=8): cv.int_range(min=5, max=8),
@@ -135,7 +135,8 @@ async def to_code(config):
cg.add(var.set_baud_rate(config[CONF_BAUD_RATE]))
cg.add(var.set_rx_buffer_size(config[CONF_RX_BUFFER_SIZE]))
cg.add(var.set_rx_full_threshold(config[CONF_RX_FULL_THRESHOLD]))
if CONF_RX_FULL_THRESHOLD in config:
cg.add(var.set_rx_full_threshold(config[CONF_RX_FULL_THRESHOLD]))
cg.add(var.set_rx_timeout(config[CONF_RX_TIMEOUT]))
cg.add(var.set_stop_bits(config[CONF_STOP_BITS]))
cg.add(var.set_data_bits(config[CONF_DATA_BITS]))
tests/integration/fixtures/uart_mock_modbus_no_threshold.yaml
+64
View File
@@ -0,0 +1,64 @@
esphome:
name: uart-mock-modbus-no-threshold-test
host:
api:
logger:
level: VERBOSE
external_components:
- source:
type: local
path: EXTERNAL_COMPONENT_PATH
# Dummy uart entry to satisfy modbus's DEPENDENCIES = ["uart"]
# The actual UART bus used is the uart_mock component below
uart:
baud_rate: 115200
port: /dev/null
# Simulate a non-hardware UART (e.g., USB UART) by not setting rx_full_threshold.
# This leaves it at the default sentinel value (0), triggering the 50ms fallback timeout.
uart_mock:
- id: virtual_uart_dev
baud_rate: 9600
auto_start: false
debug:
on_tx:
- then:
- if:
condition: #Read 80 input registers on device 2, starting at address 0 (SDM meter request)
lambda: "return data == std::vector<uint8_t>({0x02,0x04,0x00,0x00,0x00,0x50,0xF0,0x05});"
then:
- uart_mock.inject_rx: # Good response from SDM meter with CRC. Voltage is 243V
!lambda return {0x02,0x04,0xA0,0x43,0x73,0x19,0x9A,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,
0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,
0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,
0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x3F,0x80,0x00,0x00,0x00,0x00,0x00,0x00,0x00,
0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,
0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,
0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00};
- delay: 20ms # Simulate USB packet gap - would fail with old ~2ms timeout
- uart_mock.inject_rx: # Rest of response (including CRC)
!lambda return{0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,
0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x42,0x6F,0xCC,0xCD,0x43,0x7C,0xB8,0x10,0x3D,0x38,0x51,0xEC,
0x43,0x81,0x1B,0xE7,0x3B,0x03,0x12,0x6F,0x50,0x1B};
modbus:
uart_id: virtual_uart_dev
turnaround_time: 10ms
sensor:
- platform: sdm_meter
address: 2
update_interval: 1s
phase_a:
voltage:
name: sdm_voltage
button:
- platform: template
name: "Start Scenario"
id: start_scenario_btn
on_press:
- lambda: 'id(virtual_uart_dev).start_scenario();'
tests/integration/test_uart_mock_modbus.py
+79
View File
@@ -5,6 +5,10 @@ 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.
"""
from __future__ import annotations
@@ -218,3 +222,78 @@ async def test_uart_mock_modbus_timing(
f"Timeout waiting for SDM voltage change. Received sensor states:\n"
f" sdm_voltage: {sensor_states['sdm_voltage']}\n"
)
@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 20ms gap
between USB packets would cause a false timeout and CRC failure cascade.
"""
# Replace external component path placeholder
external_components_path = str(
Path(__file__).parent / "fixtures" / "external_components"
)
yaml_config = yaml_config.replace(
"EXTERNAL_COMPONENT_PATH", external_components_path
)
loop = asyncio.get_running_loop()
# Track sensor state updates (after initial state is swallowed)
sensor_states: dict[str, list[float]] = {
"sdm_voltage": [],
}
voltage_changed = loop.create_future()
def on_state(state: EntityState) -> None:
if isinstance(state, SensorState) and not state.missing_state:
sensor_name = key_to_sensor.get(state.key)
if sensor_name and sensor_name in sensor_states:
sensor_states[sensor_name].append(state.state)
# Check if this is a good voltage reading (243V)
if (
sensor_name == "sdm_voltage"
and state.state > 200.0
and not voltage_changed.done()
):
voltage_changed.set_result(True)
async with (
run_compiled(yaml_config),
api_client_connected() as client,
):
entities, _ = await client.list_entities_services()
# Build key mappings for all sensor types
all_names = list(sensor_states.keys())
key_to_sensor = build_key_to_entity_mapping(entities, all_names)
# Set up initial state helper
initial_state_helper = InitialStateHelper(entities)
client.subscribe_states(initial_state_helper.on_state_wrapper(on_state))
try:
await initial_state_helper.wait_for_initial_states()
except TimeoutError:
pytest.fail("Timeout waiting for initial states")
# Start the UART mock scenario now that we're subscribed
start_btn = find_entity(entities, "start_scenario", ButtonInfo)
assert start_btn is not None, "Start Scenario button not found"
client.button_command(start_btn.key)
# Wait for voltage to be updated with successful parse
try:
await asyncio.wait_for(voltage_changed, timeout=2.0)
except TimeoutError:
pytest.fail(
f"Timeout waiting for SDM voltage change. Received sensor states:\n"
f" sdm_voltage: {sensor_states['sdm_voltage']}\n"
)