From c4b26fd3f54d1d1015ab3a21d575baf2ca2430cd Mon Sep 17 00:00:00 2001 From: "J. Nick Koston" Date: Sat, 7 Mar 2026 17:41:56 -1000 Subject: [PATCH] Add integration test for non-hardware UART and extract constants - 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 --- esphome/components/modbus/modbus.cpp | 16 ++-- .../external_components/uart_mock/__init__.py | 5 +- .../uart_mock_modbus_no_threshold.yaml | 64 +++++++++++++++ tests/integration/test_uart_mock_modbus.py | 79 +++++++++++++++++++ 4 files changed, 157 insertions(+), 7 deletions(-) create mode 100644 tests/integration/fixtures/uart_mock_modbus_no_threshold.yaml diff --git a/esphome/components/modbus/modbus.cpp b/esphome/components/modbus/modbus.cpp index da5bfa5ef8..4f683995fd 100644 --- a/esphome/components/modbus/modbus.cpp +++ b/esphome/components/modbus/modbus.cpp @@ -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 diff --git a/tests/integration/fixtures/external_components/uart_mock/__init__.py b/tests/integration/fixtures/external_components/uart_mock/__init__.py index c10d73354e..6b9d0e2497 100644 --- a/tests/integration/fixtures/external_components/uart_mock/__init__.py +++ b/tests/integration/fixtures/external_components/uart_mock/__init__.py @@ -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])) diff --git a/tests/integration/fixtures/uart_mock_modbus_no_threshold.yaml b/tests/integration/fixtures/uart_mock_modbus_no_threshold.yaml new file mode 100644 index 0000000000..407e1c858a --- /dev/null +++ b/tests/integration/fixtures/uart_mock_modbus_no_threshold.yaml @@ -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({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();' diff --git a/tests/integration/test_uart_mock_modbus.py b/tests/integration/test_uart_mock_modbus.py index 6901dc27fe..83407a6794 100644 --- a/tests/integration/test_uart_mock_modbus.py +++ b/tests/integration/test_uart_mock_modbus.py @@ -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" + )