Merge remote-tracking branch 'upstream/dev' into 20260218-zigbee-proxy

This commit is contained in:
kbx81
2026-03-13 16:45:23 -05:00
969 changed files with 29931 additions and 7819 deletions
@@ -1,5 +1,7 @@
"""Tests for the binary sensor component."""
from tests.component_tests.helpers import INTERNAL_BIT, extract_packed_value
def test_binary_sensor_is_setup(generate_main):
"""
@@ -29,7 +31,7 @@ def test_binary_sensor_sets_mandatory_fields(generate_main):
)
# Then
assert 'bs_1->set_name("test bs1",' in main_cpp
assert 'bs_1->configure_entity_("test bs1",' in main_cpp
assert "bs_1->set_pin(" in main_cpp
@@ -44,9 +46,9 @@ def test_binary_sensor_config_value_internal_set(generate_main):
"tests/component_tests/binary_sensor/test_binary_sensor.yaml"
)
# Then
assert "bs_1->set_internal(true);" in main_cpp
assert "bs_2->set_internal(false);" in main_cpp
# Then: bs_1 has internal: true, bs_2 has internal: false
assert extract_packed_value(main_cpp, "bs_1") & INTERNAL_BIT != 0
assert extract_packed_value(main_cpp, "bs_2") & INTERNAL_BIT == 0
def test_binary_sensor_config_value_use_raw_set(generate_main):
+6 -4
View File
@@ -1,5 +1,7 @@
"""Tests for the button component"""
from tests.component_tests.helpers import INTERNAL_BIT, extract_packed_value
def test_button_is_setup(generate_main):
"""
@@ -26,7 +28,7 @@ def test_button_sets_mandatory_fields(generate_main):
main_cpp = generate_main("tests/component_tests/button/test_button.yaml")
# Then
assert 'wol_1->set_name("wol_test_1",' in main_cpp
assert 'wol_1->configure_entity_("wol_test_1",' in main_cpp
assert "wol_2->set_macaddr(18, 52, 86, 120, 144, 171);" in main_cpp
@@ -39,6 +41,6 @@ def test_button_config_value_internal_set(generate_main):
# When
main_cpp = generate_main("tests/component_tests/button/test_button.yaml")
# Then
assert "wol_1->set_internal(true);" in main_cpp
assert "wol_2->set_internal(false);" in main_cpp
# Then: wol_1 has internal: true, wol_2 has internal: false
assert extract_packed_value(main_cpp, "wol_1") & INTERNAL_BIT != 0
assert extract_packed_value(main_cpp, "wol_2") & INTERNAL_BIT == 0
+19
View File
@@ -0,0 +1,19 @@
"""Shared helpers for component tests."""
from __future__ import annotations
import re
INTERNAL_BIT = 1 << 24
def extract_packed_value(main_cpp: str, var_name: str) -> int:
"""Extract the third (packed) argument from a configure_entity_ call."""
pattern = (
rf"{re.escape(var_name)}->configure_entity_\("
r'"(?:\\.|[^"\\])*"'
r",\s*\w+,\s*(\d+)\)"
)
match = re.search(pattern, main_cpp)
assert match, f"configure_entity_ call not found for {var_name}"
return int(match.group(1))
@@ -0,0 +1,50 @@
"""Tests for the logger component."""
import re
def test_logger_pre_setup_before_other_components(generate_main):
"""Logger::pre_setup() must be called before any other component is created.
Log functions call global_logger->log_vprintf_() without a null check,
so global_logger must be set before anything can log.
"""
main_cpp = generate_main("tests/component_tests/logger/test_logger.yaml")
# Find the logger's pre_setup() call specifically
logger_pre_setup = re.search(r"logger_logger->pre_setup\(\)", main_cpp)
if logger_pre_setup is None:
# Fall back to finding any logger-related pre_setup
logger_pre_setup = re.search(r"logger\w*->pre_setup\(\)", main_cpp)
assert logger_pre_setup is not None, (
"Logger pre_setup() not found in generated code"
)
# Find all "new " allocations (component creation)
new_allocations = list(re.finditer(r"\bnew [\w:]+", main_cpp))
assert len(new_allocations) > 0, "No component allocations found"
# Separate logger and non-logger allocations
logger_allocs = [a for a in new_allocations if "logger" in a.group().lower()]
non_logger_allocs = [
a
for a in new_allocations
if "logger" not in a.group().lower()
# Skip placement new for App
and "(&App)" not in main_cpp[max(0, a.start() - 5) : a.start()]
]
assert len(logger_allocs) > 0, (
f"Logger allocation not found in: {[a.group() for a in new_allocations]}"
)
assert len(non_logger_allocs) > 0, (
"No non-logger component allocations found — "
"add a component to test_logger.yaml so the ordering check is meaningful"
)
# All non-logger allocations must appear after logger pre_setup()
for alloc in non_logger_allocs:
assert alloc.start() > logger_pre_setup.start(), (
f"Component allocation '{alloc.group()}' at position {alloc.start()} "
f"appears before logger pre_setup() at position {logger_pre_setup.start()}"
)
@@ -0,0 +1,14 @@
---
esphome:
name: test
esp8266:
board: d1_mini_lite
logger:
level: DEBUG
# Need at least one non-logger component so the ordering test
# can verify that logger pre_setup() comes before other allocations.
preferences:
flash_write_interval: 1min
+5 -2
View File
@@ -1,5 +1,7 @@
"""Tests for the sensor component."""
from tests.component_tests.helpers import extract_packed_value
def test_sensor_device_class_set(generate_main):
"""
@@ -10,5 +12,6 @@ def test_sensor_device_class_set(generate_main):
# When
main_cpp = generate_main("tests/component_tests/sensor/test_sensor.yaml")
# Then
assert 's_1->set_device_class("voltage");' in main_cpp
# Then: device_class: voltage means packed value must be non-zero
packed = extract_packed_value(main_cpp, "s_1")
assert packed != 0
+23 -6
View File
@@ -1,4 +1,6 @@
"""Tests for the binary sensor component."""
"""Tests for the text component."""
from tests.component_tests.helpers import INTERNAL_BIT, extract_packed_value
def test_text_is_setup(generate_main):
@@ -25,7 +27,7 @@ def test_text_sets_mandatory_fields(generate_main):
main_cpp = generate_main("tests/component_tests/text/test_text.yaml")
# Then
assert 'it_1->set_name("test 1 text",' in main_cpp
assert 'it_1->configure_entity_("test 1 text",' in main_cpp
def test_text_config_value_internal_set(generate_main):
@@ -37,9 +39,9 @@ def test_text_config_value_internal_set(generate_main):
# When
main_cpp = generate_main("tests/component_tests/text/test_text.yaml")
# Then
assert "it_2->set_internal(false);" in main_cpp
assert "it_3->set_internal(true);" in main_cpp
# Then: it_2 has internal: false, it_3 has internal: true
assert extract_packed_value(main_cpp, "it_2") & INTERNAL_BIT == 0
assert extract_packed_value(main_cpp, "it_3") & INTERNAL_BIT != 0
def test_text_config_value_mode_set(generate_main):
@@ -66,5 +68,20 @@ def test_text_config_lamda_is_set(generate_main):
main_cpp = generate_main("tests/component_tests/text/test_text.yaml")
# Then
assert "it_4->set_template([]() -> esphome::optional<std::string> {" in main_cpp
assert "it_4->set_template([]() -> std::optional<std::string> {" in main_cpp
assert 'return std::string{"Hello"};' in main_cpp
def test_esphome_optional_alias_works(generate_main):
"""
Test that esphome::optional alias compiles (backward compatibility)
"""
# Given
# When
main_cpp = generate_main("tests/component_tests/text/test_text.yaml")
# Then
# Codegen emits std::optional, but esphome::optional must also work
# via the using alias in esphome/core/optional.h
assert "std::optional<std::string>" in main_cpp
@@ -1,5 +1,7 @@
"""Tests for the text sensor component."""
from tests.component_tests.helpers import INTERNAL_BIT, extract_packed_value
def test_text_sensor_is_setup(generate_main):
"""
@@ -25,9 +27,9 @@ def test_text_sensor_sets_mandatory_fields(generate_main):
main_cpp = generate_main("tests/component_tests/text_sensor/test_text_sensor.yaml")
# Then
assert 'ts_1->set_name("Template Text Sensor 1",' in main_cpp
assert 'ts_2->set_name("Template Text Sensor 2",' in main_cpp
assert 'ts_3->set_name("Template Text Sensor 3",' in main_cpp
assert 'ts_1->configure_entity_("Template Text Sensor 1",' in main_cpp
assert 'ts_2->configure_entity_("Template Text Sensor 2",' in main_cpp
assert 'ts_3->configure_entity_("Template Text Sensor 3",' in main_cpp
def test_text_sensor_config_value_internal_set(generate_main):
@@ -39,9 +41,9 @@ def test_text_sensor_config_value_internal_set(generate_main):
# When
main_cpp = generate_main("tests/component_tests/text_sensor/test_text_sensor.yaml")
# Then
assert "ts_2->set_internal(true);" in main_cpp
assert "ts_3->set_internal(false);" in main_cpp
# Then: ts_2 has internal: true, ts_3 has internal: false
assert extract_packed_value(main_cpp, "ts_2") & INTERNAL_BIT != 0
assert extract_packed_value(main_cpp, "ts_3") & INTERNAL_BIT == 0
def test_text_sensor_device_class_set(generate_main):
@@ -53,6 +55,9 @@ def test_text_sensor_device_class_set(generate_main):
# When
main_cpp = generate_main("tests/component_tests/text_sensor/test_text_sensor.yaml")
# Then
assert 'ts_2->set_device_class("timestamp");' in main_cpp
assert 'ts_3->set_device_class("date");' in main_cpp
# Then: ts_2 has device_class: timestamp, ts_3 has device_class: date
# so their packed values must be non-zero
packed_ts_2 = extract_packed_value(main_cpp, "ts_2")
assert packed_ts_2 != 0
packed_ts_3 = extract_packed_value(main_cpp, "ts_3")
assert packed_ts_3 != 0
+13
View File
@@ -7,10 +7,23 @@
testing binaries that combine many components. By convention, this unique namespace is `esphome::component::testing`
(where "component" is the component under test), for example: `esphome::uart::testing`.
### Platform components
For components that expose to a platform component, create a folder under your component test folder with the platform component name, e.g. `binary_sensor` and
include the relevant `.cpp` and `.h` test files there.
### Override component code generation for testing
When generating code for testing, ESPHome won't invoke the component's `to_code` function, since most components do not
need to generate configuration code for testing.
If you do need to generate code to for example configure compilation flags or add libraries,
add the component name to the `CPP_TESTING_CODEGEN_COMPONENTS` allowlist in `script/cpp_unit_test.py`.
## Running component unit tests
(from the repository root)
```bash
./script/cpp_unit_test.py component1 component2 ...
```
@@ -0,0 +1,11 @@
sensor:
- id: my_sensor
platform: adc
pin: VCC
name: ADC Test sensor
update_interval: "1:01"
unit_of_measurement: "°C"
icon: "mdi:water-percent"
accuracy_decimals: 5
setup_priority: -100
force_update: true
@@ -0,0 +1,5 @@
<<: !include common.yaml
wifi:
ssid: MySSID
password: password1
+9
View File
@@ -0,0 +1,9 @@
audio_file:
- id: test_audio
file:
type: local
path: $component_dir/test.wav
media_source:
- platform: audio_file
id: audio_file_source
@@ -0,0 +1 @@
<<: !include common.yaml
Binary file not shown.
@@ -0,0 +1,4 @@
ble_nus:
type: uart
tx_buffer_size: 160
rx_buffer_size: 160
@@ -0,0 +1 @@
<<: !include common.yaml
+19
View File
@@ -0,0 +1,19 @@
sensor:
- platform: dew_point
name: Dew Point
temperature: template_temperature
humidity: template_humidity
- platform: template
id: template_humidity
lambda: |-
if (millis() > 10000) {
return 0.6;
}
return 0.0;
- platform: template
id: template_temperature
lambda: |-
if (millis() > 10000) {
return 42.0;
}
return 0.0;
@@ -0,0 +1 @@
<<: !include common.yaml
@@ -0,0 +1 @@
<<: !include common.yaml
@@ -0,0 +1 @@
<<: !include common.yaml
@@ -1,3 +1,6 @@
esp8266:
enable_full_printf: false
logger:
level: VERBOSE
+2 -2
View File
@@ -2,10 +2,10 @@ ethernet:
type: W5500
clk_pin: 19
mosi_pin: 21
miso_pin: 23
miso_pin: 17
cs_pin: 18
interrupt_pin: 36
reset_pin: 22
reset_pin: 12
clock_speed: 10Mhz
manual_ip:
static_ip: 192.168.178.56
@@ -0,0 +1 @@
<<: !include common-ip101.yaml
@@ -0,0 +1 @@
<<: !include common-w5500.yaml
@@ -1,6 +1,8 @@
external_components:
- source: github://esphome/esphome@dev
- id: my_ext
source: github://esphome/esphome@dev
refresh: 1d
components: [bh1750]
- source: ../../../esphome/components
- id: my_local
source: ../../../esphome/components
components: [sntp]
@@ -1 +1,5 @@
# WARNING: Using !extend or !remove prevents automatic component grouping in CI, making builds slower.
<<: !include common.yaml
external_components:
- id: !remove my_local
+11
View File
@@ -27,3 +27,14 @@ globals:
type: bool
restore_value: false
initial_value: "false"
# Test restore_value with string "false" - should be converted to bool false
- id: glob_no_restore_string_false
type: int
restore_value: "false"
initial_value: "42"
# Test restore_value with string "true" - should be converted to bool true
- id: glob_restore_string_true
type: int
restore_value: "true"
initial_value: "99"
update_interval: 5s
@@ -1,4 +1,6 @@
substitutions:
verify_ssl: "false"
<<: !include common.yaml
http_request:
verify_ssl: false
tls_buffer_size_rx: 16384
tls_buffer_size_tx: 512
@@ -1,9 +1,19 @@
esphome:
on_boot:
then:
- sensor.integration.reset:
id: integration_sensor
- sensor.integration.set_value:
id: integration_sensor
value: 100.0
sensor:
- platform: adc
id: my_sensor
pin: ${pin}
attenuation: 12db
- platform: integration
id: integration_sensor
sensor: my_sensor
name: Integration Sensor
time_unit: s
+1 -1
View File
@@ -16,7 +16,7 @@ class MockUARTComponent : public uart::UARTComponent {
MOCK_METHOD(bool, read_array, (uint8_t * data, size_t len), (override));
MOCK_METHOD(bool, peek_byte, (uint8_t * data), (override));
MOCK_METHOD(size_t, available, (), (override));
MOCK_METHOD(void, flush, (), (override));
MOCK_METHOD(uart::FlushResult, flush, (), (override));
MOCK_METHOD(void, check_logger_conflict, (), (override));
};
+6
View File
@@ -60,6 +60,12 @@ esphome:
}
}
# Test set_effect with const char* doesn't cause ambiguous overload (issue #14728)
- lambda: |-
auto call = id(test_monochromatic_light).turn_on();
call.set_effect("None");
call.perform();
- light.toggle: test_binary_light
- light.turn_off: test_rgb_light
- light.turn_on:
+8
View File
@@ -1,5 +1,7 @@
#include <gtest/gtest.h>
#include "esphome/components/logger/logger.h"
/*
This special main.cpp replaces the default one.
It will run all the Google Tests found in all compiled cpp files and then exit with the result
@@ -18,6 +20,12 @@ void original_setup() {
}
void setup() {
// Log functions call global_logger->log_vprintf_() without a null check,
// so we must set up a Logger before any test that triggers logging.
static esphome::logger::Logger test_logger(0);
test_logger.set_log_level(ESPHOME_LOG_LEVEL);
test_logger.pre_setup();
::testing::InitGoogleTest();
int exit_code = RUN_ALL_TESTS();
exit(exit_code);
+2
View File
@@ -1,3 +1,5 @@
modbus:
id: mod_bus1
flow_control_pin: ${flow_control_pin}
send_wait_time: 500ms
turnaround_time: 100ms
+59
View File
@@ -0,0 +1,59 @@
#include <gtest/gtest.h>
#include "esphome/components/modbus/modbus.h"
#include "esphome/core/helpers.h"
namespace esphome::modbus {
// Exposes protected methods for testing.
class TestModbus : public Modbus {
public:
bool test_parse_modbus_byte(uint8_t byte) { return this->parse_modbus_byte_(byte); }
void test_clear_rx_buffer() { this->rx_buffer_.clear(); }
void set_waiting(uint8_t addr) { this->waiting_for_response_ = addr; }
};
class MockDevice : public ModbusDevice {
public:
void on_modbus_data(const std::vector<uint8_t> &data) override { this->data_received = true; }
bool data_received{false};
};
TEST(ModbusTest, TwoByteRegressionTest) {
TestModbus modbus;
modbus.set_role(ModbusRole::CLIENT);
// First byte (at=0)
EXPECT_TRUE(modbus.test_parse_modbus_byte(0x01));
// Second byte (at=1)
// This used to reach raw[2] because it skipped the if(at==2) check, causing a
// buffer overflow.
EXPECT_TRUE(modbus.test_parse_modbus_byte(0x03));
}
TEST(ModbusTest, TestValidFrame) {
TestModbus modbus;
modbus.set_role(ModbusRole::CLIENT);
MockDevice device;
device.set_parent(&modbus);
device.set_address(0x01);
modbus.register_device(&device);
modbus.set_waiting(0x01);
// Address 1, Function 3, Length 2, Data 0x1234
uint8_t frame_data[] = {0x01, 0x03, 0x02, 0x12, 0x34};
uint16_t crc = esphome::crc16(frame_data, sizeof(frame_data));
std::vector<uint8_t> frame;
for (uint8_t b : frame_data)
frame.push_back(b);
frame.push_back(crc & 0xFF);
frame.push_back((crc >> 8) & 0xFF);
for (size_t i = 0; i < frame.size(); i++) {
bool result = modbus.test_parse_modbus_byte(frame[i]);
EXPECT_TRUE(result) << "Failed at byte " << i << " (0x" << std::hex << (int) frame[i] << ")";
}
EXPECT_TRUE(device.data_received);
}
} // namespace esphome::modbus
@@ -0,0 +1,5 @@
display:
- id: !extend main_lcd
tft_url: http://esphome.io/default35.tft
tft_upload_http_timeout: 20s
tft_upload_http_retries: 10
@@ -0,0 +1,3 @@
display:
- id: !extend main_lcd
tft_upload_watchdog_timeout: 30s
+2 -4
View File
@@ -1,7 +1,5 @@
packages:
uart: !include ../../test_build_components/common/uart/esp32-ard.yaml
base: !include common.yaml
display:
- id: !extend main_lcd
tft_url: http://esphome.io/default35.tft
tft_upload: !include common_tft_upload.yaml
tft_upload_watchdog: !include common_tft_upload_watchdog.yaml
+2 -4
View File
@@ -1,7 +1,5 @@
packages:
uart: !include ../../test_build_components/common/uart/esp32-idf.yaml
base: !include common.yaml
display:
- id: !extend main_lcd
tft_url: http://esphome.io/default35.tft
tft_upload: !include common_tft_upload.yaml
tft_upload_watchdog: !include common_tft_upload_watchdog.yaml
@@ -1,7 +1,4 @@
packages:
uart: !include ../../test_build_components/common/uart/esp8266-ard.yaml
base: !include common.yaml
display:
- id: !extend main_lcd
tft_url: http://esphome.io/default35.tft
tft_upload: !include common_tft_upload.yaml
@@ -40,6 +40,10 @@ online_image:
url: https://samples-files.com/samples/images/bmp/480-360-sample.bmp
format: BMP
type: BINARY
- id: online_rgb_bmp_8bit
url: https://samples-files.com/samples/images/bmp/480-360-sample.bmp
format: BMP
type: RGB
- id: online_jpeg_image
url: http://www.faqs.org/images/library.jpg
format: JPEG
@@ -1,3 +1,9 @@
esp32:
board: esp32-c6-devkitc-1
framework:
type: esp-idf
log_level: DEBUG
network:
enable_ipv6: true
@@ -13,3 +19,4 @@ openthread:
force_dataset: true
use_address: open-thread-test.local
poll_period: 20sec
output_power: 1dBm
@@ -0,0 +1,27 @@
zephyr_ble_server:
ota:
- platform: zephyr_mcumgr
transport:
ble: true
hardware_uart: CDC
on_begin:
then:
- logger.log: "OTA start"
on_progress:
then:
- logger.log:
format: "OTA progress %0.1f%%"
args: ["x"]
on_end:
then:
- logger.log: "OTA end"
on_error:
then:
- logger.log:
format: "OTA update error %d"
args: ["x"]
on_state_change:
then:
lambda: >-
ESP_LOGD("ota", "State %d", state);
@@ -0,0 +1,77 @@
#include "../common.h"
namespace esphome::packet_transport::testing {
TEST(PacketTransportBinarySensorTest, AddBinarySensor) {
TestablePacketTransport transport;
binary_sensor::BinarySensor bs;
transport.add_binary_sensor("motion", &bs);
ASSERT_EQ(transport.binary_sensors_.size(), 1u);
EXPECT_STREQ(transport.binary_sensors_[0].id, "motion");
EXPECT_EQ(transport.binary_sensors_[0].sensor, &bs);
}
TEST(PacketTransportBinarySensorTest, AddRemoteBinarySensor) {
TestablePacketTransport transport;
binary_sensor::BinarySensor bs;
transport.add_remote_binary_sensor("host1", "remote_motion", &bs);
EXPECT_TRUE(transport.providers_.contains("host1"));
EXPECT_EQ(transport.remote_binary_sensors_["host1"]["remote_motion"], &bs);
}
TEST(PacketTransportBinarySensorTest, UnencryptedBinarySensorRoundTrip) {
TestablePacketTransport encoder;
encoder.init_for_test("sender");
binary_sensor::BinarySensor local_bs;
local_bs.state = true;
encoder.add_binary_sensor("motion", &local_bs);
encoder.send_data_(true);
ASSERT_EQ(encoder.sent_packets.size(), 1u);
TestablePacketTransport decoder;
decoder.init_for_test("receiver");
binary_sensor::BinarySensor remote_bs;
decoder.add_remote_binary_sensor("sender", "motion", &remote_bs);
auto &packet = encoder.sent_packets[0];
decoder.process_({packet.data(), packet.size()});
EXPECT_TRUE(remote_bs.state);
}
TEST(PacketTransportBinarySensorTest, MultipleSensorsRoundTrip) {
TestablePacketTransport encoder;
encoder.init_for_test("sender");
sensor::Sensor s1, s2;
s1.state = 10.0f;
s2.state = 20.0f;
encoder.add_sensor("s1", &s1);
encoder.add_sensor("s2", &s2);
binary_sensor::BinarySensor bs1;
bs1.state = true;
encoder.add_binary_sensor("bs1", &bs1);
encoder.send_data_(true);
ASSERT_EQ(encoder.sent_packets.size(), 1u);
TestablePacketTransport decoder;
decoder.init_for_test("receiver");
sensor::Sensor rs1, rs2;
binary_sensor::BinarySensor rbs1;
rs1.state = -999.0f;
rs2.state = -999.0f;
decoder.add_remote_sensor("sender", "s1", &rs1);
decoder.add_remote_sensor("sender", "s2", &rs2);
decoder.add_remote_binary_sensor("sender", "bs1", &rbs1);
auto &packet = encoder.sent_packets[0];
decoder.process_({packet.data(), packet.size()});
EXPECT_FLOAT_EQ(rs1.state, 10.0f);
EXPECT_FLOAT_EQ(rs2.state, 20.0f);
EXPECT_TRUE(rbs1.state);
}
} // namespace esphome::packet_transport::testing
@@ -0,0 +1,98 @@
#pragma once
#include <cstdint>
#include <cstring>
#include <cstdio>
#include <vector>
#include <gtest/gtest.h>
#include "esphome/components/packet_transport/packet_transport.h"
namespace esphome::packet_transport::testing {
// Protocol constants mirrored from packet_transport.cpp for test packet construction.
static constexpr uint16_t MAGIC_NUMBER = 0x4553;
static constexpr uint16_t MAGIC_PING = 0x5048;
// Concrete testable implementation of PacketTransport.
// Captures sent packets and exposes protected members for verification.
//
// Sensor round-trip tests require USE_SENSOR / USE_BINARY_SENSOR to be defined,
// which happens when 'sensor' and 'binary_sensor' components are in the build.
// Run with --all or include those components to enable the full test suite.
class TestablePacketTransport : public PacketTransport {
public:
using PacketTransport::add_key_;
using PacketTransport::data_;
using PacketTransport::encryption_key_;
using PacketTransport::flush_;
using PacketTransport::header_;
using PacketTransport::increment_code_;
using PacketTransport::init_data_;
using PacketTransport::is_encrypted_;
using PacketTransport::is_provider_;
using PacketTransport::name_;
using PacketTransport::ping_key_;
using PacketTransport::ping_keys_;
using PacketTransport::ping_pong_enable_;
using PacketTransport::ping_pong_recyle_time_;
using PacketTransport::process_;
using PacketTransport::providers_;
using PacketTransport::rolling_code_;
using PacketTransport::rolling_code_enable_;
using PacketTransport::send_data_;
using PacketTransport::updated_;
#ifdef USE_SENSOR
using PacketTransport::add_data_;
using PacketTransport::remote_sensors_;
using PacketTransport::sensors_;
#endif
#ifdef USE_BINARY_SENSOR
using PacketTransport::add_binary_data_;
using PacketTransport::binary_sensors_;
using PacketTransport::remote_binary_sensors_;
#endif
// NOTE: std::vector is used here for test convenience. For production code,
// consider using StaticVector or FixedVector from esphome/core/helpers.h instead.
mutable std::vector<std::vector<uint8_t>> sent_packets;
size_t max_packet_size{512};
bool send_enabled{true};
void send_packet(const std::vector<uint8_t> &buf) const override { this->sent_packets.push_back(buf); }
size_t get_max_packet_size() override { return this->max_packet_size; }
bool should_send() override { return this->send_enabled; }
/// Build the packet header for testing without requiring App or global_preferences.
void init_for_test(const char *name) {
this->name_ = name;
this->header_.clear();
// MAGIC_NUMBER as uint16_t little-endian
this->header_.push_back(MAGIC_NUMBER & 0xFF);
this->header_.push_back((MAGIC_NUMBER >> 8) & 0xFF);
// Length-prefixed hostname
auto len = strlen(name);
this->header_.push_back(static_cast<uint8_t>(len));
for (size_t i = 0; i < len; i++)
this->header_.push_back(name[i]);
// Pad to 4-byte boundary
while (this->header_.size() & 0x3)
this->header_.push_back(0);
}
};
/// Build a MAGIC_PING packet for testing add_key_ / ping-pong flows.
inline std::vector<uint8_t> build_ping_packet(const char *hostname, uint32_t key) {
std::vector<uint8_t> packet;
packet.push_back(MAGIC_PING & 0xFF);
packet.push_back((MAGIC_PING >> 8) & 0xFF);
auto len = strlen(hostname);
packet.push_back(static_cast<uint8_t>(len));
for (size_t i = 0; i < len; i++)
packet.push_back(hostname[i]);
packet.push_back(key & 0xFF);
packet.push_back((key >> 8) & 0xFF);
packet.push_back((key >> 16) & 0xFF);
packet.push_back((key >> 24) & 0xFF);
return packet;
}
} // namespace esphome::packet_transport::testing
@@ -0,0 +1,186 @@
#include "common.h"
namespace esphome::packet_transport::testing {
// --- Configuration setter tests ---
TEST(PacketTransportTest, SetIsProvider) {
TestablePacketTransport transport;
transport.set_is_provider(true);
EXPECT_TRUE(transport.is_provider_);
}
TEST(PacketTransportTest, SetEncryptionKey) {
TestablePacketTransport transport;
std::vector<uint8_t> key(32, 0xAB);
transport.set_encryption_key(key);
EXPECT_EQ(transport.encryption_key_, key);
EXPECT_TRUE(transport.is_encrypted_());
}
TEST(PacketTransportTest, NoEncryptionByDefault) {
TestablePacketTransport transport;
EXPECT_FALSE(transport.is_encrypted_());
}
TEST(PacketTransportTest, SetRollingCodeEnable) {
TestablePacketTransport transport;
transport.set_rolling_code_enable(true);
EXPECT_TRUE(transport.rolling_code_enable_);
}
TEST(PacketTransportTest, SetPingPongEnable) {
TestablePacketTransport transport;
transport.set_ping_pong_enable(true);
EXPECT_TRUE(transport.ping_pong_enable_);
}
TEST(PacketTransportTest, SetPingPongRecycleTime) {
TestablePacketTransport transport;
transport.set_ping_pong_recycle_time(600);
EXPECT_EQ(transport.ping_pong_recyle_time_, 600u);
}
// --- Provider management ---
TEST(PacketTransportTest, AddProvider) {
TestablePacketTransport transport;
transport.add_provider("host1");
EXPECT_TRUE(transport.providers_.contains("host1"));
EXPECT_EQ(transport.providers_.size(), 1u);
}
TEST(PacketTransportTest, AddProviderDuplicate) {
TestablePacketTransport transport;
transport.add_provider("host1");
transport.add_provider("host1");
EXPECT_EQ(transport.providers_.size(), 1u);
}
TEST(PacketTransportTest, SetProviderEncryption) {
TestablePacketTransport transport;
transport.add_provider("host1");
std::vector<uint8_t> key(32, 0xCD);
transport.set_provider_encryption("host1", key);
EXPECT_EQ(transport.providers_["host1"].encryption_key, key);
}
// --- Ping key tests ---
TEST(PacketTransportTest, PingKeyStoredWhenEncrypted) {
TestablePacketTransport transport;
transport.init_for_test("receiver");
transport.set_encryption_key(std::vector<uint8_t>(32, 0xAA));
auto ping = build_ping_packet("requester", 0xDEADBEEF);
transport.process_({ping.data(), ping.size()});
ASSERT_EQ(transport.ping_keys_.size(), 1u);
EXPECT_EQ(transport.ping_keys_["requester"], 0xDEADBEEFu);
}
TEST(PacketTransportTest, PingKeyIgnoredWhenNotEncrypted) {
TestablePacketTransport transport;
transport.init_for_test("receiver");
// No encryption key — add_key_ should be a no-op
auto ping = build_ping_packet("requester", 0xDEADBEEF);
transport.process_({ping.data(), ping.size()});
EXPECT_TRUE(transport.ping_keys_.empty());
}
TEST(PacketTransportTest, PingKeyUpdatedOnRepeat) {
TestablePacketTransport transport;
transport.init_for_test("receiver");
transport.set_encryption_key(std::vector<uint8_t>(32, 0xAA));
auto ping1 = build_ping_packet("host1", 0x1111);
transport.process_({ping1.data(), ping1.size()});
EXPECT_EQ(transport.ping_keys_["host1"], 0x1111u);
// Same host, new key value — should update in place
auto ping2 = build_ping_packet("host1", 0x2222);
transport.process_({ping2.data(), ping2.size()});
EXPECT_EQ(transport.ping_keys_.size(), 1u);
EXPECT_EQ(transport.ping_keys_["host1"], 0x2222u);
}
TEST(PacketTransportTest, PingKeyMaxLimit) {
TestablePacketTransport transport;
transport.init_for_test("receiver");
transport.set_encryption_key(std::vector<uint8_t>(32, 0xAA));
// Fill to MAX_PING_KEYS (4)
for (int i = 0; i < 4; i++) {
char name[16];
snprintf(name, sizeof(name), "host%d", i);
auto ping = build_ping_packet(name, 0x1000 + i);
transport.process_({ping.data(), ping.size()});
}
EXPECT_EQ(transport.ping_keys_.size(), 4u);
// 5th key should be discarded
auto ping = build_ping_packet("host4", 0x9999);
transport.process_({ping.data(), ping.size()});
EXPECT_EQ(transport.ping_keys_.size(), 4u);
EXPECT_FALSE(transport.ping_keys_.contains("host4"));
}
// --- Process error handling ---
TEST(PacketTransportTest, ProcessShortBuffer) {
TestablePacketTransport transport;
transport.init_for_test("receiver");
uint8_t buf[] = {0x53};
// Too short for a magic number - should return safely
transport.process_({buf, 1});
}
TEST(PacketTransportTest, ProcessBadMagic) {
TestablePacketTransport transport;
transport.init_for_test("receiver");
uint8_t buf[] = {0xFF, 0xFF, 0x00, 0x00};
// Wrong magic - should return safely
transport.process_({buf, sizeof(buf)});
}
TEST(PacketTransportTest, ProcessOwnHostname) {
TestablePacketTransport transport;
transport.init_for_test("myself");
// Build a packet from "myself" using a separate encoder
TestablePacketTransport fake_sender;
fake_sender.init_for_test("myself");
fake_sender.send_data_(true);
ASSERT_EQ(fake_sender.sent_packets.size(), 1u);
auto &packet = fake_sender.sent_packets[0];
// Should be silently ignored because hostname matches our own
transport.process_({packet.data(), packet.size()});
}
TEST(PacketTransportTest, ProcessUnknownHostname) {
TestablePacketTransport transport;
transport.init_for_test("receiver");
// No providers registered - "unknown" will not be found
TestablePacketTransport sender;
sender.init_for_test("unknown");
sender.send_data_(true);
ASSERT_EQ(sender.sent_packets.size(), 1u);
auto &packet = sender.sent_packets[0];
// Should return safely without crash
transport.process_({packet.data(), packet.size()});
}
// --- Send disabled ---
TEST(PacketTransportTest, NoSendWhenDisabled) {
TestablePacketTransport transport;
transport.init_for_test("sender");
transport.send_enabled = false;
transport.send_data_(true);
EXPECT_TRUE(transport.sent_packets.empty());
}
} // namespace esphome::packet_transport::testing
@@ -0,0 +1,170 @@
#include "../common.h"
namespace esphome::packet_transport::testing {
TEST(PacketTransportSensorTest, AddSensor) {
TestablePacketTransport transport;
sensor::Sensor s;
transport.add_sensor("temp", &s);
ASSERT_EQ(transport.sensors_.size(), 1u);
EXPECT_STREQ(transport.sensors_[0].id, "temp");
EXPECT_EQ(transport.sensors_[0].sensor, &s);
EXPECT_TRUE(transport.sensors_[0].updated);
}
TEST(PacketTransportSensorTest, AddRemoteSensor) {
TestablePacketTransport transport;
sensor::Sensor s;
transport.add_remote_sensor("host1", "remote_temp", &s);
EXPECT_TRUE(transport.providers_.contains("host1"));
EXPECT_EQ(transport.remote_sensors_["host1"]["remote_temp"], &s);
}
TEST(PacketTransportSensorTest, UnencryptedSensorRoundTrip) {
// Encoder
TestablePacketTransport encoder;
encoder.init_for_test("sender");
sensor::Sensor local_sensor;
local_sensor.state = 42.5f;
encoder.add_sensor("temp", &local_sensor);
encoder.send_data_(true);
ASSERT_EQ(encoder.sent_packets.size(), 1u);
// Decoder
TestablePacketTransport decoder;
decoder.init_for_test("receiver");
sensor::Sensor remote_sensor;
remote_sensor.state = -999.0f; // sentinel
decoder.add_remote_sensor("sender", "temp", &remote_sensor);
auto &packet = encoder.sent_packets[0];
decoder.process_({packet.data(), packet.size()});
EXPECT_FLOAT_EQ(remote_sensor.state, 42.5f);
}
TEST(PacketTransportSensorTest, EncryptedSensorRoundTrip) {
std::vector<uint8_t> key(32);
for (int i = 0; i < 32; i++)
key[i] = i;
TestablePacketTransport encoder;
encoder.init_for_test("sender");
encoder.set_encryption_key(key);
sensor::Sensor local_sensor;
local_sensor.state = 99.9f;
encoder.add_sensor("temp", &local_sensor);
encoder.send_data_(true);
ASSERT_EQ(encoder.sent_packets.size(), 1u);
TestablePacketTransport decoder;
decoder.init_for_test("receiver");
sensor::Sensor remote_sensor;
remote_sensor.state = -999.0f;
decoder.add_remote_sensor("sender", "temp", &remote_sensor);
decoder.set_provider_encryption("sender", key);
auto &packet = encoder.sent_packets[0];
decoder.process_({packet.data(), packet.size()});
EXPECT_FLOAT_EQ(remote_sensor.state, 99.9f);
}
TEST(PacketTransportSensorTest, SendDataOnlyUpdated) {
TestablePacketTransport encoder;
encoder.init_for_test("sender");
sensor::Sensor s1, s2;
s1.state = 1.0f;
s2.state = 2.0f;
encoder.add_sensor("s1", &s1);
encoder.add_sensor("s2", &s2);
// Mark s1 as not updated, only s2 as updated
encoder.sensors_[0].updated = false;
encoder.sensors_[1].updated = true;
encoder.send_data_(false);
ASSERT_EQ(encoder.sent_packets.size(), 1u);
TestablePacketTransport decoder;
decoder.init_for_test("receiver");
sensor::Sensor rs1, rs2;
rs1.state = -999.0f;
rs2.state = -999.0f;
decoder.add_remote_sensor("sender", "s1", &rs1);
decoder.add_remote_sensor("sender", "s2", &rs2);
auto &packet = encoder.sent_packets[0];
decoder.process_({packet.data(), packet.size()});
EXPECT_FLOAT_EQ(rs1.state, -999.0f); // not updated, not sent
EXPECT_FLOAT_EQ(rs2.state, 2.0f); // updated, sent
}
TEST(PacketTransportSensorTest, PingKeyIncludedInTransmittedPacket) {
std::vector<uint8_t> key(32, 0xBB);
// Responder: encrypted, owns a sensor
TestablePacketTransport responder;
responder.init_for_test("responder");
responder.set_encryption_key(key);
sensor::Sensor local_sensor;
local_sensor.state = 77.7f;
responder.add_sensor("temp", &local_sensor);
// Requester sends a MAGIC_PING that the responder processes
auto ping = build_ping_packet("requester", 0xDEADBEEF);
responder.process_({ping.data(), ping.size()});
ASSERT_EQ(responder.ping_keys_.size(), 1u);
// Responder sends sensor data — ping key should be embedded
responder.send_data_(true);
ASSERT_EQ(responder.sent_packets.size(), 1u);
// Requester: encrypted provider, ping-pong enabled, expects key 0xDEADBEEF
TestablePacketTransport requester;
requester.init_for_test("requester");
requester.set_ping_pong_enable(true);
requester.ping_key_ = 0xDEADBEEF;
sensor::Sensor remote_sensor;
remote_sensor.state = -999.0f;
requester.add_remote_sensor("responder", "temp", &remote_sensor);
requester.set_provider_encryption("responder", key);
// The requester decrypts the packet and finds its ping key echoed back,
// which gates the sensor data — if the key is missing, data is blocked.
auto &packet = responder.sent_packets[0];
requester.process_({packet.data(), packet.size()});
EXPECT_FLOAT_EQ(remote_sensor.state, 77.7f);
}
TEST(PacketTransportSensorTest, MissingPingKeyBlocksSensorData) {
std::vector<uint8_t> key(32, 0xBB);
// Responder sends data WITHOUT receiving any MAGIC_PING first — no ping keys
TestablePacketTransport responder;
responder.init_for_test("responder");
responder.set_encryption_key(key);
sensor::Sensor local_sensor;
local_sensor.state = 77.7f;
responder.add_sensor("temp", &local_sensor);
responder.send_data_(true);
ASSERT_EQ(responder.sent_packets.size(), 1u);
// Requester with ping-pong enabled expects a key that isn't in the packet
TestablePacketTransport requester;
requester.init_for_test("requester");
requester.set_ping_pong_enable(true);
requester.ping_key_ = 0xDEADBEEF;
sensor::Sensor remote_sensor;
remote_sensor.state = -999.0f;
requester.add_remote_sensor("responder", "temp", &remote_sensor);
requester.set_provider_encryption("responder", key);
auto &packet = responder.sent_packets[0];
requester.process_({packet.data(), packet.size()});
EXPECT_FLOAT_EQ(remote_sensor.state, -999.0f); // blocked — ping key not found
}
} // namespace esphome::packet_transport::testing
@@ -1,3 +1,6 @@
rp2040:
enable_full_printf: false
logger:
level: VERBOSE
+1
View File
@@ -0,0 +1 @@
rp2040_ble:
@@ -0,0 +1,2 @@
rp2040_ble:
enable_on_boot: false
@@ -0,0 +1 @@
<<: !include common.yaml
+10
View File
@@ -0,0 +1,10 @@
wifi:
ssid: MySSID
password: password1
api:
serial_proxy:
- id: serial_proxy_1
name: Test Serial Port
port_type: RS232
@@ -0,0 +1,8 @@
substitutions:
tx_pin: GPIO4
rx_pin: GPIO5
packages:
uart: !include ../../test_build_components/common/uart/esp32-idf.yaml
<<: !include common.yaml
@@ -0,0 +1,8 @@
substitutions:
tx_pin: GPIO0
rx_pin: GPIO2
packages:
uart: !include ../../test_build_components/common/uart/esp8266-ard.yaml
<<: !include common.yaml
@@ -0,0 +1,8 @@
substitutions:
tx_pin: GPIO4
rx_pin: GPIO5
packages:
uart: !include ../../test_build_components/common/uart/rp2040-ard.yaml
<<: !include common.yaml
@@ -0,0 +1,18 @@
<<: !include common.yaml
media_player:
- platform: speaker
id: speaker_media_player_id
announcement_pipeline:
speaker: speaker_id
buffer_size: 1000000
volume_increment: 0.02
volume_max: 0.95
volume_min: 0.0
task_stack_in_psram: true
on_turn_on:
then:
- logger.log: "Turn On Media Player"
on_turn_off:
then:
- logger.log: "Turn Off Media Player"
@@ -0,0 +1,9 @@
substitutions:
scl_pin: GPIO16
sda_pin: GPIO17
i2s_bclk_pin: GPIO27
i2s_lrclk_pin: GPIO26
i2s_mclk_pin: GPIO25
i2s_dout_pin: GPIO23
<<: !include common-media_player_off_on.yaml
@@ -0,0 +1,61 @@
i2s_audio:
i2s_lrclk_pin: ${i2s_bclk_pin}
i2s_bclk_pin: ${i2s_lrclk_pin}
i2s_mclk_pin: ${i2s_mclk_pin}
speaker:
- platform: i2s_audio
id: speaker_id
dac_type: external
i2s_dout_pin: ${i2s_dout_pin}
sample_rate: 48000
num_channels: 2
- platform: mixer
output_speaker: speaker_id
source_speakers:
- id: announcement_mixer_speaker_id
- id: media_mixer_speaker_id
audio_file:
- id: test_audio
file:
type: local
path: $component_dir/test.wav
media_source:
- platform: audio_file
id: announcement_audio_file_source
- platform: audio_file
id: media_audio_file_source
media_player:
- platform: speaker_source
id: media_player_id
name: Media Player
volume_increment: 0.02
volume_initial: 0.75
volume_max: 0.95
volume_min: 0.0
announcement_pipeline:
speaker: announcement_mixer_speaker_id
format: FLAC
num_channels: 1
sources:
- announcement_audio_file_source
media_pipeline:
speaker: media_mixer_speaker_id
format: FLAC
num_channels: 1
sources:
- media_audio_file_source
on_mute:
- media_player.shuffle:
id: media_player_id
on_unmute:
- media_player.unshuffle:
id: media_player_id
on_volume:
- speaker_source.set_playlist_delay:
id: media_player_id
pipeline: media
delay: 500ms
@@ -0,0 +1,9 @@
substitutions:
scl_pin: GPIO16
sda_pin: GPIO17
i2s_bclk_pin: GPIO27
i2s_lrclk_pin: GPIO26
i2s_mclk_pin: GPIO25
i2s_dout_pin: GPIO23
<<: !include common.yaml
Binary file not shown.
@@ -0,0 +1,6 @@
substitutions:
clk_pin: GPIO2
mosi_pin: GPIO3
miso_pin: GPIO4
<<: !include common.yaml
+6 -1
View File
@@ -28,9 +28,14 @@ esphome:
# Test C++ API: set_template() with stateless lambda (no captures)
# NOTE: set_template() is not intended to be a public API, but we test it to ensure it doesn't break.
- lambda: |-
id(template_sens).set_template([]() -> esphome::optional<float> {
id(template_sens).set_template([]() -> std::optional<float> {
return 123.0f;
});
# Test that esphome::optional alias still works for backward compatibility
- lambda: |-
id(template_sens).set_template([]() -> esphome::optional<float> {
return 42.0f;
});
- datetime.date.set:
id: test_date
+1 -1
View File
@@ -30,7 +30,7 @@ class MockUARTComponent : public UARTComponent {
MOCK_METHOD(bool, read_array, (uint8_t * data, size_t len), (override));
MOCK_METHOD(bool, peek_byte, (uint8_t * data), (override));
MOCK_METHOD(size_t, available, (), (override));
MOCK_METHOD(void, flush, (), (override));
MOCK_METHOD(FlushResult, flush, (), (override));
MOCK_METHOD(void, check_logger_conflict, (), (override));
};
+13 -3
View File
@@ -1,11 +1,15 @@
esphome:
on_boot:
then:
- uart.write: 'Hello World'
- uart.write: [0x00, 0x20, 0x42]
- uart.write:
id: uart_id
data: 'Hello World'
- uart.write:
id: uart_id
data: [0x00, 0x20, 0x42]
uart:
- id: uart_uart
- id: uart_id
tx_pin: 4
rx_pin: 5
flow_control_pin: 6
@@ -16,3 +20,9 @@ uart:
rx_timeout: 1
parity: EVEN
stop_bits: 2
- id: uart_debug
tx_pin: 18
rx_pin: 19
baud_rate: 115200
debug:
debug_prefix: "[UART1] "
+45 -15
View File
@@ -1,13 +1,19 @@
esphome:
on_boot:
then:
- uart.write: 'Hello World'
- uart.write: [0x00, 0x20, 0x42]
- uart.write: !lambda |-
return {0xAA, 0xBB, 0xCC};
- uart.write:
id: uart_id
data: 'Hello World'
- uart.write:
id: uart_id
data: [0x00, 0x20, 0x42]
- uart.write:
id: uart_id
data: !lambda |-
return {0xAA, 0xBB, 0xCC};
uart:
- id: uart_uart
- id: uart_id
tx_pin: 17
rx_pin: 16
flow_control_pin: 4
@@ -18,32 +24,54 @@ uart:
rx_timeout: 1
parity: EVEN
stop_bits: 2
- id: uart_debug
tx_pin: 21
rx_pin: 22
baud_rate: 115200
debug:
debug_prefix: "[UART1] "
- id: uart_debug_custom
tx_pin: 25
rx_pin: 26
baud_rate: 9600
debug:
debug_prefix: "[UART2] "
after:
delimiter: "\n"
sequence:
- lambda: UARTDebug::log_string(direction, bytes, debug_prefix);
- id: uart_debug_no_prefix
tx_pin: 32
rx_pin: 33
baud_rate: 9600
debug:
packet_transport:
- platform: uart
uart_id: uart_id
switch:
# Test uart switch with single state (array)
- platform: uart
name: "UART Switch Single Array"
uart_id: uart_uart
uart_id: uart_id
data: [0x01, 0x02, 0x03]
# Test uart switch with single state (string)
- platform: uart
name: "UART Switch Single String"
uart_id: uart_uart
uart_id: uart_id
data: "ON"
# Test uart switch with turn_on/turn_off (arrays)
- platform: uart
name: "UART Switch Dual Array"
uart_id: uart_uart
uart_id: uart_id
data:
turn_on: [0xA0, 0xA1, 0xA2]
turn_off: [0xB0, 0xB1, 0xB2]
# Test uart switch with turn_on/turn_off (strings)
- platform: uart
name: "UART Switch Dual String"
uart_id: uart_uart
uart_id: uart_id
data:
turn_on: "TURN_ON"
turn_off: "TURN_OFF"
@@ -61,24 +89,26 @@ button:
# Test uart button with array data
- platform: uart
name: "UART Button Array"
uart_id: uart_uart
uart_id: uart_id
data: [0xFF, 0xEE, 0xDD]
# Test uart button with string data
- platform: uart
name: "UART Button String"
uart_id: uart_uart
uart_id: uart_id
data: "BUTTON_PRESS"
# Test uart button with lambda (function pointer)
- platform: template
name: "UART Lambda Test"
on_press:
- uart.write: !lambda |-
std::string cmd = "VALUE=" + str_sprintf("%.0f", id(test_number).state) + "\r\n";
return std::vector<uint8_t>(cmd.begin(), cmd.end());
- uart.write:
id: uart_id
data: !lambda |-
std::string cmd = "VALUE=" + str_sprintf("%.0f", id(test_number).state) + "\r\n";
return std::vector<uint8_t>(cmd.begin(), cmd.end());
event:
- platform: uart
uart_id: uart_uart
uart_id: uart_id
name: "UART Event"
event_types:
- "string_event_A": "*A#"
+17 -7
View File
@@ -1,11 +1,15 @@
esphome:
on_boot:
then:
- uart.write: 'Hello World'
- uart.write: [0x00, 0x20, 0x42]
- uart.write:
id: uart_id
data: 'Hello World'
- uart.write:
id: uart_id
data: [0x00, 0x20, 0x42]
uart:
- id: uart_uart
- id: uart_id
tx_pin: 4
rx_pin: 5
baud_rate: 9600
@@ -13,15 +17,21 @@ uart:
rx_buffer_size: 512
parity: EVEN
stop_bits: 2
- id: uart_debug
tx_pin: 14
rx_pin: 12
baud_rate: 115200
debug:
debug_prefix: "[UART1] "
switch:
- platform: uart
name: "UART Switch Array"
uart_id: uart_uart
uart_id: uart_id
data: [0x01, 0x02, 0x03]
- platform: uart
name: "UART Switch Dual"
uart_id: uart_uart
uart_id: uart_id
data:
turn_on: [0xA0, 0xA1]
turn_off: [0xB0, 0xB1]
@@ -29,12 +39,12 @@ switch:
button:
- platform: uart
name: "UART Button"
uart_id: uart_uart
uart_id: uart_id
data: [0xFF, 0xEE]
event:
- platform: uart
uart_id: uart_uart
uart_id: uart_id
name: "UART Event"
event_types:
- "string_event_A": "*A#"
+1 -1
View File
@@ -5,7 +5,7 @@ esphome:
- uart.write: [0x00, 0x20, 0x42]
uart:
- id: uart_uart
- id: uart_id
port: "/dev/ttyS0"
baud_rate: 9600
data_bits: 8
+16 -3
View File
@@ -1,11 +1,15 @@
esphome:
on_boot:
then:
- uart.write: 'Hello World'
- uart.write: [0x00, 0x20, 0x42]
- uart.write:
id: uart_id
data: 'Hello World'
- uart.write:
id: uart_id
data: [0x00, 0x20, 0x42]
uart:
- id: uart_uart
- id: uart_id
tx_pin: 4
rx_pin: 5
baud_rate: 9600
@@ -13,3 +17,12 @@ uart:
rx_buffer_size: 512
parity: EVEN
stop_bits: 2
- id: uart_debug
tx_pin: 8
rx_pin: 9
baud_rate: 115200
debug:
debug_prefix: "[UART1] "
- id: uart_rx_only
rx_pin: 17
baud_rate: 1200
+8
View File
@@ -34,3 +34,11 @@ usb_uart:
- id: channel_4_1
debug: true
dummy_receiver: true
debug_prefix: "[ESP_JTAG] "
- id: uart_5
type: cp210x
channels:
- id: channel_5_1
baud_rate: 9600
debug: true
debug_prefix: "[CP210X] "
+8
View File
@@ -19,6 +19,10 @@ binary_sensor:
name: BS2 Sensor 3 Error
sensor4_error:
name: BS2 Sensor 4 Error
- platform: vbus
model: deltasol_cs4
sensor1_error:
name: "DeltaSol CS4 Sensor 1 Error"
- platform: vbus
model: custom
command: 0x100
@@ -44,6 +48,10 @@ sensor:
name: DeltaSol C Heat Quantity
time:
name: DeltaSol C System Time
- platform: vbus
model: deltasol_cs4
temperature_1:
name: "DeltaSol CS4 Temperature 1"
- platform: vbus
model: deltasol_bs2
temperature_1:
+3 -1
View File
@@ -12,7 +12,9 @@
using namespace esphome;
void setup() {
App.pre_setup("livingroom", "LivingRoom", false);
static char name[] = "livingroom";
static char friendly_name[] = "LivingRoom";
App.pre_setup(name, sizeof(name) - 1, friendly_name, sizeof(friendly_name) - 1);
auto *log = new logger::Logger(115200); // NOLINT
log->pre_setup();
log->set_uart_selection(logger::UART_SELECTION_UART0);
+4 -7
View File
@@ -73,11 +73,6 @@ def shared_platformio_cache() -> Generator[Path]:
test_cache_dir = Path.home() / ".esphome-integration-tests"
cache_dir = test_cache_dir / "platformio"
# Create the temp directory that PlatformIO uses to avoid race conditions
# This ensures it exists and won't be deleted by parallel processes
platformio_tmp_dir = cache_dir / ".cache" / "tmp"
platformio_tmp_dir.mkdir(parents=True, exist_ok=True)
# Use a lock file in the home directory to ensure only one process initializes the cache
# This is needed when running with pytest-xdist
# The lock file must be in a directory that already exists to avoid race conditions
@@ -87,8 +82,9 @@ def shared_platformio_cache() -> Generator[Path]:
with open(lock_file, "w") as lock_fd:
fcntl.flock(lock_fd.fileno(), fcntl.LOCK_EX)
# Check if cache needs initialization while holding the lock
if not cache_dir.exists() or not any(cache_dir.iterdir()):
# Check if the native platform is installed (the actual indicator of a populated cache)
native_platform = cache_dir / "platforms" / "native"
if not native_platform.exists():
# Create the test cache directory if it doesn't exist
test_cache_dir.mkdir(exist_ok=True)
@@ -197,6 +193,7 @@ async def yaml_config(request: pytest.FixtureRequest, unused_tcp_port: int) -> s
" platformio_options:\n"
" build_flags:\n"
' - "-DDEBUG" # Enable assert() statements\n'
' - "-DESPHOME_DEBUG" # Enable ESPHOME_DEBUG_ASSERT checks\n'
' - "-DESPHOME_DEBUG_API" # Enable API protocol asserts\n'
' - "-g" # Add debug symbols',
)
@@ -0,0 +1,174 @@
from esphome import automation
import esphome.codegen as cg
from esphome.components import uart
from esphome.components.const import CONF_DATA_BITS, CONF_PARITY, CONF_STOP_BITS
from esphome.components.uart import (
CONF_RX_FULL_THRESHOLD,
CONF_RX_TIMEOUT,
debug_to_code,
maybe_empty_debug,
validate_raw_data,
)
import esphome.config_validation as cv
from esphome.const import (
CONF_BAUD_RATE,
CONF_DATA,
CONF_DEBUG,
CONF_DELAY,
CONF_ID,
CONF_INTERVAL,
CONF_RX_BUFFER_SIZE,
CONF_TRIGGER_ID,
)
from esphome.core import ID
CODEOWNERS = ["@esphome/tests"]
MULTI_CONF = True
uart_mock_ns = cg.esphome_ns.namespace("uart_mock")
MockUartComponent = uart_mock_ns.class_(
"MockUartComponent", uart.UARTComponent, cg.Component
)
MockUartInjectRXAction = uart_mock_ns.class_(
"MockUartInjectRXAction", automation.Action
)
MockUartTXTrigger = uart_mock_ns.class_(
"MockUartTXTrigger",
automation.Trigger.template(cg.std_vector.template(cg.uint8)),
)
CONF_INJECTIONS = "injections"
CONF_RESPONSES = "responses"
CONF_INJECT_RX = "inject_rx"
CONF_EXPECT_TX = "expect_tx"
CONF_PERIODIC_RX = "periodic_rx"
CONF_ON_TX = "on_tx"
CONF_AUTO_START = "auto_start"
UART_PARITY_OPTIONS = {
"NONE": uart.UARTParityOptions.UART_CONFIG_PARITY_NONE,
"EVEN": uart.UARTParityOptions.UART_CONFIG_PARITY_EVEN,
"ODD": uart.UARTParityOptions.UART_CONFIG_PARITY_ODD,
}
INJECTION_SCHEMA = cv.Schema(
{
cv.Required(CONF_INJECT_RX): [cv.hex_uint8_t],
cv.Optional(CONF_DELAY, default="0ms"): cv.positive_time_period_milliseconds,
}
)
CONFIG_INJECT_RX_SCHEMA = cv.maybe_simple_value(
{
cv.GenerateID(): cv.use_id(MockUartComponent),
cv.Required("data"): cv.templatable(validate_raw_data),
cv.Optional(CONF_DELAY): cv.positive_time_period_milliseconds,
},
key=CONF_DATA,
)
RESPONSE_SCHEMA = cv.Schema(
{
cv.Required(CONF_EXPECT_TX): [cv.hex_uint8_t],
cv.Required(CONF_INJECT_RX): [cv.hex_uint8_t],
cv.Optional(CONF_DELAY, default="0ms"): cv.positive_time_period_milliseconds,
}
)
PERIODIC_RX_SCHEMA = cv.Schema(
{
cv.Required(CONF_DATA): [cv.hex_uint8_t],
cv.Required(CONF_INTERVAL): cv.positive_time_period_milliseconds,
}
)
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): 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),
cv.Optional(CONF_PARITY, default="NONE"): cv.enum(
UART_PARITY_OPTIONS, upper=True
),
cv.Optional(CONF_INJECTIONS, default=[]): cv.ensure_list(INJECTION_SCHEMA),
cv.Optional(CONF_RESPONSES, default=[]): cv.ensure_list(RESPONSE_SCHEMA),
cv.Optional(CONF_PERIODIC_RX, default=[]): cv.ensure_list(PERIODIC_RX_SCHEMA),
cv.Optional(CONF_AUTO_START, default=True): cv.boolean,
cv.Optional(CONF_ON_TX): automation.validate_automation(
{
cv.GenerateID(CONF_TRIGGER_ID): cv.declare_id(MockUartTXTrigger),
}
),
cv.Optional(CONF_DEBUG): maybe_empty_debug,
}
).extend(cv.COMPONENT_SCHEMA)
@automation.register_action(
"uart_mock.inject_rx", MockUartInjectRXAction, CONFIG_INJECT_RX_SCHEMA
)
async def inject_rx_to_code(config, action_id, template_arg, args):
var = cg.new_Pvariable(action_id, template_arg)
await cg.register_parented(var, config[CONF_ID])
data = config[CONF_DATA]
if isinstance(data, bytes):
data = list(data)
if cg.is_template(data):
templ = await cg.templatable(data, args, cg.std_vector.template(cg.uint8))
cg.add(var.set_data_template(templ))
else:
# Generate static array in flash to avoid RAM copy
arr_id = ID(f"{action_id}_data", is_declaration=True, type=cg.uint8)
arr = cg.static_const_array(arr_id, cg.ArrayInitializer(*data))
cg.add(var.set_data_static(arr, len(data)))
if CONF_DELAY in config:
cg.add(var.set_delay(config[CONF_DELAY]))
return var
async def to_code(config):
var = cg.new_Pvariable(config[CONF_ID])
await cg.register_component(var, config)
cg.add(var.set_baud_rate(config[CONF_BAUD_RATE]))
cg.add(var.set_rx_buffer_size(config[CONF_RX_BUFFER_SIZE]))
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]))
cg.add(var.set_parity(config[CONF_PARITY]))
if not config[CONF_AUTO_START]:
cg.add(var.set_auto_start(False))
for injection in config[CONF_INJECTIONS]:
rx_data = injection[CONF_INJECT_RX]
delay_ms = injection[CONF_DELAY]
cg.add(var.add_injection(rx_data, delay_ms))
for response in config[CONF_RESPONSES]:
tx_data = response[CONF_EXPECT_TX]
rx_data = response[CONF_INJECT_RX]
delay_ms = response[CONF_DELAY]
cg.add(var.add_response(tx_data, rx_data, delay_ms))
for periodic in config[CONF_PERIODIC_RX]:
data = periodic[CONF_DATA]
interval = periodic[CONF_INTERVAL]
cg.add(var.add_periodic_rx(data, interval))
for conf in config.get(CONF_ON_TX, []):
trigger = cg.new_Pvariable(conf[CONF_TRIGGER_ID], var)
await automation.build_automation(
trigger, [(cg.std_vector.template(cg.uint8), "data")], conf
)
if CONF_DEBUG in config:
await debug_to_code(config[CONF_DEBUG], var)
@@ -0,0 +1,63 @@
#pragma once
#include "esphome/core/component.h"
#include "esphome/core/automation.h"
#include "uart_mock.h"
namespace esphome::uart_mock {
// This pattern is similar to UARTWriteAction but calls inject_rx instead of write_array, and is parented to VirtualUART
// instead of UARTComponent
template<typename... Ts> class MockUartInjectRXAction : public Action<Ts...>, public Parented<MockUartComponent> {
public:
void set_data_template(std::vector<uint8_t> (*func)(Ts...)) {
// Stateless lambdas (generated by ESPHome) implicitly convert to function pointers
this->code_.func = func;
this->len_ = -1; // Sentinel value indicates template mode
}
// Store pointer to static data in flash (no RAM copy)
void set_data_static(const uint8_t *data, size_t len) {
this->code_.data = data;
this->len_ = len; // Length >= 0 indicates static mode
}
void set_delay(uint32_t delay_ms) { this->delay_ms_ = delay_ms; }
void play(const Ts &...x) override {
if (this->len_ >= 0) {
// Static mode: use pointer and length
if (this->delay_ms_ > 0) {
std::vector<uint8_t> data(this->code_.data, this->code_.data + this->len_);
this->parent_->inject_to_rx_buffer_delayed(data, this->delay_ms_);
} else {
this->parent_->inject_to_rx_buffer(this->code_.data, static_cast<size_t>(this->len_));
}
} else {
// Template mode: call function
auto val = this->code_.func(x...);
if (this->delay_ms_ > 0) {
this->parent_->inject_to_rx_buffer_delayed(val, this->delay_ms_);
} else {
this->parent_->inject_to_rx_buffer(val);
}
}
}
protected:
uint32_t delay_ms_{0};
ssize_t len_{-1}; // -1 = template mode, >=0 = static mode with length
union Code {
std::vector<uint8_t> (*func)(Ts...); // Function pointer (stateless lambdas)
const uint8_t *data; // Pointer to static data in flash
} code_;
};
class MockUartTXTrigger : public Trigger<std::vector<uint8_t>> {
public:
explicit MockUartTXTrigger(MockUartComponent *parent) {
parent->set_tx_hook([this](std::vector<uint8_t> data) { this->trigger(data); });
}
};
} // namespace esphome::uart_mock
@@ -0,0 +1,232 @@
// Host-only test component — do not copy to production code.
// See uart_mock.h for details.
#include "uart_mock.h"
#include "esphome/core/application.h"
#include "esphome/core/helpers.h"
#include "esphome/core/log.h"
namespace esphome::uart_mock {
static const char *const TAG = "uart_mock";
void MockUartComponent::setup() {
ESP_LOGI(TAG, "Mock UART initialized with %zu injections, %zu responses, %zu periodic", this->injections_.size(),
this->responses_.size(), this->periodic_rx_.size());
}
void MockUartComponent::loop() {
if (!this->loop_started_) {
this->loop_started_ = true;
if (this->auto_start_) {
this->start_scenario();
} else {
ESP_LOGD(TAG, "Scenario waiting for manual start");
}
}
if (!this->scenario_active_) {
return;
}
uint32_t now = App.get_loop_component_start_time();
// Process at most ONE timed injection per loop iteration.
// This ensures each injection is in a separate loop cycle, giving the consuming
// component (e.g., LD2410) a chance to process each batch independently.
if (this->injection_index_ < this->injections_.size()) {
auto &injection = this->injections_[this->injection_index_];
uint32_t total_delay = this->cumulative_delay_ms_ + injection.delay_ms;
if (now - this->scenario_start_ms_ >= total_delay) {
ESP_LOGD(TAG, "Injecting %zu RX bytes (injection %u)", injection.rx_data.size(), this->injection_index_);
this->inject_to_rx_buffer(injection.rx_data);
this->cumulative_delay_ms_ += injection.delay_ms;
this->injection_index_++;
}
}
// Process periodic RX
for (auto &periodic : this->periodic_rx_) {
if (now - periodic.last_inject_ms >= periodic.interval_ms) {
this->inject_to_rx_buffer(periodic.data);
periodic.last_inject_ms = now;
}
}
// Process staged RX - deliver bytes whose delay has elapsed
uint32_t now_ms = millis();
while (!this->staged_rx_.empty() && (static_cast<int32_t>(now_ms - this->staged_rx_.front().available_at_ms) >= 0)) {
auto &staged = this->staged_rx_.front();
ESP_LOGD(TAG, "Delivering %zu staged RX bytes", staged.data.size());
this->inject_to_rx_buffer(staged.data);
this->staged_rx_.pop_front();
}
// Process delayed responses
for (auto &response : this->responses_) {
if (response.delay_ms > 0 && response.last_match_ms > 0 && now - response.last_match_ms >= response.delay_ms) {
ESP_LOGD(TAG, "Injecting %zu RX bytes for delayed response", response.inject_rx.size());
this->inject_to_rx_buffer(response.inject_rx);
response.last_match_ms = 0; // Reset to prevent repeated injection
}
}
}
void MockUartComponent::start_scenario() {
uint32_t now = App.get_loop_component_start_time();
this->scenario_active_ = true;
this->scenario_start_ms_ = now;
this->cumulative_delay_ms_ = 0;
this->injection_index_ = 0;
this->tx_buffer_.clear();
for (auto &periodic : this->periodic_rx_) {
periodic.last_inject_ms = now;
}
ESP_LOGD(TAG, "Scenario started at %u ms", now);
}
void MockUartComponent::dump_config() {
ESP_LOGCONFIG(TAG,
"Mock UART Component:\n"
" Baud Rate: %u\n"
" Injections: %zu\n"
" Responses: %zu\n"
" Periodic RX: %zu",
this->baud_rate_, this->injections_.size(), this->responses_.size(), this->periodic_rx_.size());
}
void MockUartComponent::write_array(const uint8_t *data, size_t len) {
this->tx_count_ += len;
this->tx_buffer_.insert(this->tx_buffer_.end(), data, data + len);
// Log all TX data so tests can verify what the component sends
if (len > 0 && len <= 64) {
char hex_buf[format_hex_pretty_size(64)];
ESP_LOGD(TAG, "TX %zu bytes: %s", len, format_hex_pretty_to(hex_buf, sizeof(hex_buf), data, len));
} else if (len > 64) {
ESP_LOGD(TAG, "TX %zu bytes (too large to log)", len);
}
#ifdef USE_UART_DEBUGGER
for (size_t i = 0; i < len; i++) {
this->debug_callback_.call(uart::UART_DIRECTION_TX, data[i]);
}
#endif
// Responses are always active - they are request-response pairs triggered by
// component TX, not timed injections. No race condition with test subscription.
this->try_match_response_();
// This directly calls a tx_hook (lambda) as an alternative to the simpler match_response mechanism.
if (this->tx_hook_) {
std::vector<uint8_t> buf(data, data + len);
this->tx_hook_(buf);
}
}
bool MockUartComponent::peek_byte(uint8_t *data) {
if (this->rx_buffer_.empty()) {
return false;
}
*data = this->rx_buffer_.front();
return true;
}
bool MockUartComponent::read_array(uint8_t *data, size_t len) {
if (this->rx_buffer_.size() < len) {
return false;
}
for (size_t i = 0; i < len; i++) {
data[i] = this->rx_buffer_.front();
this->rx_buffer_.pop_front();
}
this->rx_count_ += len;
#ifdef USE_UART_DEBUGGER
for (size_t i = 0; i < len; i++) {
this->debug_callback_.call(uart::UART_DIRECTION_RX, data[i]);
}
#endif
return true;
}
size_t MockUartComponent::available() { return this->rx_buffer_.size(); }
uart::FlushResult MockUartComponent::flush() {
// Nothing to flush in mock
return uart::FlushResult::ASSUMED_SUCCESS;
}
void MockUartComponent::set_rx_full_threshold(size_t rx_full_threshold) {
this->rx_full_threshold_ = rx_full_threshold;
}
void MockUartComponent::set_rx_timeout(size_t rx_timeout) { this->rx_timeout_ = rx_timeout; }
void MockUartComponent::add_injection(const std::vector<uint8_t> &rx_data, uint32_t delay_ms) {
this->injections_.push_back({rx_data, delay_ms});
}
void MockUartComponent::add_response(const std::vector<uint8_t> &expect_tx, const std::vector<uint8_t> &inject_rx,
uint32_t delay_ms) {
this->responses_.push_back({expect_tx, inject_rx, delay_ms, 0});
}
void MockUartComponent::add_periodic_rx(const std::vector<uint8_t> &data, uint32_t interval_ms) {
this->periodic_rx_.push_back({data, interval_ms, 0});
}
void MockUartComponent::try_match_response_() {
for (auto &response : this->responses_) {
if (this->tx_buffer_.size() < response.expect_tx.size()) {
continue;
}
// Check if tx_buffer_ ends with expect_tx
size_t offset = this->tx_buffer_.size() - response.expect_tx.size();
if (std::equal(response.expect_tx.begin(), response.expect_tx.end(), this->tx_buffer_.begin() + offset)) {
ESP_LOGD(TAG, "TX match found, injecting %zu RX bytes", response.inject_rx.size());
if (response.delay_ms > 0) {
ESP_LOGD(TAG, "Delaying response by %u ms", response.delay_ms);
// Schedule the response injection as a future injection
response.last_match_ms = App.get_loop_component_start_time();
} else {
this->inject_to_rx_buffer(response.inject_rx);
}
this->tx_buffer_.clear();
return;
}
}
}
void MockUartComponent::inject_to_rx_buffer(const uint8_t *data, size_t len) {
std::vector<uint8_t> vec(data, data + len);
this->inject_to_rx_buffer(vec);
}
void MockUartComponent::inject_to_rx_buffer(const std::vector<uint8_t> &data) {
// Log injected RX data so tests can see what's being fed to the component
if (!data.empty() && data.size() <= 64) {
char hex_buf[format_hex_pretty_size(64)];
ESP_LOGD(TAG, "RX inject %zu bytes: %s", data.size(),
format_hex_pretty_to(hex_buf, sizeof(hex_buf), data.data(), data.size()));
} else if (data.size() > 64) {
ESP_LOGD(TAG, "RX inject %zu bytes (too large to log inline)", data.size());
}
for (uint8_t byte : data) {
this->rx_buffer_.push_back(byte);
}
}
void MockUartComponent::inject_to_rx_buffer_delayed(const std::vector<uint8_t> &data, uint32_t delay_ms) {
if (!data.empty() && data.size() <= 64) {
char hex_buf[format_hex_pretty_size(64)];
ESP_LOGD(TAG, "Staging %zu RX bytes with %ums delay: %s", data.size(), delay_ms,
format_hex_pretty_to(hex_buf, sizeof(hex_buf), data.data(), data.size()));
} else if (data.size() > 64) {
ESP_LOGD(TAG, "Staging %zu RX bytes with %ums delay (too large to log inline)", data.size(), delay_ms);
}
this->staged_rx_.push_back({data, millis() + delay_ms});
}
} // namespace esphome::uart_mock
@@ -0,0 +1,103 @@
#pragma once
// ============================================================================
// HOST-ONLY TEST COMPONENT — DO NOT COPY TO PRODUCTION CODE
//
// This component runs exclusively on the host platform for integration testing.
// It intentionally uses std::vector, std::deque, and dynamic allocation which
// would be inappropriate for production embedded components. Do not use this
// code as a reference for writing ESPHome components targeting real hardware.
// ============================================================================
#include "esphome/core/component.h"
#include "esphome/components/uart/uart_component.h"
#include <deque>
#include <vector>
namespace esphome::uart_mock {
class MockUartComponent : public uart::UARTComponent, public Component {
public:
void setup() override;
void loop() override;
void dump_config() override;
float get_setup_priority() const override { return setup_priority::BUS; }
// UARTComponent interface
void write_array(const uint8_t *data, size_t len) override;
bool peek_byte(uint8_t *data) override;
bool read_array(uint8_t *data, size_t len) override;
size_t available() override;
uart::FlushResult flush() override;
void set_rx_full_threshold(size_t rx_full_threshold) override;
void set_rx_timeout(size_t rx_timeout) override;
// Scenario configuration - called from generated code
void add_injection(const std::vector<uint8_t> &rx_data, uint32_t delay_ms);
void add_response(const std::vector<uint8_t> &expect_tx, const std::vector<uint8_t> &inject_rx,
uint32_t delay_ms = 0);
void add_periodic_rx(const std::vector<uint8_t> &data, uint32_t interval_ms);
void start_scenario();
void set_auto_start(bool auto_start) { this->auto_start_ = auto_start; }
void set_tx_hook(std::function<void(const std::vector<uint8_t> &)> &&cb) { this->tx_hook_ = std::move(cb); }
void inject_to_rx_buffer(const std::vector<uint8_t> &data);
void inject_to_rx_buffer(const uint8_t *data, size_t len);
// Stage bytes for delayed delivery - simulates transport-level latency (e.g., USB packets)
void inject_to_rx_buffer_delayed(const std::vector<uint8_t> &data, uint32_t delay_ms);
protected:
void check_logger_conflict() override {}
void try_match_response_();
// Timed injections
struct Injection {
std::vector<uint8_t> rx_data;
uint32_t delay_ms;
};
std::vector<Injection> injections_;
uint32_t injection_index_{0};
uint32_t scenario_start_ms_{0};
uint32_t cumulative_delay_ms_{0};
bool loop_started_{false};
bool auto_start_{true};
bool scenario_active_{false};
// TX-triggered responses
struct Response {
std::vector<uint8_t> expect_tx;
std::vector<uint8_t> inject_rx;
uint32_t delay_ms;
uint32_t last_match_ms{0};
};
std::vector<Response> responses_;
std::vector<uint8_t> tx_buffer_;
// RX buffer
std::deque<uint8_t> rx_buffer_;
// Periodic RX
struct PeriodicRx {
std::vector<uint8_t> data;
uint32_t interval_ms;
uint32_t last_inject_ms{0};
};
std::vector<PeriodicRx> periodic_rx_;
// Staged RX - bytes that are pending delivery after a delay
// Simulates transport-level latency (e.g., USB packet delivery)
struct StagedRx {
std::vector<uint8_t> data;
uint32_t available_at_ms; // millis() time when bytes become available
};
std::deque<StagedRx> staged_rx_;
// Observability
uint32_t tx_count_{0};
uint32_t rx_count_{0};
// Direct TX hook for tests that want to bypass the response-matching logic
std::function<void(const std::vector<uint8_t> &)> tx_hook_;
};
} // namespace esphome::uart_mock
@@ -0,0 +1,61 @@
esphome:
name: micros-to-millis-test
platformio_options:
build_flags:
- "-DDEBUG"
on_boot:
- lambda: |-
using esphome::micros_to_millis;
const char *TAG = "MTM";
int pass = 0, fail = 0;
auto check = [&](const char *name, uint64_t us) {
uint32_t got = micros_to_millis(us);
uint32_t want = (uint32_t)(us / 1000ULL);
if (got == want) { pass++; }
else { ESP_LOGE(TAG, "%s FAILED: got=%u want=%u", name, got, want); fail++; }
};
// Basic values
check("zero", 0);
check("below_1ms", 999);
check("exactly_1ms", 1000);
check("above_1ms", 1001);
// Shift boundary (1000 = 8 * 125, exercises the >>3 shift)
check("shift_7999", 7999);
check("shift_8000", 8000);
check("shift_8001", 8001);
// 32-bit boundary
check("u32max_minus1", 0xFFFFFFFEULL);
check("u32max", 0xFFFFFFFFULL);
check("u32max_plus1", 0x100000000ULL);
// Realistic uptimes
check("30_days", 2592000000000ULL);
check("1_year", 31536000000000ULL);
// Carry path: construct x = us>>3 with specific hi/lo that trigger adj overflow
{ uint64_t x = (603ULL << 32) | 0xFFFFFFFFU; check("carry_603", x << 3); }
{ uint64_t x = (5000ULL << 32) | 0xFFFFFFFFU; check("carry_5000", x << 3); }
// Carry boundary: exact transition where adj overflows (hi=1000, R=46)
{
uint32_t hi = 1000;
uint32_t thr = 0xFFFFFFFFU - hi * 46U;
uint64_t h = (uint64_t)hi << 32;
check("carry_before", (h | (thr - 1)) << 3);
check("carry_at", (h | thr) << 3);
check("carry_after", (h | (thr + 1)) << 3);
}
// Mod-8 variations (exercises the >>3 truncation)
for (int i = 0; i < 8; i++) { check("mod8", 2592000000000ULL + i); }
if (fail == 0) { ESP_LOGI(TAG, "ALL_PASSED %d tests", pass); }
else { ESP_LOGE(TAG, "%d FAILED out of %d", fail, pass + fail); }
host:
api:
logger:
@@ -0,0 +1,27 @@
esphome:
name: online-image-bmp
host:
http_request:
display:
online_image:
- url: http://127.0.0.1:HTTP_PORT/foo.bmp
id: myimg
format: BMP
type: RGB
on_download_finished:
logger.log:
format: "download finished. cache hit: %u"
args: [cached]
api:
actions:
- action: fetch_image
then:
- component.update: myimg
logger:
level: DEBUG
@@ -1,4 +1,5 @@
esphome:
debug_scheduler: true # Enable scheduler leak detection
name: scheduler-bulk-cleanup
external_components:
@@ -1,4 +1,5 @@
esphome:
debug_scheduler: true # Enable scheduler leak detection
name: scheduler-defer-cancel
host:
@@ -1,4 +1,5 @@
esphome:
debug_scheduler: true # Enable scheduler leak detection
name: scheduler-defer-cancel-regular
host:
@@ -1,4 +1,5 @@
esphome:
debug_scheduler: true # Enable scheduler leak detection
name: scheduler-defer-fifo-simple
host:
@@ -1,4 +1,5 @@
esphome:
debug_scheduler: true # Enable scheduler leak detection
name: scheduler-defer-stress-test
external_components:
@@ -1,4 +1,5 @@
esphome:
debug_scheduler: true # Enable scheduler leak detection
name: scheduler-heap-stress-test
external_components:
@@ -1,4 +1,5 @@
esphome:
debug_scheduler: true # Enable scheduler leak detection
name: scheduler-internal-id-test
on_boot:
priority: -100
@@ -1,4 +1,5 @@
esphome:
debug_scheduler: true # Enable scheduler leak detection
name: scheduler-null-name
host:
@@ -1,4 +1,5 @@
esphome:
debug_scheduler: true # Enable scheduler leak detection
name: scheduler-numeric-id-test
on_boot:
priority: -100
@@ -1,4 +1,5 @@
esphome:
debug_scheduler: true # Enable scheduler leak detection
name: sched-rapid-cancel-test
external_components:
@@ -1,4 +1,5 @@
esphome:
debug_scheduler: true # Enable scheduler leak detection
name: sched-recursive-timeout
external_components:
@@ -1,4 +1,5 @@
esphome:
debug_scheduler: true # Enable scheduler leak detection
name: scheduler-removed-item-race
host:
@@ -1,4 +1,5 @@
esphome:
debug_scheduler: true # Enable scheduler leak detection
name: scheduler-retry-test
on_boot:
priority: -100
@@ -1,4 +1,5 @@
esphome:
debug_scheduler: true # Enable scheduler leak detection
name: sched-simul-callbacks-test
external_components:
@@ -1,4 +1,5 @@
esphome:
debug_scheduler: true # Enable scheduler leak detection
name: scheduler-string-lifetime-test
external_components:
@@ -1,4 +1,5 @@
esphome:
debug_scheduler: true # Enable scheduler leak detection
name: sched-string-name-stress
external_components:
@@ -0,0 +1,23 @@
esphome:
name: host-template-text-save-test
host:
api:
batch_delay: 0ms
logger:
preferences:
flash_write_interval: 0s
text:
- platform: template
name: "Test Text Restore"
id: test_text_restore
optimistic: true
min_length: 0
max_length: 10
mode: text
initial_value: "hello"
restore_value: true
@@ -0,0 +1,153 @@
esphome:
name: uart-mock-ld2410-test
host:
api:
logger:
level: VERBOSE
external_components:
- source:
type: local
path: EXTERNAL_COMPONENT_PATH
# Dummy uart entry to satisfy ld2410's DEPENDENCIES = ["uart"]
# The actual UART bus used is the uart_mock component below
uart:
baud_rate: 115200
port: /dev/null
uart_mock:
id: mock_uart
baud_rate: 256000
auto_start: false
injections:
# Phase 1 (t=100ms): Valid LD2410 normal mode data frame - happy path
# The buffer is clean at this point, so this frame should parse correctly.
# Moving target: 100cm, energy 50
# Still target: 120cm, energy 25
# Detection distance: 300cm
# Target state: 0x03 (moving + still)
#
# Note: LD2410's two_byte_to_int() uses signed char, so low bytes must be
# <=127 to produce correct values. Values >127 wrap negative.
#
# Frame layout (24 bytes):
# [0-3] F4 F3 F2 F1 = data frame header
# [4-5] 0D 00 = length 13
# [6] 02 = data type (normal)
# [7] AA = data header marker
# [8] 03 = target states (moving+still)
# [9-10] 64 00 = moving distance 100 (0x0064)
# [11] 32 = moving energy 50
# [12-13] 78 00 = still distance 120 (0x0078)
# [14] 19 = still energy 25
# [15-16] 2C 01 = detection distance 300 (0x012C)
# [17] 00 = padding
# [18] 55 = data footer marker
# [19] 00 = CRC/check
# [20-23] F8 F7 F6 F5 = data frame footer
- delay: 100ms
inject_rx:
[
0xF4, 0xF3, 0xF2, 0xF1,
0x0D, 0x00,
0x02, 0xAA,
0x03,
0x64, 0x00,
0x32,
0x78, 0x00,
0x19,
0x2C, 0x01,
0x00,
0x55, 0x00,
0xF8, 0xF7, 0xF6, 0xF5,
]
# Phase 2 (t=200ms): Garbage bytes - corrupt the buffer
# These random bytes will accumulate in the LD2410's internal buffer.
# No footer will be found, so the buffer just grows.
- delay: 100ms
inject_rx: [0xDE, 0xAD, 0xBE, 0xEF, 0x00, 0x11, 0x22]
# Phase 3 (t=300ms): Truncated frame (header + partial data, no footer)
# More bytes accumulating in the buffer without a footer match.
- delay: 100ms
inject_rx: [0xF4, 0xF3, 0xF2, 0xF1, 0x0D, 0x00, 0x02, 0xAA]
# Phase 4 (t=500ms): Overflow - inject 85 bytes of 0xFF (MAX_LINE_LENGTH=50)
# Buffer has 15 bytes from phases 2+3.
# Overflow math: need 35 bytes to trigger first overflow (pos 15->49),
# then 50 more to trigger second overflow (pos 0->49). Total = 85 bytes.
# After two overflows, buffer_pos_ = 0 with a completely clean buffer.
# The LD2410 logs "Max command length exceeded" at each overflow.
- delay: 200ms
inject_rx:
[
0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF,
0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF,
0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF,
0xFF, 0xFF, 0xFF, 0xFF, 0xFF,
0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF,
0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF,
0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF,
0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF,
0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF,
]
# Phase 5 (t=600ms): Valid frame after overflow - recovery test
# Buffer was reset by overflow. This valid frame should parse correctly.
# Moving target: 50cm, energy 100
# Still target: 75cm, energy 80
# Detection distance: 127cm
- delay: 100ms
inject_rx:
[
0xF4, 0xF3, 0xF2, 0xF1,
0x0D, 0x00,
0x02, 0xAA,
0x03,
0x32, 0x00,
0x64,
0x4B, 0x00,
0x50,
0x7F, 0x00,
0x00,
0x55, 0x00,
0xF8, 0xF7, 0xF6, 0xF5,
]
ld2410:
id: ld2410_dev
uart_id: mock_uart
sensor:
- platform: ld2410
ld2410_id: ld2410_dev
moving_distance:
name: "Moving Distance"
still_distance:
name: "Still Distance"
moving_energy:
name: "Moving Energy"
still_energy:
name: "Still Energy"
detection_distance:
name: "Detection Distance"
binary_sensor:
- platform: ld2410
ld2410_id: ld2410_dev
has_target:
name: "Has Target"
has_moving_target:
name: "Has Moving Target"
has_still_target:
name: "Has Still Target"
button:
- platform: template
name: "Start Scenario"
id: start_scenario_btn
on_press:
- lambda: 'id(mock_uart).start_scenario();'
@@ -0,0 +1,164 @@
esphome:
name: uart-mock-ld2410-eng-test
host:
api:
logger:
level: VERBOSE
external_components:
- source:
type: local
path: EXTERNAL_COMPONENT_PATH
# Dummy uart entry to satisfy ld2410's DEPENDENCIES = ["uart"]
uart:
baud_rate: 115200
port: /dev/null
uart_mock:
id: mock_uart
baud_rate: 256000
auto_start: false
injections:
# Phase 1 (t=100ms): Valid LD2410 engineering mode data frame
# Captured from a real Screek Human Presence Sensor 1U with LD2410 firmware 2.4.x
#
# Engineering mode frame layout (45 bytes):
# [0-3] F4 F3 F2 F1 = data frame header
# [4-5] 23 00 = length 35
# [6] 01 = data type (engineering mode)
# [7] AA = data header marker
# [8] 03 = target states (moving+still)
# [9-10] 1E 00 = moving distance 30 (0x001E)
# [11] 64 = moving energy 100
# [12-13] 1E 00 = still distance 30 (0x001E)
# [14] 64 = still energy 100
# [15-16] 00 00 = detection distance 0
# [17] 08 = max moving distance gate
# [18] 08 = max still distance gate
# [19-27] gate moving energies (gates 0-8)
# [28-36] gate still energies (gates 0-8)
# [37] 57 = light sensor value 87
# [38] 01 = out pin presence (HIGH)
# [39] 55 = data footer marker
# [40] 00 = check
# [41-44] F8 F7 F6 F5 = data frame footer
- delay: 100ms
inject_rx:
[
0xF4, 0xF3, 0xF2, 0xF1,
0x23, 0x00,
0x01, 0xAA,
0x03,
0x1E, 0x00,
0x64,
0x1E, 0x00,
0x64,
0x00, 0x00,
0x08, 0x08,
0x64, 0x41, 0x06, 0x0E, 0x2B, 0x16, 0x03, 0x03, 0x07,
0x00, 0x00, 0x64, 0x64, 0x64, 0x64, 0x64, 0x64, 0x64,
0x57, 0x01,
0x55, 0x00,
0xF8, 0xF7, 0xF6, 0xF5,
]
# Phase 2 (t=200ms): Second engineering mode frame with different values
# Real capture: moving at 73cm, still at 30cm, detection at 33cm
- delay: 100ms
inject_rx:
[
0xF4, 0xF3, 0xF2, 0xF1,
0x23, 0x00,
0x01, 0xAA,
0x03,
0x49, 0x00,
0x64,
0x1E, 0x00,
0x64,
0x21, 0x00,
0x08, 0x08,
0x11, 0x64, 0x05, 0x29, 0x39, 0x10, 0x03, 0x11, 0x0E,
0x00, 0x00, 0x64, 0x64, 0x64, 0x64, 0x64, 0x64, 0x64,
0x57, 0x01,
0x55, 0x00,
0xF8, 0xF7, 0xF6, 0xF5,
]
# Phase 3 (t=300ms): Frame with still target at 291cm (multi-byte distance)
# This tests the two_byte_to_int function with high byte > 0
# Note: low byte 0x2F < 0x80 so it avoids the signed char bug
- delay: 100ms
inject_rx:
[
0xF4, 0xF3, 0xF2, 0xF1,
0x23, 0x00,
0x01, 0xAA,
0x03,
0x2F, 0x00,
0x36,
0x23, 0x01,
0x64,
0x21, 0x00,
0x08, 0x08,
0x2F, 0x36, 0x09, 0x0D, 0x15, 0x0B, 0x06, 0x06, 0x08,
0x00, 0x00, 0x64, 0x64, 0x64, 0x64, 0x64, 0x5A, 0x3D,
0x57, 0x01,
0x55, 0x00,
0xF8, 0xF7, 0xF6, 0xF5,
]
ld2410:
id: ld2410_dev
uart_id: mock_uart
sensor:
- platform: ld2410
ld2410_id: ld2410_dev
moving_distance:
name: "Moving Distance"
still_distance:
name: "Still Distance"
moving_energy:
name: "Moving Energy"
still_energy:
name: "Still Energy"
detection_distance:
name: "Detection Distance"
light:
name: "Light"
g0:
move_energy:
name: "Gate 0 Move Energy"
still_energy:
name: "Gate 0 Still Energy"
g1:
move_energy:
name: "Gate 1 Move Energy"
still_energy:
name: "Gate 1 Still Energy"
g2:
move_energy:
name: "Gate 2 Move Energy"
still_energy:
name: "Gate 2 Still Energy"
binary_sensor:
- platform: ld2410
ld2410_id: ld2410_dev
has_target:
name: "Has Target"
has_moving_target:
name: "Has Moving Target"
has_still_target:
name: "Has Still Target"
out_pin_presence_status:
name: "Out Pin Presence"
button:
- platform: template
name: "Start Scenario"
id: start_scenario_btn
on_press:
- lambda: 'id(mock_uart).start_scenario();'
@@ -0,0 +1,179 @@
esphome:
name: uart-mock-ld2412-test
host:
api:
logger:
level: VERBOSE
external_components:
- source:
type: local
path: EXTERNAL_COMPONENT_PATH
# Dummy uart entry to satisfy ld2412's DEPENDENCIES = ["uart"]
# The actual UART bus used is the uart_mock component below
uart:
baud_rate: 115200
port: /dev/null
uart_mock:
id: mock_uart
baud_rate: 256000
auto_start: false
injections:
# Phase 1 (t=100ms): Valid LD2412 normal mode data frame - happy path
# The buffer is clean at this point, so this frame should parse correctly.
# Moving target: 100cm, energy 50
# Still target: 120cm, energy 25
# Target state: 0x03 (moving + still)
# detection_distance = 100 (LD2412 computes from moving target when MOVE_BITMASK set)
#
# Frame layout (24 bytes):
# [0-3] F4 F3 F2 F1 = data frame header
# [4-5] 0D 00 = length 13
# [6] 02 = data type (normal)
# [7] AA = data header marker
# [8] 03 = target states (moving+still)
# [9-10] 64 00 = moving distance 100 (0x0064)
# [11] 32 = moving energy 50
# [12-13] 78 00 = still distance 120 (0x0078)
# [14] 19 = still energy 25
# [15-16] 64 00 = detect distance bytes (ignored by LD2412 code)
# [17] 00 = padding
# [18] 55 = data footer marker
# [19] 00 = CRC/check
# [20-23] F8 F7 F6 F5 = data frame footer
- delay: 100ms
inject_rx:
[
0xF4, 0xF3, 0xF2, 0xF1,
0x0D, 0x00,
0x02, 0xAA,
0x03,
0x64, 0x00,
0x32,
0x78, 0x00,
0x19,
0x64, 0x00,
0x00,
0x55, 0x00,
0xF8, 0xF7, 0xF6, 0xF5,
]
# Phase 2 (t=300ms): Garbage bytes
# LD2412's parser rejects bytes that don't match the frame header at
# position 0 (must start with F4 or FD), so buffer stays empty.
- delay: 200ms
inject_rx: [0xDE, 0xAD, 0xBE, 0xEF, 0x00, 0x11, 0x22]
# Phase 3 (t=400ms): Truncated frame (header + partial data, no footer)
# Starts with valid data frame header so parser accepts it.
# After this, buffer_pos_ = 8.
- delay: 100ms
inject_rx: [0xF4, 0xF3, 0xF2, 0xF1, 0x0D, 0x00, 0x02, 0xAA]
# Phase 4 (t=600ms): Overflow - inject 60 bytes of 0xFF (MAX_LINE_LENGTH=54)
# Buffer has 8 bytes from phase 3 (garbage in phase 2 was rejected).
# Overflow math: buffer_pos_ starts at 8, overflow triggers when
# buffer_pos_ reaches 53 (MAX_LINE_LENGTH - 1). Need 45 more bytes to
# fill positions 8-52, then byte 46 triggers overflow. After overflow,
# buffer_pos_ = 0 and remaining 0xFF bytes are rejected (don't match header).
- delay: 200ms
inject_rx:
[
0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF,
0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF,
0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF,
0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF,
0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF,
0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF,
]
# Phase 5 (t=700ms): Valid frame after overflow - recovery test
# Buffer was reset by overflow. This valid frame should parse correctly.
# Moving target: 50cm, energy 100
# Still target: 75cm, energy 80
# detection_distance = 50 (moving target distance, since MOVE_BITMASK set)
- delay: 100ms
inject_rx:
[
0xF4, 0xF3, 0xF2, 0xF1,
0x0D, 0x00,
0x02, 0xAA,
0x03,
0x32, 0x00,
0x64,
0x4B, 0x00,
0x50,
0x32, 0x00,
0x00,
0x55, 0x00,
0xF8, 0xF7, 0xF6, 0xF5,
]
ld2412:
id: ld2412_dev
uart_id: mock_uart
sensor:
- platform: ld2412
ld2412_id: ld2412_dev
moving_distance:
name: "Moving Distance"
filters:
- timeout:
timeout: 50ms
value: last
- throttle_with_priority: 50ms
still_distance:
name: "Still Distance"
filters:
- timeout:
timeout: 50ms
value: last
- throttle_with_priority: 50ms
moving_energy:
name: "Moving Energy"
filters:
- timeout:
timeout: 50ms
value: last
- throttle_with_priority: 50ms
still_energy:
name: "Still Energy"
filters:
- timeout:
timeout: 50ms
value: last
- throttle_with_priority: 50ms
detection_distance:
name: "Detection Distance"
filters:
- timeout:
timeout: 50ms
value: last
- throttle_with_priority: 50ms
binary_sensor:
- platform: ld2412
ld2412_id: ld2412_dev
has_target:
name: "Has Target"
filters:
- settle: 50ms
has_moving_target:
name: "Has Moving Target"
filters:
- settle: 50ms
has_still_target:
name: "Has Still Target"
filters:
- settle: 50ms
button:
- platform: template
name: "Start Scenario"
id: start_scenario_btn
on_press:
- lambda: 'id(mock_uart).start_scenario();'
@@ -0,0 +1,182 @@
esphome:
name: uart-mock-ld2412-eng-test
host:
api:
logger:
level: VERBOSE
external_components:
- source:
type: local
path: EXTERNAL_COMPONENT_PATH
# Dummy uart entry to satisfy ld2412's DEPENDENCIES = ["uart"]
uart:
baud_rate: 115200
port: /dev/null
uart_mock:
id: mock_uart
baud_rate: 256000
auto_start: false
injections:
# Phase 1 (t=100ms): Valid LD2412 engineering mode data frame
#
# Engineering mode frame layout (52 bytes):
# [0-3] F4 F3 F2 F1 = data frame header
# [4-5] 2A 00 = length 42
# [6] 01 = data type (engineering mode)
# [7] AA = data header marker
# [8] 03 = target states (moving+still)
# [9-10] 1E 00 = moving distance 30 (0x001E)
# [11] 64 = moving energy 100
# [12-13] 1E 00 = still distance 30 (0x001E)
# [14] 64 = still energy 100
# [15-16] 00 00 = detection distance bytes (ignored)
# [17-30] gate moving energies (14 gates)
# [31-44] gate still energies (14 gates)
# [45] 57 = light sensor value 87
# [46] 55 = data footer marker
# [47] 00 = check
# [48-51] F8 F7 F6 F5 = data frame footer
- delay: 100ms
inject_rx:
[
0xF4, 0xF3, 0xF2, 0xF1,
0x2A, 0x00,
0x01, 0xAA,
0x03,
0x1E, 0x00,
0x64,
0x1E, 0x00,
0x64,
0x00, 0x00,
0x64, 0x41, 0x06, 0x0E, 0x2B, 0x16, 0x03, 0x03, 0x07, 0x05, 0x09, 0x08, 0x07, 0x06,
0x00, 0x00, 0x64, 0x64, 0x64, 0x64, 0x64, 0x64, 0x64, 0x50, 0x40, 0x30, 0x20, 0x10,
0x57,
0x55, 0x00,
0xF8, 0xF7, 0xF6, 0xF5,
]
# Phase 2 (t=200ms): Second engineering mode frame with different values
# Moving at 73cm, still at 30cm
- delay: 100ms
inject_rx:
[
0xF4, 0xF3, 0xF2, 0xF1,
0x2A, 0x00,
0x01, 0xAA,
0x03,
0x49, 0x00,
0x64,
0x1E, 0x00,
0x64,
0x21, 0x00,
0x11, 0x64, 0x05, 0x29, 0x39, 0x10, 0x03, 0x11, 0x0E, 0x08, 0x06, 0x04, 0x03, 0x02,
0x00, 0x00, 0x64, 0x64, 0x64, 0x64, 0x64, 0x64, 0x64, 0x50, 0x40, 0x30, 0x20, 0x10,
0x57,
0x55, 0x00,
0xF8, 0xF7, 0xF6, 0xF5,
]
# Phase 3 (t=300ms): Frame with still target at 291cm (multi-byte distance)
# This tests encode_uint16 with high byte > 0
# Target state: 0x02 (still only) -> detection_distance = still distance = 291
- delay: 100ms
inject_rx:
[
0xF4, 0xF3, 0xF2, 0xF1,
0x2A, 0x00,
0x01, 0xAA,
0x02,
0x2F, 0x00,
0x36,
0x23, 0x01,
0x64,
0x21, 0x00,
0x2F, 0x36, 0x09, 0x0D, 0x15, 0x0B, 0x06, 0x06, 0x08, 0x09, 0x08, 0x07, 0x06, 0x05,
0x00, 0x00, 0x64, 0x64, 0x64, 0x64, 0x64, 0x5A, 0x3D, 0x30, 0x20, 0x10, 0x08, 0x04,
0x57,
0x55, 0x00,
0xF8, 0xF7, 0xF6, 0xF5,
]
# Common filter definitions
.sensor_filters: &sensor_filters
filters:
- timeout:
timeout: 50ms
value: last
- throttle_with_priority: 50ms
.binary_filters: &binary_filters
filters:
- settle: 50ms
ld2412:
id: ld2412_dev
uart_id: mock_uart
sensor:
- platform: ld2412
ld2412_id: ld2412_dev
moving_distance:
name: "Moving Distance"
<<: *sensor_filters
still_distance:
name: "Still Distance"
<<: *sensor_filters
moving_energy:
name: "Moving Energy"
<<: *sensor_filters
still_energy:
name: "Still Energy"
<<: *sensor_filters
detection_distance:
name: "Detection Distance"
<<: *sensor_filters
light:
name: "Light"
<<: *sensor_filters
gate_0:
move_energy:
name: "Gate 0 Move Energy"
<<: *sensor_filters
still_energy:
name: "Gate 0 Still Energy"
<<: *sensor_filters
gate_1:
move_energy:
name: "Gate 1 Move Energy"
<<: *sensor_filters
still_energy:
name: "Gate 1 Still Energy"
<<: *sensor_filters
gate_2:
move_energy:
name: "Gate 2 Move Energy"
<<: *sensor_filters
still_energy:
name: "Gate 2 Still Energy"
<<: *sensor_filters
binary_sensor:
- platform: ld2412
ld2412_id: ld2412_dev
has_target:
name: "Has Target"
<<: *binary_filters
has_moving_target:
name: "Has Moving Target"
<<: *binary_filters
has_still_target:
name: "Has Still Target"
<<: *binary_filters
button:
- platform: template
name: "Start Scenario"
id: start_scenario_btn
on_press:
- lambda: 'id(mock_uart).start_scenario();'
@@ -0,0 +1,167 @@
esphome:
name: uart-mock-ld2412-eng-trunc
host:
api:
logger:
level: VERBOSE
external_components:
- source:
type: local
path: EXTERNAL_COMPONENT_PATH
# Dummy uart entry to satisfy ld2412's DEPENDENCIES = ["uart"]
uart:
baud_rate: 115200
port: /dev/null
uart_mock:
id: mock_uart
baud_rate: 256000
auto_start: false
injections:
# Phase 1 (t=100ms): Valid engineering mode frame (52 bytes, buffer_pos_=52)
# Establishes baseline: gate_0_move=100, light=87
- delay: 100ms
inject_rx:
[
0xF4, 0xF3, 0xF2, 0xF1,
0x2A, 0x00,
0x01, 0xAA,
0x03,
0x1E, 0x00,
0x64,
0x1E, 0x00,
0x64,
0x00, 0x00,
0x64, 0x41, 0x06, 0x0E, 0x2B, 0x16, 0x03, 0x03, 0x07, 0x05, 0x09, 0x08, 0x07, 0x06,
0x00, 0x00, 0x64, 0x64, 0x64, 0x64, 0x64, 0x64, 0x64, 0x50, 0x40, 0x30, 0x20, 0x10,
0x57,
0x55, 0x00,
0xF8, 0xF7, 0xF6, 0xF5,
]
# Phase 2 (t=200ms): Truncated engineering mode frame (24 bytes, buffer_pos_=24)
# This frame has data_type=0x01 (engineering) but only enough data for the
# basic target fields, not the gate energies or light sensor.
# buffer_pos_=24 passes the old check (>= 12) but fails the new check (< 46).
# Without the fix, indices 17-45 would read stale buffer data from Phase 1.
#
# Layout (24 bytes):
# [0-3] F4 F3 F2 F1 = data frame header
# [4-5] 0E 00 = length 14
# [6] 01 = data type (engineering mode)
# [7] AA = data header marker
# [8] 03 = target states (moving+still)
# [9-10] 1E 00 = moving distance 30
# [11] 50 = moving energy 80
# [12-13] 1E 00 = still distance 30
# [14] 50 = still energy 80
# [15-16] FF FF = garbage detection distance bytes
# [17] FF = padding (would be gate data in full frame)
# [18] 55 = data footer marker (at buffer_pos_ - 6)
# [19] 00 = check byte
# [20-23] F8 F7 F6 F5 = data frame footer
- delay: 100ms
inject_rx:
[
0xF4, 0xF3, 0xF2, 0xF1,
0x0E, 0x00,
0x01, 0xAA,
0x03,
0x1E, 0x00,
0x50,
0x1E, 0x00,
0x50,
0xFF, 0xFF,
0xFF,
0x55, 0x00,
0xF8, 0xF7, 0xF6, 0xF5,
]
# Phase 3 (t=300ms): Valid recovery frame with different values
# gate_0_move=50, light=42 — proves component recovered
- delay: 100ms
inject_rx:
[
0xF4, 0xF3, 0xF2, 0xF1,
0x2A, 0x00,
0x01, 0xAA,
0x03,
0x1E, 0x00,
0x64,
0x1E, 0x00,
0x64,
0x00, 0x00,
0x32, 0x20, 0x06, 0x0E, 0x2B, 0x16, 0x03, 0x03, 0x07, 0x05, 0x09, 0x08, 0x07, 0x06,
0x00, 0x00, 0x32, 0x32, 0x32, 0x32, 0x32, 0x32, 0x32, 0x28, 0x20, 0x18, 0x10, 0x08,
0x2A,
0x55, 0x00,
0xF8, 0xF7, 0xF6, 0xF5,
]
# Common filter definitions
.sensor_filters: &sensor_filters
filters:
- timeout:
timeout: 50ms
value: last
- throttle_with_priority: 50ms
.binary_filters: &binary_filters
filters:
- settle: 50ms
ld2412:
id: ld2412_dev
uart_id: mock_uart
sensor:
- platform: ld2412
ld2412_id: ld2412_dev
moving_distance:
name: "Moving Distance"
<<: *sensor_filters
still_distance:
name: "Still Distance"
<<: *sensor_filters
moving_energy:
name: "Moving Energy"
<<: *sensor_filters
still_energy:
name: "Still Energy"
<<: *sensor_filters
detection_distance:
name: "Detection Distance"
<<: *sensor_filters
light:
name: "Light"
<<: *sensor_filters
gate_0:
move_energy:
name: "Gate 0 Move Energy"
<<: *sensor_filters
still_energy:
name: "Gate 0 Still Energy"
<<: *sensor_filters
binary_sensor:
- platform: ld2412
ld2412_id: ld2412_dev
has_target:
name: "Has Target"
<<: *binary_filters
has_moving_target:
name: "Has Moving Target"
<<: *binary_filters
has_still_target:
name: "Has Still Target"
<<: *binary_filters
button:
- platform: template
name: "Start Scenario"
id: start_scenario_btn
on_press:
- lambda: 'id(mock_uart).start_scenario();'

Some files were not shown because too many files have changed in this diff Show More