github/esphome
2
0
Fork 0
mirror of https://github.com/esphome/esphome.git synced 2026-09-26 22:40:21 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'upstream/dev' into rp2040-upload-improvements

Browse Source
This commit is contained in:
J. Nick Koston
2026-03-09 17:45:00 -10:00
parent 8359705fab e82f0f4432
commit 399d695a39
354 changed files with 8457 additions and 3199 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/component_tests/binary_sensor/test_binary_sensor.py
+6 -4
View File
@@ -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):
tests/component_tests/button/test_button.py
+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
tests/component_tests/helpers.py
+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))
tests/component_tests/sensor/test_sensor.py
+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_entity_strings(" 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
tests/component_tests/text/test_text.py
+7 -5
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):
tests/component_tests/text_sensor/test_text_sensor.py
+14 -9
View File
@@ -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_entity_strings(" in main_cpp
assert "ts_3->set_entity_strings(" 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
tests/components/README.md
+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 ...
```
tests/components/captive_portal/test.rp2040-ard.yaml
+1
View File
@@ -0,0 +1 @@
<<: !include common.yaml
tests/components/esp8266/test.esp8266-ard.yaml
+3
View File
@@ -1,3 +1,6 @@
esp8266:
enable_full_printf: false
logger:
level: VERBOSE
tests/components/http_request/test.esp8266-ard.yaml
+5 -3
View File
@@ -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
tests/components/ld2450/common.h
+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));
};
tests/components/nextion/common_tft_upload.yaml
+5
View File
@@ -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
tests/components/nextion/common_tft_upload_watchdog.yaml
+3
View File
@@ -0,0 +1,3 @@
display:
- id: !extend main_lcd
tft_upload_watchdog_timeout: 30s
tests/components/nextion/test.esp32-ard.yaml
+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
tests/components/nextion/test.esp32-idf.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
tests/components/nextion/test.esp8266-ard.yaml
+1 -4
View File
@@ -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
tests/components/ota/test.nrf52-mcumgr.yaml
+27
View File
@@ -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);
tests/components/packet_transport/binary_sensor/binary_sensor_test.cpp
+77
View File
@@ -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
tests/components/packet_transport/cpp_test.yaml
-11
View File
@@ -1,11 +0,0 @@
# Extra component configuration required by C++ unit tests.
# Loaded by cpp_unit_test.py and merged into the test build config
# before validation, so that platform defines (USE_SENSOR, etc.) are generated.
sensor:
- platform: template
id: test_cpp_sensor
binary_sensor:
- platform: template
id: test_cpp_binary_sensor
tests/components/packet_transport/packet_transport_test.cpp
-259
View File
@@ -65,198 +65,6 @@ TEST(PacketTransportTest, SetProviderEncryption) {
EXPECT_EQ(transport.providers_["host1"].encryption_key, key);
}
// --- Sensor management (requires USE_SENSOR / USE_BINARY_SENSOR) ---
#ifdef USE_SENSOR
TEST(PacketTransportTest, 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(PacketTransportTest, 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);
}
#endif
#ifdef USE_BINARY_SENSOR
TEST(PacketTransportTest, 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(PacketTransportTest, 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);
}
#endif
// --- Unencrypted round-trip tests (require USE_SENSOR / USE_BINARY_SENSOR) ---
#ifdef USE_SENSOR
TEST(PacketTransportTest, 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);
}
#endif
#ifdef USE_BINARY_SENSOR
TEST(PacketTransportTest, 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);
}
#endif
#if defined(USE_SENSOR) && defined(USE_BINARY_SENSOR)
TEST(PacketTransportTest, 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);
}
#endif
// --- Encrypted round-trip ---
#ifdef USE_SENSOR
TEST(PacketTransportTest, 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);
}
// --- Selective send ---
TEST(PacketTransportTest, 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
}
#endif
// --- Ping key tests ---
TEST(PacketTransportTest, PingKeyStoredWhenEncrypted) {
@@ -319,73 +127,6 @@ TEST(PacketTransportTest, PingKeyMaxLimit) {
EXPECT_FALSE(transport.ping_keys_.contains("host4"));
}
#ifdef USE_SENSOR
TEST(PacketTransportTest, 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(PacketTransportTest, 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
}
#endif
// --- Process error handling ---
TEST(PacketTransportTest, ProcessShortBuffer) {
tests/components/packet_transport/sensor/sensor_test.cpp
+170
View File
@@ -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
tests/components/rp2040/test.rp2040-ard.yaml
+3
View File
@@ -1,3 +1,6 @@
rp2040:
enable_full_printf: false
logger:
level: VERBOSE
tests/components/serial_proxy/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
tests/components/serial_proxy/test.esp32-idf.yaml
+8
View File
@@ -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
tests/components/serial_proxy/test.esp8266-ard.yaml
+8
View File
@@ -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
tests/components/serial_proxy/test.rp2040-ard.yaml
+8
View File
@@ -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
tests/components/uart/common.h
+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));
};
tests/components/uart/test.esp32-c3-idf.yaml
+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] "
tests/components/uart/test.esp32-idf.yaml
+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#"
tests/components/uart/test.esp8266-ard.yaml
+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#"
tests/components/uart/test.host.yaml
+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
tests/components/uart/test.rp2040-ard.yaml
+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
baud_rate: 9600
@@ -13,3 +17,9 @@ 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] "
tests/components/usb_uart/common.yaml
+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] "
tests/integration/fixtures/external_components/uart_mock/__init__.py
+6 -2
View File
@@ -62,6 +62,7 @@ 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,
)
@@ -87,7 +88,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),
@@ -126,6 +127,8 @@ async def inject_rx_to_code(config, action_id, template_arg, args):
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
@@ -135,7 +138,8 @@ async def to_code(config):
cg.add(var.set_baud_rate(config[CONF_BAUD_RATE]))
cg.add(var.set_rx_buffer_size(config[CONF_RX_BUFFER_SIZE]))
cg.add(var.set_rx_full_threshold(config[CONF_RX_FULL_THRESHOLD]))
if CONF_RX_FULL_THRESHOLD in config:
cg.add(var.set_rx_full_threshold(config[CONF_RX_FULL_THRESHOLD]))
cg.add(var.set_rx_timeout(config[CONF_RX_TIMEOUT]))
cg.add(var.set_stop_bits(config[CONF_STOP_BITS]))
cg.add(var.set_data_bits(config[CONF_DATA_BITS]))
tests/integration/fixtures/external_components/uart_mock/automation.h
+14 -2
View File
@@ -22,18 +22,30 @@ template<typename... Ts> class MockUartInjectRXAction : public Action<Ts...>, pu
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
this->parent_->inject_to_rx_buffer(this->code_.data, static_cast<size_t>(this->len_));
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...);
this->parent_->inject_to_rx_buffer(val);
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)
tests/integration/fixtures/external_components/uart_mock/uart_mock.cpp
+26 -5
View File
@@ -53,6 +53,15 @@ void MockUartComponent::loop() {
}
}
// 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) {
@@ -104,12 +113,12 @@ void MockUartComponent::write_array(const uint8_t *data, size_t len) {
}
#endif
if (this->scenario_active_) {
this->try_match_response_();
}
// 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_ && this->scenario_active_) {
if (this->tx_hook_) {
std::vector<uint8_t> buf(data, data + len);
this->tx_hook_(buf);
}
@@ -144,8 +153,9 @@ bool MockUartComponent::read_array(uint8_t *data, size_t len) {
size_t MockUartComponent::available() { return this->rx_buffer_.size(); }
void MockUartComponent::flush() {
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) {
@@ -208,4 +218,15 @@ void MockUartComponent::inject_to_rx_buffer(const std::vector<uint8_t> &data) {
}
}
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
tests/integration/fixtures/external_components/uart_mock/uart_mock.h
+11 -1
View File
@@ -28,7 +28,7 @@ class MockUartComponent : public uart::UARTComponent, public Component {
bool peek_byte(uint8_t *data) override;
bool read_array(uint8_t *data, size_t len) override;
size_t available() override;
void flush() 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;
@@ -43,6 +43,8 @@ class MockUartComponent : public uart::UARTComponent, public Component {
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 {}
@@ -82,6 +84,14 @@ class MockUartComponent : public uart::UARTComponent, public Component {
};
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};
tests/integration/fixtures/uart_mock_ld2412_engineering.yaml
+27 -66
View File
@@ -102,6 +102,18 @@ uart_mock:
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
@@ -111,107 +123,56 @@ sensor:
ld2412_id: ld2412_dev
moving_distance:
name: "Moving Distance"
filters:
- timeout:
timeout: 50ms
value: last
- throttle_with_priority: 50ms
<<: *sensor_filters
still_distance:
name: "Still Distance"
filters:
- timeout:
timeout: 50ms
value: last
- throttle_with_priority: 50ms
<<: *sensor_filters
moving_energy:
name: "Moving Energy"
filters:
- timeout:
timeout: 50ms
value: last
- throttle_with_priority: 50ms
<<: *sensor_filters
still_energy:
name: "Still Energy"
filters:
- timeout:
timeout: 50ms
value: last
- throttle_with_priority: 50ms
<<: *sensor_filters
detection_distance:
name: "Detection Distance"
filters:
- timeout:
timeout: 50ms
value: last
- throttle_with_priority: 50ms
<<: *sensor_filters
light:
name: "Light"
filters:
- timeout:
timeout: 50ms
value: last
- throttle_with_priority: 50ms
<<: *sensor_filters
gate_0:
move_energy:
name: "Gate 0 Move Energy"
filters:
- timeout:
timeout: 50ms
value: last
- throttle_with_priority: 50ms
<<: *sensor_filters
still_energy:
name: "Gate 0 Still Energy"
filters:
- timeout:
timeout: 50ms
value: last
- throttle_with_priority: 50ms
<<: *sensor_filters
gate_1:
move_energy:
name: "Gate 1 Move Energy"
filters:
- timeout:
timeout: 50ms
value: last
- throttle_with_priority: 50ms
<<: *sensor_filters
still_energy:
name: "Gate 1 Still Energy"
filters:
- timeout:
timeout: 50ms
value: last
- throttle_with_priority: 50ms
<<: *sensor_filters
gate_2:
move_energy:
name: "Gate 2 Move Energy"
filters:
- timeout:
timeout: 50ms
value: last
- throttle_with_priority: 50ms
<<: *sensor_filters
still_energy:
name: "Gate 2 Still Energy"
filters:
- timeout:
timeout: 50ms
value: last
- throttle_with_priority: 50ms
<<: *sensor_filters
binary_sensor:
- platform: ld2412
ld2412_id: ld2412_dev
has_target:
name: "Has Target"
filters:
- settle: 50ms
<<: *binary_filters
has_moving_target:
name: "Has Moving Target"
filters:
- settle: 50ms
<<: *binary_filters
has_still_target:
name: "Has Still Target"
filters:
- settle: 50ms
<<: *binary_filters
button:
- platform: template
tests/integration/fixtures/uart_mock_ld2412_engineering_truncated.yaml
+167
View File
@@ -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();'
tests/integration/fixtures/uart_mock_ld2420.yaml
+187
View File
@@ -0,0 +1,187 @@
esphome:
name: uart-mock-ld2420-test
host:
api:
logger:
level: VERBOSE
external_components:
- source:
type: local
path: EXTERNAL_COMPONENT_PATH
# Dummy uart entry to satisfy ld2420'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: 115200
auto_start: false
responses:
# Version-specific response: match the complete firmware version command TX.
# CMD_READ_VERSION (0x0000) TX = FD FC FB FA 02 00 00 00 04 03 02 01
# Returns "v2.0.0" → get_firmware_int = 200 >= 154 → energy mode
#
# Response layout:
# [0-3] FD FC FB FA = header
# [4-5] 0C 00 = length 12
# [6] 00 = cmd (CMD_READ_VERSION)
# [7] 01 = status (ACK)
# [8-9] 00 00 = error = 0
# [10] 06 = ver_len = 6
# [11] 00 = padding
# [12-17] "v2.0.0" = version string
# [18-21] 04 03 02 01 = footer
- expect_tx:
[0xFD, 0xFC, 0xFB, 0xFA, 0x02, 0x00, 0x00, 0x00, 0x04, 0x03, 0x02, 0x01]
inject_rx:
[
0xFD, 0xFC, 0xFB, 0xFA,
0x0C, 0x00,
0x00, 0x01,
0x00, 0x00,
0x06, 0x00,
0x76, 0x32, 0x2E, 0x30, 0x2E, 0x30,
0x04, 0x03, 0x02, 0x01,
]
# Catch-all response: match any command footer (04 03 02 01).
# Returns a generic ACK with cmd=0xFF (CMD_ENABLE_CONF case in switch).
# All commands get unblocked via cmd_reply_.ack = true.
# Data fields stay zeroed (min_gate=0, max_gate=0, timeout=0, thresholds=0).
#
# Response layout:
# [0-3] FD FC FB FA = header
# [4-5] 04 00 = length 4
# [6] FF = cmd (handled as CMD_ENABLE_CONF)
# [7] 01 = status (ACK)
# [8-9] 00 00 = error = 0
# [10-13] 04 03 02 01 = footer
- expect_tx: [0x04, 0x03, 0x02, 0x01]
inject_rx:
[
0xFD, 0xFC, 0xFB, 0xFA,
0x04, 0x00,
0xFF, 0x01,
0x00, 0x00,
0x04, 0x03, 0x02, 0x01,
]
injections:
# Phase 1 (t=100ms): Valid LD2420 energy mode data frame - happy path
# Buffer is clean (buffer_pos_=0). This frame should parse correctly.
# Presence: 1 (target detected), Distance: 100cm, Gate energies: all 0
#
# Energy frame layout (45 bytes):
# [0-3] F4 F3 F2 F1 = energy frame header
# [4-5] 23 00 = length 35 (1+2+32)
# [6] 01 = presence (1 = target)
# [7-8] 64 00 = distance 100 (uint16_t LE)
# [9-40] 00 00 x16 = 16 gate energies (uint16_t LE each)
# [41-44] F8 F7 F6 F5 = energy frame footer
- delay: 100ms
inject_rx:
[
0xF4, 0xF3, 0xF2, 0xF1,
0x23, 0x00,
0x01,
0x64, 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,
0xF8, 0xF7, 0xF6, 0xF5,
]
# Phase 2 (t=300ms): Garbage bytes
# LD2420's readline_ does NOT check frame headers at position 0 (unlike LD2412),
# so these bytes accumulate in the buffer. buffer_pos_ goes from 0 to 7.
- delay: 200ms
inject_rx: [0xDE, 0xAD, 0xBE, 0xEF, 0x00, 0x11, 0x22]
# Phase 3 (t=400ms): Truncated energy frame WITH footer (13 bytes)
# This tests PR #14458 bug #3: missing length validation in handle_energy_mode_.
# The 7 garbage bytes from Phase 2 are still in the buffer (buffer_pos_=7).
# These 13 bytes are appended at positions 7-19 (buffer_pos_=20).
# Energy footer at positions 16-19 triggers handle_energy_mode_(buffer, 20).
#
# Pre-fix: handle_energy_mode_ reads 32 bytes of gate energy from buffer[9:40],
# which is past the 20 actual bytes. Reads uninitialized data.
# No "Energy frame too short" warning exists.
# Post-fix: len=20 < 41 → logs "Energy frame too short: 20 bytes", returns early.
#
# Frame: header + length + presence + distance + footer (no gate data)
- delay: 100ms
inject_rx:
[
0xF4, 0xF3, 0xF2, 0xF1,
0x23, 0x00,
0x01,
0x64, 0x00,
0xF8, 0xF7, 0xF6, 0xF5,
]
# Phase 4 (t=600ms): Overflow - inject 50 bytes of 0xFF (MAX_LINE_LENGTH=50)
# After Phase 3, buffer_pos_=0 (reset after energy footer detection).
# 49 bytes fill positions 0-48 (buffer_pos_=49), 50th byte triggers overflow.
# Logs "Max command length exceeded; ignoring", buffer_pos_=0.
- 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,
]
# Phase 5 (t=1500ms): Valid frame after overflow - recovery test
# Buffer was reset by overflow. This valid frame should parse correctly.
# Presence: 1 (target), Distance: 50cm
# Delay=900ms ensures >1000ms gap from Phase 1 for REFRESH_RATE_MS throttle.
- delay: 900ms
inject_rx:
[
0xF4, 0xF3, 0xF2, 0xF1,
0x23, 0x00,
0x01,
0x32, 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,
0xF8, 0xF7, 0xF6, 0xF5,
]
button:
- platform: template
name: "Start Scenario"
on_press:
- lambda: 'id(mock_uart).start_scenario();'
ld2420:
id: ld2420_dev
uart_id: mock_uart
sensor:
- platform: ld2420
ld2420_id: ld2420_dev
moving_distance:
name: "Moving Distance"
filters:
- timeout:
timeout: 50ms
value: last
- throttle_with_priority: 50ms
binary_sensor:
- platform: ld2420
ld2420_id: ld2420_dev
has_target:
name: "Has Target"
filters:
- settle: 50ms
tests/integration/fixtures/uart_mock_ld2420_simple.yaml
+141
View File
@@ -0,0 +1,141 @@
esphome:
name: uart-mock-ld2420-simple-test
host:
api:
logger:
level: VERBOSE
external_components:
- source:
type: local
path: EXTERNAL_COMPONENT_PATH
# Dummy uart entry to satisfy ld2420's DEPENDENCIES = ["uart"]
uart:
baud_rate: 115200
port: /dev/null
uart_mock:
id: mock_uart
baud_rate: 115200
auto_start: false
responses:
# Catch-all response only (no version-specific response).
# Without a version response, firmware_ver_ stays at default "v0.0.0".
# get_firmware_int("v0.0.0") = 0 < 154 → simple mode (CMD_SYSTEM_MODE_SIMPLE).
- expect_tx: [0x04, 0x03, 0x02, 0x01]
inject_rx:
[
0xFD, 0xFC, 0xFB, 0xFA,
0x04, 0x00,
0xFF, 0x01,
0x00, 0x00,
0x04, 0x03, 0x02, 0x01,
]
injections:
# Phase 1 (t=100ms): Valid simple mode text frame - happy path
# "ON Range 0100\r\n" → presence=true, distance=100
# Simple mode frames end with \r\n (0x0D 0x0A), triggering handle_simple_mode_.
- delay: 100ms
inject_rx:
[
0x4F, 0x4E, 0x20, 0x52, 0x61, 0x6E, 0x67, 0x65, 0x20,
0x30, 0x31, 0x30, 0x30,
0x0D, 0x0A,
]
# Phase 2 (t=300ms): Garbage bytes
# LD2420's readline_ stores all bytes regardless of header. buffer_pos_ = 7.
- delay: 200ms
inject_rx: [0xDE, 0xAD, 0xBE, 0xEF, 0x00, 0x11, 0x22]
# Phase 3 (t=500ms): Overflow - inject 50 bytes of 0xFF (MAX_LINE_LENGTH=50)
# buffer_pos_ starts at 7 (from Phase 2 garbage).
# Positions 7-48 fill (42 bytes), byte 43 triggers overflow (buffer_pos_=49).
# After overflow: buffer_pos_=0, remaining 7 bytes fill positions 0-6.
# Final buffer_pos_ = 7.
- 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,
]
# Phase 4 (t=1400ms): Recovery after overflow
# buffer_pos_ = 7 (from overflow remainder). These 15 bytes fill positions 7-21.
# At position 21 (0x0A), \r\n detected → handle_simple_mode_(buffer, 22).
# Parser skips 0xFF bytes at positions 0-6, finds "ON" at positions 7-8,
# parses digits "0050" → distance=50.
# Delay=900ms ensures >1000ms gap from Phase 1 for REFRESH_RATE_MS throttle.
- delay: 900ms
inject_rx:
[
0x4F, 0x4E, 0x20, 0x52, 0x61, 0x6E, 0x67, 0x65, 0x20,
0x30, 0x30, 0x35, 0x30,
0x0D, 0x0A,
]
# Phase 5 (t=2500ms): 16-digit distance - tests PR #14458 bug #1
# "ON Range 0000000000000000\r\n" has 16 digit characters.
# handle_simple_mode_ outbuf is 16 bytes, can hold 15 digits (index 0-14).
#
# Pre-fix: At the 16th digit, index=15, (index < bufsize-1) is false.
# The digit branch doesn't increment pos. The else branch is skipped.
# pos stays at the 16th digit FOREVER → INFINITE LOOP.
# The binary hangs, no more state updates, test times out.
# Post-fix: pos always increments (moved outside digit branch).
# 16th digit skipped, loop continues to \r\n. distance=0.
- delay: 1100ms
inject_rx:
[
0x4F, 0x4E, 0x20, 0x52, 0x61, 0x6E, 0x67, 0x65, 0x20,
0x30, 0x30, 0x30, 0x30, 0x30, 0x30, 0x30, 0x30,
0x30, 0x30, 0x30, 0x30, 0x30, 0x30, 0x30, 0x30,
0x0D, 0x0A,
]
# Phase 6 (t=3700ms): Post-bug-trigger recovery
# If Phase 5 didn't hang, this frame should parse correctly.
# "ON Range 0025\r\n" → distance=25
# Delay=1200ms ensures >1000ms gap from Phase 5 for throttle.
- delay: 1200ms
inject_rx:
[
0x4F, 0x4E, 0x20, 0x52, 0x61, 0x6E, 0x67, 0x65, 0x20,
0x30, 0x30, 0x32, 0x35,
0x0D, 0x0A,
]
button:
- platform: template
name: "Start Scenario"
on_press:
- lambda: 'id(mock_uart).start_scenario();'
ld2420:
id: ld2420_dev
uart_id: mock_uart
sensor:
- platform: ld2420
ld2420_id: ld2420_dev
moving_distance:
name: "Moving Distance"
filters:
- timeout:
timeout: 50ms
value: last
- throttle_with_priority: 50ms
binary_sensor:
- platform: ld2420
ld2420_id: ld2420_dev
has_target:
name: "Has Target"
filters:
- settle: 50ms
tests/integration/fixtures/uart_mock_ld2450.yaml
+221
View File
@@ -0,0 +1,221 @@
esphome:
name: uart-mock-ld2450-test
host:
api:
logger:
level: VERBOSE
external_components:
- source:
type: local
path: EXTERNAL_COMPONENT_PATH
# Dummy uart entry to satisfy ld2450'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
responses:
# Catch-all response: match any command footer (04 03 02 01).
# Returns a generic ACK to unblock setup commands.
#
# Response layout:
# [0-3] FD FC FB FA = header
# [4-5] 04 00 = length 4
# [6] FF = cmd (handled as CMD_ENABLE_CONF)
# [7] 01 = status (ACK)
# [8-9] 00 00 = error = 0
# [10-13] 04 03 02 01 = footer
- expect_tx: [0x04, 0x03, 0x02, 0x01]
inject_rx:
[
0xFD, 0xFC, 0xFB, 0xFA,
0x04, 0x00,
0xFF, 0x01,
0x00, 0x00,
0x04, 0x03, 0x02, 0x01,
]
injections:
# Phase 1 (t=100ms): Valid LD2450 periodic data frame - happy path
# The buffer is clean at this point, so this frame should parse correctly.
#
# Target 1: X=-500mm, Y=1000mm, Speed=-50mm/s (approaching), Res=320mm
# X: magnitude=500 (0x01F4), negative → high=0x01, low=0xF4
# Y: magnitude=1000 (0x03E8), positive → high=0x83, low=0xE8
# Speed: raw=5, negative (approaching) → high=0x00, low=0x05, decoded=-50mm/s
# Resolution: 320 → low=0x40, high=0x01
# Distance: sqrt(500²+1000²) = sqrt(1250000) ≈ 1118mm
#
# Target 2: X=200mm, Y=500mm, Speed=0 (stationary), Res=100mm
# X: magnitude=200 (0x00C8), positive → high=0x80, low=0xC8
# Y: magnitude=500 (0x01F4), positive → high=0x81, low=0xF4
# Speed: 0 → 0x00, 0x00
# Resolution: 100 → low=0x64, high=0x00
# Distance: sqrt(200²+500²) = sqrt(290000) ≈ 538mm
#
# Target 3: No target (all zeros)
# Distance: 0 → sensors publish unknown/NaN
#
# Counts: target_count=2, moving_target_count=1, still_target_count=1
#
# Frame layout (30 bytes):
# [0-3] AA FF 03 00 = periodic data header
# [4-11] Target 1 (8 bytes): X_L X_H Y_L Y_H SPD_L SPD_H RES_L RES_H
# [12-19] Target 2 (8 bytes)
# [20-27] Target 3 (8 bytes)
# [28-29] 55 CC = periodic data footer
- delay: 100ms
inject_rx:
[
0xAA, 0xFF, 0x03, 0x00,
0xF4, 0x01, 0xE8, 0x83, 0x05, 0x00, 0x40, 0x01,
0xC8, 0x80, 0xF4, 0x81, 0x00, 0x00, 0x64, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x55, 0xCC,
]
# Phase 2 (t=300ms): Garbage bytes
# LD2450's readline_ does NOT reject bytes at position 0 (unlike LD2412),
# so these bytes accumulate in the buffer. buffer_pos_ goes from 0 to 7.
- delay: 200ms
inject_rx: [0xDE, 0xAD, 0xBE, 0xEF, 0x00, 0x11, 0x22]
# Phase 3 (t=400ms): Truncated frame (header + partial data, no footer)
# More bytes accumulating in the buffer without a footer match.
# After this, buffer_pos_ = 7 + 8 = 15.
- delay: 100ms
inject_rx: [0xAA, 0xFF, 0x03, 0x00, 0x01, 0x02, 0x03, 0x04]
# Phase 4 (t=600ms): Overflow - inject 75 bytes of 0xFF (MAX_LINE_LENGTH=45)
# Buffer has 15 bytes from phases 2+3.
# readline_() stores bytes while buffer_pos_ < 44. When buffer_pos_ == 44,
# the next byte triggers overflow: logs warning, resets buffer_pos_ to 0,
# and discards that byte.
#
# First overflow: 29 bytes fill positions 15-43 (buffer_pos_=44), byte 30
# triggers overflow (discarded). Total consumed: 30 bytes.
# Second overflow: 44 bytes fill positions 0-43 (buffer_pos_=44), byte 45
# triggers overflow (discarded). Total consumed: 30+45 = 75 bytes.
# After both overflows, buffer_pos_ = 0 (clean state for recovery frame).
- 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,
]
# Phase 5 (t=700ms): Valid frame after overflow - recovery test
# Buffer was reset by overflow. This valid frame should parse correctly.
#
# Target 1: X=300mm, Y=400mm, Speed=30mm/s (moving away), Res=100mm
# X: magnitude=300 (0x012C), positive → high=0x81, low=0x2C
# Y: magnitude=400 (0x0190), positive → high=0x81, low=0x90
# Speed: raw=3, positive (moving away) → high=0x80, low=0x03, decoded=30mm/s
# Resolution: 100 → low=0x64, high=0x00
# Distance: sqrt(300²+400²) = 500mm
#
# Target 2 & 3: No target (all zeros)
# Counts: target_count=1, moving_target_count=1, still_target_count=0
- delay: 100ms
inject_rx:
[
0xAA, 0xFF, 0x03, 0x00,
0x2C, 0x81, 0x90, 0x81, 0x03, 0x80, 0x64, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x55, 0xCC,
]
ld2450:
id: ld2450_dev
uart_id: mock_uart
sensor:
- platform: ld2450
ld2450_id: ld2450_dev
target_count:
name: "Target Count"
filters: &sensor_filters
- timeout:
timeout: 50ms
value: last
- throttle_with_priority: 50ms
still_target_count:
name: "Still Target Count"
filters: *sensor_filters
moving_target_count:
name: "Moving Target Count"
filters: *sensor_filters
target_1:
x:
name: "Target 1 X"
filters: *sensor_filters
y:
name: "Target 1 Y"
filters: *sensor_filters
speed:
name: "Target 1 Speed"
filters: *sensor_filters
distance:
name: "Target 1 Distance"
filters: *sensor_filters
resolution:
name: "Target 1 Resolution"
filters: *sensor_filters
angle:
name: "Target 1 Angle"
filters: *sensor_filters
target_2:
x:
name: "Target 2 X"
filters: *sensor_filters
y:
name: "Target 2 Y"
filters: *sensor_filters
speed:
name: "Target 2 Speed"
filters: *sensor_filters
distance:
name: "Target 2 Distance"
filters: *sensor_filters
binary_sensor:
- platform: ld2450
ld2450_id: ld2450_dev
has_target:
name: "Has Target"
filters: &binary_sensor_filters
- settle: 50ms
has_moving_target:
name: "Has Moving Target"
filters: *binary_sensor_filters
has_still_target:
name: "Has Still Target"
filters: *binary_sensor_filters
text_sensor:
- platform: ld2450
ld2450_id: ld2450_dev
target_1:
direction:
name: "Target 1 Direction"
button:
- platform: template
name: "Start Scenario"
id: start_scenario_btn
on_press:
- lambda: 'id(mock_uart).start_scenario();'
tests/integration/fixtures/uart_mock_modbus_no_threshold.yaml
+64
View File
@@ -0,0 +1,64 @@
esphome:
name: uart-mock-modbus-no-thresh
host:
api:
logger:
level: VERBOSE
external_components:
- source:
type: local
path: EXTERNAL_COMPONENT_PATH
# Dummy uart entry to satisfy modbus's DEPENDENCIES = ["uart"]
# The actual UART bus used is the uart_mock component below
uart:
baud_rate: 115200
port: /dev/null
# Simulate a non-hardware UART (e.g., USB UART) by not setting rx_full_threshold.
# This leaves it at the default sentinel value (0), triggering the 50ms fallback timeout.
uart_mock:
- id: virtual_uart_dev
baud_rate: 9600
auto_start: false
debug:
on_tx:
- then:
- if:
condition: #Read 80 input registers on device 2, starting at address 0 (SDM meter request)
lambda: "return data == std::vector<uint8_t>({0x02,0x04,0x00,0x00,0x00,0x50,0xF0,0x05});"
then:
- uart_mock.inject_rx: # First USB packet: SDM meter response part 1
!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};
- uart_mock.inject_rx: # Second USB packet: rest of response (staged with 40ms latency)
delay: 40ms
data: !lambda return{0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,
0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x42,0x6F,0xCC,0xCD,0x43,0x7C,0xB8,0x10,0x3D,0x38,0x51,0xEC,
0x43,0x81,0x1B,0xE7,0x3B,0x03,0x12,0x6F,0x50,0x1B};
modbus:
uart_id: virtual_uart_dev
turnaround_time: 10ms
sensor:
- platform: sdm_meter
address: 2
update_interval: 1s
phase_a:
voltage:
name: sdm_voltage
button:
- platform: template
name: "Start Scenario"
id: start_scenario_btn
on_press:
- lambda: 'id(virtual_uart_dev).start_scenario();'
tests/integration/state_utils.py
+23 -5
View File
@@ -13,6 +13,7 @@ from aioesphomeapi import (
EntityInfo,
EntityState,
SensorState,
TextSensorState,
)
_LOGGER = logging.getLogger(__name__)
@@ -244,12 +245,13 @@ class InitialStateHelper:
class SensorStateCollector:
"""Collects sensor and binary sensor state updates and provides wait helpers.
"""Collects sensor, binary sensor, and text sensor state updates with wait helpers.
Usage:
collector = SensorStateCollector(
sensor_names=["moving_distance", "still_distance"],
binary_sensor_names=["has_target"],
text_sensor_names=["direction"],
)
# Use collector.on_state as the callback (or wrap it)
client.subscribe_states(helper.on_state_wrapper(collector.on_state))
@@ -259,18 +261,23 @@ class SensorStateCollector:
# Access collected states
assert collector.sensor_states["moving_distance"][0] == approx(100.0)
assert collector.text_sensor_states["direction"][0] == "Approaching"
"""
def __init__(
self,
sensor_names: list[str],
binary_sensor_names: list[str] | None = None,
text_sensor_names: list[str] | None = None,
entities: list[EntityInfo] | None = None,
) -> None:
self.sensor_states: dict[str, list[float]] = {name: [] for name in sensor_names}
self.binary_states: dict[str, list[bool]] = {
name: [] for name in (binary_sensor_names or [])
}
self.text_sensor_states: dict[str, list[str]] = {
name: [] for name in (text_sensor_names or [])
}
self._key_to_sensor: dict[int, str] = {}
self._waiters: list[tuple[Callable[[], bool], asyncio.Future[bool]]] = []
@@ -279,7 +286,11 @@ class SensorStateCollector:
def build_key_mapping(self, entities: list[EntityInfo]) -> None:
"""Build key-to-name mapping from entities. Sorted by descending length."""
all_names = list(self.sensor_states.keys()) + list(self.binary_states.keys())
all_names = (
list(self.sensor_states.keys())
+ list(self.binary_states.keys())
+ list(self.text_sensor_states.keys())
)
all_names.sort(key=len, reverse=True)
self._key_to_sensor = build_key_to_entity_mapping(entities, all_names)
@@ -295,6 +306,11 @@ class SensorStateCollector:
if sensor_name and sensor_name in self.binary_states:
self.binary_states[sensor_name].append(state.state)
self._check_waiters()
elif isinstance(state, TextSensorState) and not state.missing_state:
sensor_name = self._key_to_sensor.get(state.key)
if sensor_name and sensor_name in self.text_sensor_states:
self.text_sensor_states[sensor_name].append(state.state)
self._check_waiters()
def _check_waiters(self) -> None:
"""Check all pending waiters and resolve any whose condition is met."""
@@ -303,9 +319,11 @@ class SensorStateCollector:
future.set_result(True)
def _all_have_values(self) -> bool:
"""Check if all sensor and binary sensor lists have at least one value."""
return all(len(v) >= 1 for v in self.sensor_states.values()) and all(
len(v) >= 1 for v in self.binary_states.values()
"""Check if all sensor, binary sensor, and text sensor lists have at least one value."""
return (
all(len(v) >= 1 for v in self.sensor_states.values())
and all(len(v) >= 1 for v in self.binary_states.values())
and all(len(v) >= 1 for v in self.text_sensor_states.values())
)
async def wait_for_all(self, timeout: float = 3.0) -> None:
tests/integration/test_uart_mock_ld2412.py
+125
View File
@@ -14,6 +14,12 @@ test_uart_mock_ld2412_engineering (engineering mode):
2. Multi-byte still distance (291cm) using high byte > 0
3. Gate energy sensor values
4. Detection distance computed from target state
test_uart_mock_ld2412_engineering_truncated (truncated engineering mode):
1. Valid engineering frame establishes baseline sensor values
2. Truncated engineering frame (24 bytes) is rejected — gate/light sensors
must not receive garbage from stale buffer data or frame footer bytes
3. Recovery frame with different values proves the component survived
"""
from __future__ import annotations
@@ -273,3 +279,122 @@ async def test_uart_mock_ld2412_engineering(
)
assert pytest.approx(291.0) in collector.sensor_states["detection_distance"]
@pytest.mark.asyncio
async def test_uart_mock_ld2412_engineering_truncated(
yaml_config: str,
run_compiled: RunCompiledFunction,
api_client_connected: APIClientConnectedFactory,
) -> None:
"""Test that truncated engineering mode frames don't corrupt sensor values.
Without the fix, a 24-byte engineering mode frame passes the old buffer_pos_ >= 12
check but reads indices 17-45 from stale buffer data, publishing garbage values
(e.g. frame footer bytes 0xF8=248 as gate energy).
"""
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 the truncated frame warning
truncated_warning_seen = loop.create_future()
def line_callback(line: str) -> None:
if (
"Engineering mode packet too short" in line
and not truncated_warning_seen.done()
):
truncated_warning_seen.set_result(True)
collector = SensorStateCollector(
sensor_names=[
"moving_distance",
"still_distance",
"moving_energy",
"still_energy",
"detection_distance",
"light",
"gate_0_move_energy",
"gate_0_still_energy",
],
binary_sensor_names=[
"has_target",
"has_moving_target",
"has_still_target",
],
)
# Signal when we see Phase 3 recovery values (gate_0_move=50)
recovery_received = collector.add_waiter(
lambda: pytest.approx(50.0) in collector.sensor_states["gate_0_move_energy"]
)
async with (
run_compiled(yaml_config, line_callback=line_callback),
api_client_connected() as client,
):
entities, _ = await client.list_entities_services()
collector.build_key_mapping(entities)
initial_state_helper = InitialStateHelper(entities)
client.subscribe_states(
initial_state_helper.on_state_wrapper(collector.on_state)
)
try:
await initial_state_helper.wait_for_initial_states()
except TimeoutError:
pytest.fail("Timeout waiting for initial states")
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 Phase 1 — valid engineering frame establishes baseline
try:
await collector.wait_for_all(timeout=3.0)
except TimeoutError:
pytest.fail(
f"Timeout waiting for Phase 1 frame. Received:\n"
f" sensor_states: {collector.sensor_states}\n"
f" binary_states: {collector.binary_states}"
)
# Phase 1 baseline: gate_0_move=100, light=87
assert collector.sensor_states["gate_0_move_energy"][0] == pytest.approx(100.0)
assert collector.sensor_states["light"][0] == pytest.approx(87.0)
# Wait for Phase 3 recovery frame (gate_0_move=50)
try:
await asyncio.wait_for(recovery_received, timeout=3.0)
except TimeoutError:
pytest.fail(
f"Timeout waiting for recovery frame. Received:\n"
f" gate_0_move_energy: {collector.sensor_states['gate_0_move_energy']}\n"
f" light: {collector.sensor_states['light']}"
)
# Verify the truncated frame warning was logged
assert truncated_warning_seen.done(), (
"Expected 'Engineering mode packet too short' warning in logs"
)
# Phase 3 recovery: gate_0_move=50, light=42
assert pytest.approx(50.0) in collector.sensor_states["gate_0_move_energy"]
assert pytest.approx(42.0) in collector.sensor_states["light"]
# The critical assertion: gate_0_move_energy must never have received
# garbage values from the truncated frame. Without the fix,
# buffer_data_[17] = 0xFF = 255 would be published as gate_0_move.
for value in collector.sensor_states["gate_0_move_energy"]:
assert value == pytest.approx(100.0) or value == pytest.approx(50.0), (
f"gate_0_move_energy got unexpected value {value} — "
f"truncated frame likely leaked stale buffer data. "
f"All values: {collector.sensor_states['gate_0_move_energy']}"
)
tests/integration/test_uart_mock_ld2420.py
+273
View File
@@ -0,0 +1,273 @@
"""Integration test for LD2420 component with mock UART.
Tests:
test_uart_mock_ld2420 (energy mode):
1. Happy path - valid energy frame publishes correct sensor values
2. Garbage resilience - random bytes don't crash the component
3. Truncated energy frame - triggers "Energy frame too short" warning (PR #14458 bug #3)
4. Buffer overflow recovery - overflow resets the parser
5. Post-overflow parsing - next valid frame after overflow is parsed correctly
6. TX logging - verifies LD2420 sends expected setup commands
test_uart_mock_ld2420_simple (simple mode):
1. Happy path - valid simple mode text frame publishes correct values
2. Garbage resilience
3. Buffer overflow recovery
4. 16-digit distance triggers infinite loop pre-fix (PR #14458 bug #1)
5. Post-bug-trigger recovery proves the parser survived
"""
from __future__ import annotations
import asyncio
from pathlib import Path
from aioesphomeapi import ButtonInfo
import pytest
from .state_utils import InitialStateHelper, SensorStateCollector, find_entity
from .types import APIClientConnectedFactory, RunCompiledFunction
@pytest.mark.asyncio
async def test_uart_mock_ld2420(
yaml_config: str,
run_compiled: RunCompiledFunction,
api_client_connected: APIClientConnectedFactory,
) -> None:
"""Test LD2420 energy mode: happy path, truncated frame, overflow, and recovery."""
# 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 overflow warning in logs
overflow_seen = loop.create_future()
# Track "Energy frame too short" warning (PR #14458 bug #3 fix)
# This message ONLY exists after the fix. Pre-fix, handle_energy_mode_
# silently reads past the buffer without any warning.
truncated_frame_warning_seen = loop.create_future()
# Track TX data logged by the mock for assertions
tx_log_lines: list[str] = []
def line_callback(line: str) -> None:
if "Max command length exceeded" in line and not overflow_seen.done():
overflow_seen.set_result(True)
if "Energy frame too short" in line and not truncated_frame_warning_seen.done():
truncated_frame_warning_seen.set_result(True)
# Capture all TX log lines from uart_mock
if "uart_mock" in line and "TX " in line:
tx_log_lines.append(line)
collector = SensorStateCollector(
sensor_names=["moving_distance"],
binary_sensor_names=["has_target"],
)
# Signal when we see recovery frame values
recovery_received = collector.add_waiter(
lambda: pytest.approx(50.0) in collector.sensor_states["moving_distance"]
)
async with (
run_compiled(yaml_config, line_callback=line_callback),
api_client_connected() as client,
):
entities, _ = await client.list_entities_services()
collector.build_key_mapping(entities)
# Set up initial state helper
initial_state_helper = InitialStateHelper(entities)
client.subscribe_states(
initial_state_helper.on_state_wrapper(collector.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 Phase 1 - all sensors and binary sensors have at least one value
try:
await collector.wait_for_all(timeout=3.0)
except TimeoutError:
pytest.fail(
f"Timeout waiting for Phase 1 frame. Received:\n"
f" sensor_states: {collector.sensor_states}\n"
f" binary_states: {collector.binary_states}"
)
# Phase 1 values: moving=100, has_target=true
assert collector.sensor_states["moving_distance"][0] == pytest.approx(100.0)
assert collector.binary_states["has_target"][0] is True
# Wait for the recovery frame (Phase 5) to be parsed
# This proves the component survived garbage + truncated + overflow
try:
await asyncio.wait_for(recovery_received, timeout=5.0)
except TimeoutError:
pytest.fail(
f"Timeout waiting for recovery frame. Received:\n"
f" sensor_states: {collector.sensor_states}"
)
# Verify overflow warning was logged
assert overflow_seen.done(), (
"Expected 'Max command length exceeded' warning in logs"
)
# Verify truncated frame warning was logged (PR #14458 bug #3)
# This assertion FAILS before PR #14458 because the length check
# and warning message did not exist.
assert truncated_frame_warning_seen.done(), (
"Expected 'Energy frame too short' warning in logs. "
"This indicates PR #14458 fix for handle_energy_mode_ length "
"validation is missing."
)
# Verify LD2420 sent setup commands (TX logging)
assert len(tx_log_lines) > 0, "Expected TX log lines from uart_mock"
tx_data = " ".join(tx_log_lines)
# Verify command frame header appears (FD:FC:FB:FA)
assert "FD.FC.FB.FA" in tx_data or "FD:FC:FB:FA" in tx_data, (
"Expected LD2420 command frame header FD:FC:FB:FA in TX log"
)
# Verify command frame footer appears (04:03:02:01)
assert "04.03.02.01" in tx_data or "04:03:02:01" in tx_data, (
"Expected LD2420 command frame footer 04:03:02:01 in TX log"
)
# Recovery frame values (Phase 5, after overflow)
recovery_values = [
v
for v in collector.sensor_states["moving_distance"]
if v == pytest.approx(50.0)
]
assert len(recovery_values) >= 1, (
f"Expected moving_distance=50 in recovery, got: {collector.sensor_states['moving_distance']}"
)
@pytest.mark.asyncio
async def test_uart_mock_ld2420_simple(
yaml_config: str,
run_compiled: RunCompiledFunction,
api_client_connected: APIClientConnectedFactory,
) -> None:
"""Test LD2420 simple mode: happy path, overflow, and 16-digit bug trigger."""
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 overflow warning in logs
overflow_seen = loop.create_future()
def line_callback(line: str) -> None:
if "Max command length exceeded" in line and not overflow_seen.done():
overflow_seen.set_result(True)
collector = SensorStateCollector(
sensor_names=["moving_distance"],
binary_sensor_names=["has_target"],
)
# Signal for recovery frames
recovery_received = collector.add_waiter(
lambda: pytest.approx(50.0) in collector.sensor_states["moving_distance"]
)
post_bug_received = collector.add_waiter(
lambda: pytest.approx(25.0) in collector.sensor_states["moving_distance"]
)
async with (
run_compiled(yaml_config, line_callback=line_callback),
api_client_connected() as client,
):
entities, _ = await client.list_entities_services()
collector.build_key_mapping(entities)
initial_state_helper = InitialStateHelper(entities)
client.subscribe_states(
initial_state_helper.on_state_wrapper(collector.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 Phase 1 - all sensors and binary sensors have at least one value
try:
await collector.wait_for_all(timeout=3.0)
except TimeoutError:
pytest.fail(
f"Timeout waiting for Phase 1 frame. Received:\n"
f" sensor_states: {collector.sensor_states}\n"
f" binary_states: {collector.binary_states}"
)
# Phase 1: simple mode "ON Range 0100\r\n" → distance=100, presence=true
assert collector.sensor_states["moving_distance"][0] == pytest.approx(100.0)
assert collector.binary_states["has_target"][0] is True
# Wait for Phase 4 recovery (distance=50) after overflow
try:
await asyncio.wait_for(recovery_received, timeout=5.0)
except TimeoutError:
pytest.fail(
f"Timeout waiting for recovery frame. Received:\n"
f" moving_distance: {collector.sensor_states['moving_distance']}"
)
# Verify overflow warning was logged
assert overflow_seen.done(), (
"Expected 'Max command length exceeded' warning in logs"
)
# Wait for Phase 6: distance=25 (post-16-digit-bug recovery)
# This assertion FAILS before PR #14458 because the 16-digit frame
# in Phase 5 causes an infinite loop in handle_simple_mode_ pre-fix.
# The binary hangs, Phase 6 never fires, and this wait times out.
try:
await asyncio.wait_for(post_bug_received, timeout=8.0)
except TimeoutError:
pytest.fail(
f"Timeout waiting for post-bug recovery (distance=25). "
f"This likely means Phase 5 (16-digit frame) caused an infinite "
f"loop in handle_simple_mode_, indicating PR #14458 bug #1 fix "
f"is missing.\n"
f" moving_distance values: {collector.sensor_states['moving_distance']}"
)
# Verify post-bug value
post_bug_values = [
v
for v in collector.sensor_states["moving_distance"]
if v == pytest.approx(25.0)
]
assert len(post_bug_values) >= 1, (
f"Expected moving_distance=25 after 16-digit test, "
f"got: {collector.sensor_states['moving_distance']}"
)
tests/integration/test_uart_mock_ld2450.py
+204
View File
@@ -0,0 +1,204 @@
"""Integration test for LD2450 component with mock UART.
Tests:
test_uart_mock_ld2450:
1. Happy path - valid periodic data frame publishes correct target sensor values
2. Multi-target tracking - verifies target count, moving/still counts
3. Target coordinate decoding - signed X/Y coordinates with sign-magnitude encoding
4. Speed decoding - approaching (negative) and stationary (zero) targets
5. Distance calculation - computed from X/Y via sqrt(x²+y²)
6. Direction text sensor - "Approaching" for negative speed target
7. Garbage resilience - random bytes don't crash the component
8. Truncated frame handling - partial frame doesn't corrupt state
9. Buffer overflow recovery - overflow resets the parser
10. Post-overflow parsing - next valid frame after overflow is parsed correctly
11. TX logging - verifies LD2450 sends expected setup commands
"""
from __future__ import annotations
import asyncio
from pathlib import Path
from aioesphomeapi import ButtonInfo
import pytest
from .state_utils import InitialStateHelper, SensorStateCollector, find_entity
from .types import APIClientConnectedFactory, RunCompiledFunction
@pytest.mark.asyncio
async def test_uart_mock_ld2450(
yaml_config: str,
run_compiled: RunCompiledFunction,
api_client_connected: APIClientConnectedFactory,
) -> None:
"""Test LD2450 data parsing with happy path, garbage, overflow, and recovery."""
# 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 overflow warning in logs
overflow_seen = loop.create_future()
# Track TX data logged by the mock for assertions
tx_log_lines: list[str] = []
def line_callback(line: str) -> None:
if "Max command length exceeded" in line and not overflow_seen.done():
overflow_seen.set_result(True)
# Capture all TX log lines from uart_mock
if "uart_mock" in line and "TX " in line:
tx_log_lines.append(line)
collector = SensorStateCollector(
sensor_names=[
"target_1_x",
"target_1_y",
"target_1_speed",
"target_1_distance",
"target_1_resolution",
"target_1_angle",
"target_2_x",
"target_2_y",
"target_2_speed",
"target_2_distance",
"target_count",
"still_target_count",
"moving_target_count",
],
binary_sensor_names=[
"has_target",
"has_moving_target",
"has_still_target",
],
text_sensor_names=[
"target_1_direction",
],
)
# Signal when we see recovery frame values (target 1 distance ≈ 500mm)
recovery_received = collector.add_waiter(
lambda: (
pytest.approx(500.0, abs=1.0)
in collector.sensor_states["target_1_distance"]
)
)
async with (
run_compiled(yaml_config, line_callback=line_callback),
api_client_connected() as client,
):
entities, _ = await client.list_entities_services()
collector.build_key_mapping(entities)
# Set up initial state helper
initial_state_helper = InitialStateHelper(entities)
client.subscribe_states(
initial_state_helper.on_state_wrapper(collector.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 Phase 1 - all sensors and binary sensors have at least one value
try:
await collector.wait_for_all(timeout=5.0)
except TimeoutError:
pytest.fail(
f"Timeout waiting for Phase 1 frame. Received:\n"
f" sensor_states: {collector.sensor_states}\n"
f" binary_states: {collector.binary_states}\n"
f" text_states: {collector.text_sensor_states}"
)
# Phase 1 values:
# Target 1: X=-500, Y=1000, Speed=-50 (approaching), Res=320
# Distance = sqrt(500²+1000²) ≈ 1118mm
assert collector.sensor_states["target_1_x"][0] == pytest.approx(-500.0)
assert collector.sensor_states["target_1_y"][0] == pytest.approx(1000.0)
assert collector.sensor_states["target_1_speed"][0] == pytest.approx(-50.0)
assert collector.sensor_states["target_1_resolution"][0] == pytest.approx(320.0)
# Distance computed from X/Y
assert collector.sensor_states["target_1_distance"][0] == pytest.approx(
1118.0, abs=1.0
)
# Target 2: X=200, Y=500, Speed=0 (stationary), Res=100
# Distance = sqrt(200²+500²) ≈ 538mm
assert collector.sensor_states["target_2_x"][0] == pytest.approx(200.0)
assert collector.sensor_states["target_2_y"][0] == pytest.approx(500.0)
assert collector.sensor_states["target_2_speed"][0] == pytest.approx(0.0)
assert collector.sensor_states["target_2_distance"][0] == pytest.approx(
538.0, abs=1.0
)
# Target counts: 2 targets total, 1 moving, 1 still
assert collector.sensor_states["target_count"][0] == pytest.approx(2.0)
assert collector.sensor_states["moving_target_count"][0] == pytest.approx(1.0)
assert collector.sensor_states["still_target_count"][0] == pytest.approx(1.0)
# Binary sensors: all true (targets detected)
assert collector.binary_states["has_target"][0] is True
assert collector.binary_states["has_moving_target"][0] is True
assert collector.binary_states["has_still_target"][0] is True
# Direction text sensor: Target 1 is approaching (speed < 0)
assert collector.text_sensor_states["target_1_direction"][0] == "Approaching"
# Wait for the recovery frame (Phase 5) to be parsed
# This proves the component survived garbage + truncated + overflow
try:
await asyncio.wait_for(recovery_received, timeout=5.0)
except TimeoutError:
pytest.fail(
f"Timeout waiting for recovery frame. Received:\n"
f" sensor_states: {collector.sensor_states}"
)
# Verify overflow warning was logged
assert overflow_seen.done(), (
"Expected 'Max command length exceeded' warning in logs"
)
# Verify LD2450 sent setup commands (TX logging)
assert len(tx_log_lines) > 0, "Expected TX log lines from uart_mock"
tx_data = " ".join(tx_log_lines)
# Verify command frame header appears (FD:FC:FB:FA)
assert "FD:FC:FB:FA" in tx_data, (
"Expected LD2450 command frame header FD:FC:FB:FA in TX log"
)
# Verify command frame footer appears (04:03:02:01)
assert "04:03:02:01" in tx_data, (
"Expected LD2450 command frame footer 04:03:02:01 in TX log"
)
# Recovery frame values (Phase 5, after overflow):
# Target 1: X=300, Y=400, Distance=500, Speed=30 (moving away)
# target_count=1, moving=1, still=0
#
# Note: throttle filters cause sensor lists to have different lengths,
# so we check each value appeared somewhere rather than using a shared index.
assert (
pytest.approx(500.0, abs=1.0)
in collector.sensor_states["target_1_distance"]
)
assert pytest.approx(300.0) in collector.sensor_states["target_1_x"]
assert pytest.approx(400.0) in collector.sensor_states["target_1_y"]
assert pytest.approx(30.0) in collector.sensor_states["target_1_speed"]
assert pytest.approx(1.0) in collector.sensor_states["target_count"]
assert pytest.approx(1.0) in collector.sensor_states["moving_target_count"]
assert pytest.approx(0.0) in collector.sensor_states["still_target_count"]
tests/integration/test_uart_mock_modbus.py
+79
View File
@@ -5,6 +5,10 @@ test_uart_mock_modbus :
1. Read a single register and parse successfully (basic_register)
2. Read multiple registers from SDM meter and parse successfully (sdm_voltage), with some intermediate delay to simulate UART buffer time.
test_uart_mock_modbus_no_threshold :
Test modbus with no rx_full_threshold set (simulating USB UART / non-hardware UART).
Verifies the 50ms fallback timeout handles chunked data with USB packet gaps.
"""
from __future__ import annotations
@@ -218,3 +222,78 @@ async def test_uart_mock_modbus_timing(
f"Timeout waiting for SDM voltage change. Received sensor states:\n"
f" sdm_voltage: {sensor_states['sdm_voltage']}\n"
)
@pytest.mark.asyncio
async def test_uart_mock_modbus_no_threshold(
yaml_config: str,
run_compiled: RunCompiledFunction,
api_client_connected: APIClientConnectedFactory,
) -> None:
"""Test modbus with no rx_full_threshold (simulating USB UART).
Without the 50ms fallback timeout, the chunked response with a 40ms 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"
)
tests/script/test_helpers.py
+91
View File
@@ -1027,6 +1027,73 @@ def test_get_all_dependencies_empty_set() -> None:
assert result == set()
def test_get_all_dependencies_platform_component() -> None:
"""Platform components (domain.component) are looked up via get_platform,
not get_component."""
platform_comp = Mock()
platform_comp.dependencies = []
platform_comp.auto_load = []
with (
patch("esphome.loader.get_component") as mock_get_component,
patch("helpers.get_platform") as mock_get_platform,
):
mock_get_platform.return_value = platform_comp
mock_get_component.return_value = None
result = helpers.get_all_dependencies({"sensor.bthome"})
mock_get_platform.assert_called_once_with("sensor", "bthome")
mock_get_component.assert_not_called()
assert result == {"sensor.bthome"}
def test_get_all_dependencies_platform_component_with_dependencies() -> None:
"""Dependencies of a platform component are resolved transitively."""
platform_comp = Mock()
platform_comp.dependencies = ["sensor"]
platform_comp.auto_load = []
sensor_comp = Mock()
sensor_comp.dependencies = []
sensor_comp.auto_load = []
with (
patch("esphome.loader.get_component") as mock_get_component,
patch("helpers.get_platform") as mock_get_platform,
):
mock_get_platform.return_value = platform_comp
mock_get_component.side_effect = lambda name: (
sensor_comp if name == "sensor" else None
)
result = helpers.get_all_dependencies({"sensor.bthome"})
assert result == {"sensor.bthome", "sensor"}
def test_get_all_dependencies_cpp_testing_flag() -> None:
"""cpp_testing=True propagates to CORE.cpp_testing during resolution."""
from esphome.core import CORE
with (
patch("esphome.loader.get_component") as mock_get_component,
patch("esphome.loader.get_platform"),
):
observed: list[bool] = []
def capturing_get_component(name: str):
observed.append(CORE.cpp_testing)
mock_get_component.side_effect = capturing_get_component
helpers.get_all_dependencies({"some_comp"}, cpp_testing=True)
assert observed and all(observed), (
"CORE.cpp_testing should be True during resolution"
)
def test_get_components_from_integration_fixtures() -> None:
"""Test extraction of components from fixture YAML files."""
yaml_content = {
@@ -1057,6 +1124,30 @@ def test_get_components_from_integration_fixtures() -> None:
assert components == expected_components
def test_get_components_from_integration_fixtures_skips_yaml_anchors() -> None:
"""Test that YAML anchor keys (starting with '.') are excluded."""
yaml_content = {
"sensor": [{"platform": "template", "name": "test"}],
"esphome": {"name": "test"},
".sensor_filters": {"filters": [{"timeout": "50ms"}]},
".binary_filters": {"filters": [{"settle": "50ms"}]},
}
mock_yaml_file = Mock()
with (
patch("pathlib.Path.glob") as mock_glob,
patch("esphome.yaml_util.load_yaml", return_value=yaml_content),
):
mock_glob.return_value = [mock_yaml_file]
components = helpers.get_components_from_integration_fixtures()
assert ".sensor_filters" not in components
assert ".binary_filters" not in components
assert components == {"sensor", "esphome", "template"}
@pytest.mark.parametrize(
"output,expected",
[
tests/unit_tests/analyze_memory/__init__.py
View File
tests/unit_tests/analyze_memory/test_ci_memory_impact_comment.py
+99
View File
@@ -0,0 +1,99 @@
"""Tests for script/ci_memory_impact_comment.py symbol matching."""
from pathlib import Path
import sys
# Add script directory to path so we can import the module
sys.path.insert(0, str(Path(__file__).parent.parent.parent.parent / "script"))
from ci_memory_impact_comment import prepare_symbol_changes_data # noqa: E402
def test_prepare_symbol_changes_signature_match() -> None:
"""Symbols with same base name but different args are matched as changed."""
target = {
"Foo::bar(std::vector<unsigned char>&, int)": 300,
"unchanged()": 50,
}
pr = {
"Foo::bar(ProtoByteBuffer&, int)": 320,
"unchanged()": 50,
}
result = prepare_symbol_changes_data(target, pr)
assert result is not None
assert len(result["changed_symbols"]) == 1
assert len(result["new_symbols"]) == 0
assert len(result["removed_symbols"]) == 0
sym, t_size, p_size, delta = result["changed_symbols"][0]
assert sym == "Foo::bar(ProtoByteBuffer&, int)"
assert t_size == 300
assert p_size == 320
assert delta == 20
def test_prepare_symbol_changes_ambiguous_overloads_not_matched() -> None:
"""Multiple overloads with same base name stay as new/removed."""
target = {
"Foo::bar(int)": 100,
"Foo::bar(float)": 200,
}
pr = {
"Foo::bar(double)": 150,
"Foo::bar(long)": 250,
}
result = prepare_symbol_changes_data(target, pr)
assert result is not None
assert len(result["changed_symbols"]) == 0
assert len(result["new_symbols"]) == 2
assert len(result["removed_symbols"]) == 2
def test_prepare_symbol_changes_no_parens_not_matched() -> None:
"""Symbols without parens (variables) are not fuzzy-matched."""
target = {"my_global_var": 100}
pr = {"my_global_var_v2": 120}
result = prepare_symbol_changes_data(target, pr)
assert result is not None
assert len(result["changed_symbols"]) == 0
assert len(result["new_symbols"]) == 1
assert len(result["removed_symbols"]) == 1
def test_prepare_symbol_changes_nested_symbols_matched_separately() -> None:
"""Nested symbols like ::__pstr__ don't collide with parent function."""
target = {
"Foo::bar(std::vector<unsigned char>&, int)": 300,
"Foo::bar(std::vector<unsigned char>&, int)::__pstr__": 19,
}
pr = {
"Foo::bar(ProtoByteBuffer&, int)": 320,
"Foo::bar(ProtoByteBuffer&, int)::__pstr__": 19,
}
result = prepare_symbol_changes_data(target, pr)
assert result is not None
# Both the function and its nested __pstr__ should be matched (not new/removed)
assert len(result["new_symbols"]) == 0
assert len(result["removed_symbols"]) == 0
# __pstr__ has delta=0 so it's silently dropped, only the function shows
assert len(result["changed_symbols"]) == 1
sym, t_size, p_size, delta = result["changed_symbols"][0]
assert sym == "Foo::bar(ProtoByteBuffer&, int)"
assert delta == 20
def test_prepare_symbol_changes_exact_match_preferred() -> None:
"""Exact name matches are found before fuzzy matching runs."""
target = {
"Foo::bar(int)": 100,
}
pr = {
"Foo::bar(int)": 120,
}
result = prepare_symbol_changes_data(target, pr)
assert result is not None
assert len(result["changed_symbols"]) == 1
assert len(result["new_symbols"]) == 0
assert len(result["removed_symbols"]) == 0
sym, t_size, p_size, delta = result["changed_symbols"][0]
assert sym == "Foo::bar(int)"
assert delta == 20
tests/unit_tests/core/test_entity_helpers.py
+269 -73
View File
@@ -9,6 +9,7 @@ import pytest
from esphome.config_validation import Invalid
from esphome.const import (
CONF_DEVICE_CLASS,
CONF_DEVICE_ID,
CONF_DISABLED_BY_DEFAULT,
CONF_ENTITY_CATEGORY,
@@ -16,24 +17,31 @@ from esphome.const import (
CONF_ID,
CONF_INTERNAL,
CONF_NAME,
CONF_UNIT_OF_MEASUREMENT,
)
from esphome.core import CORE, ID, entity_helpers
from esphome.core.entity_helpers import (
_register_string,
_setup_entity_impl,
entity_duplicate_validator,
finalize_entity_strings,
get_base_entity_object_id,
register_device_class,
register_icon,
setup_device_class,
setup_entity,
setup_unit_of_measurement,
)
from esphome.cpp_generator import MockObj
from esphome.helpers import sanitize, snake_case
from .common import load_config_from_fixture
# Pre-compiled regex pattern for extracting names from set_name calls
# Matches: .set_name("name", hash) or .set_name("name")
SET_NAME_PATTERN = re.compile(r'\.set_name\(["\']([^"\']*)["\']')
# Pre-compiled regex pattern for extracting names from configure_entity_/set_name calls
# Matches: .configure_entity_("name", ...) or .set_name("name", ...)
ENTITY_NAME_PATTERN = re.compile(
r'\.(?:configure_entity_|set_name)\(["\']([^"\']*)["\']'
)
FIXTURES_DIR = Path(__file__).parent.parent / "fixtures" / "core" / "entity_helpers"
@@ -275,15 +283,23 @@ def setup_test_environment() -> Generator[list[str], None, None]:
entity_helpers.add = original_add
def extract_object_id_from_expressions(expressions: list[str]) -> str | None:
"""Extract the object ID that would be computed from set_name calls.
def extract_object_id_from_config(config: dict[str, Any]) -> str | None:
"""Extract the object ID from config keys set by _setup_entity_impl."""
name = config.get("_entity_name")
if name is None:
return None
if name:
return sanitize(snake_case(name))
# Empty name - fall back to friendly_name or device name
if CORE.friendly_name:
return sanitize(snake_case(CORE.friendly_name))
return sanitize(snake_case(CORE.name)) if CORE.name else None
Since object_id is now computed from the name (via snake_case + sanitize),
we extract the name from set_name() calls and compute the expected object_id.
For empty names, we fall back to CORE.friendly_name or CORE.name.
"""
def extract_object_id_from_expressions(expressions: list[str]) -> str | None:
"""Extract the object ID from configure_entity_() calls in generated expressions."""
for expr in expressions:
if match := SET_NAME_PATTERN.search(expr):
if match := ENTITY_NAME_PATTERN.search(expr):
name = match.group(1)
if name:
return sanitize(snake_case(name))
@@ -298,8 +314,6 @@ def extract_object_id_from_expressions(expressions: list[str]) -> str | None:
async def test_setup_entity_no_duplicates(setup_test_environment: list[str]) -> None:
"""Test setup_entity with unique names."""
added_expressions = setup_test_environment
# Create mock entities
var1 = MockObj("sensor1")
var2 = MockObj("sensor2")
@@ -311,13 +325,10 @@ async def test_setup_entity_no_duplicates(setup_test_environment: list[str]) ->
}
await _setup_entity_impl(var1, config1, "sensor")
# Get object ID from first entity
object_id1 = extract_object_id_from_expressions(added_expressions)
# Get object ID from first entity (stored in config, emitted later by finalize)
object_id1 = extract_object_id_from_config(config1)
assert object_id1 == "temperature"
# Clear for next entity
added_expressions.clear()
# Set up second entity with different name
config2 = {
CONF_NAME: "Humidity",
@@ -326,7 +337,7 @@ async def test_setup_entity_no_duplicates(setup_test_environment: list[str]) ->
await _setup_entity_impl(var2, config2, "sensor")
# Get object ID from second entity
object_id2 = extract_object_id_from_expressions(added_expressions)
object_id2 = extract_object_id_from_config(config2)
assert object_id2 == "humidity"
@@ -336,8 +347,6 @@ async def test_setup_entity_different_platforms(
) -> None:
"""Test that same name on different platforms doesn't conflict."""
added_expressions = setup_test_environment
# Create mock entities
sensor = MockObj("sensor1")
binary_sensor = MockObj("binary_sensor1")
@@ -355,15 +364,11 @@ async def test_setup_entity_different_platforms(
(text_sensor, "text_sensor"),
]
object_ids: list[str] = []
for var, platform in platforms:
added_expressions.clear()
await _setup_entity_impl(var, config, platform)
object_id = extract_object_id_from_expressions(added_expressions)
object_ids.append(object_id)
# All should get base object ID without suffix
assert all(obj_id == "status" for obj_id in object_ids)
# All should get the same object ID (name stored in config, not platform-specific)
assert extract_object_id_from_config(config) == "status"
@pytest.fixture
@@ -388,7 +393,6 @@ async def test_setup_entity_with_devices(
setup_test_environment: list[str], mock_get_variable: dict[ID, MockObj]
) -> None:
"""Test that same name on different devices doesn't conflict."""
added_expressions = setup_test_environment
# Create mock devices
device1_id = ID("device1", type="Device")
@@ -417,24 +421,18 @@ async def test_setup_entity_with_devices(
}
# Get object IDs
object_ids: list[str] = []
for var, config in [(sensor1, config1), (sensor2, config2)]:
added_expressions.clear()
await _setup_entity_impl(var, config, "sensor")
object_id = extract_object_id_from_expressions(added_expressions)
object_ids.append(object_id)
# Both should get base object ID without suffix (different devices)
assert object_ids[0] == "temperature"
assert object_ids[1] == "temperature"
assert extract_object_id_from_config(config1) == "temperature"
assert extract_object_id_from_config(config2) == "temperature"
@pytest.mark.asyncio
async def test_setup_entity_empty_name(setup_test_environment: list[str]) -> None:
"""Test setup_entity with empty entity name."""
added_expressions = setup_test_environment
var = MockObj("sensor1")
config = {
@@ -444,7 +442,7 @@ async def test_setup_entity_empty_name(setup_test_environment: list[str]) -> Non
await _setup_entity_impl(var, config, "sensor")
object_id = extract_object_id_from_expressions(added_expressions)
object_id = extract_object_id_from_config(config)
# Should use friendly name
assert object_id == "test_device"
@@ -455,8 +453,6 @@ async def test_setup_entity_special_characters(
) -> None:
"""Test setup_entity with names containing special characters."""
added_expressions = setup_test_environment
var = MockObj("sensor1")
config = {
@@ -465,7 +461,7 @@ async def test_setup_entity_special_characters(
}
await _setup_entity_impl(var, config, "sensor")
object_id = extract_object_id_from_expressions(added_expressions)
object_id = extract_object_id_from_config(config)
# Special characters should be sanitized
assert object_id == "temperature_sensor_"
@@ -475,8 +471,6 @@ async def test_setup_entity_special_characters(
async def test_setup_entity_with_icon(setup_test_environment: list[str]) -> None:
"""Test setup_entity sets icon correctly."""
setup_test_environment # noqa: F841 - fixture initializes CORE state
var = MockObj("sensor1")
config = {
@@ -497,8 +491,6 @@ async def test_setup_entity_disabled_by_default(
) -> None:
"""Test setup_entity sets disabled_by_default correctly."""
added_expressions = setup_test_environment
var = MockObj("sensor1")
config = {
@@ -508,10 +500,8 @@ async def test_setup_entity_disabled_by_default(
await _setup_entity_impl(var, config, "sensor")
# Check disabled_by_default was set
assert any(
"sensor1.set_disabled_by_default(true)" in expr for expr in added_expressions
)
# disabled_by_default is now packed into config for configure_entity_()
assert config.get("_entity_disabled_by_default") == 1
def test_entity_duplicate_validator() -> None:
@@ -796,13 +786,12 @@ async def test_setup_entity_empty_name_with_device(
entity_helpers.get_variable = original_get_variable
# Check that set_device was called
assert any("sensor1.set_device" in expr for expr in added_expressions)
# Check that set_device_ was called (separate protected call, accessible via friend)
assert any("sensor1.set_device_" in expr for expr in added_expressions)
# For empty-name entities, Python passes 0 - C++ calculates hash at runtime
assert any('set_name("", 0)' in expr for expr in added_expressions), (
f"Expected set_name with hash 0, got {added_expressions}"
)
# For empty-name entities, Python stores hash 0 - C++ calculates hash at runtime
assert config.get("_entity_name") == ""
assert config.get("_entity_object_id_hash") == 0
@pytest.mark.asyncio
@@ -814,7 +803,6 @@ async def test_setup_entity_empty_name_with_mac_suffix(
For empty-name entities, Python passes 0 and C++ calculates the hash
at runtime from friendly_name (bug-for-bug compatibility).
"""
added_expressions = setup_test_environment
# Set up CORE.config with name_add_mac_suffix enabled
CORE.config = {"name_add_mac_suffix": True}
@@ -830,10 +818,9 @@ async def test_setup_entity_empty_name_with_mac_suffix(
await _setup_entity_impl(var, config, "sensor")
# For empty-name entities, Python passes 0 - C++ calculates hash at runtime
assert any('set_name("", 0)' in expr for expr in added_expressions), (
f"Expected set_name with hash 0, got {added_expressions}"
)
# For empty-name entities, Python stores hash 0 - C++ calculates hash at runtime
assert config.get("_entity_name") == ""
assert config.get("_entity_object_id_hash") == 0
@pytest.mark.asyncio
@@ -846,7 +833,6 @@ async def test_setup_entity_empty_name_with_mac_suffix_no_friendly_name(
at runtime. In this case C++ will hash the empty friendly_name
(bug-for-bug compatibility).
"""
added_expressions = setup_test_environment
# Set up CORE.config with name_add_mac_suffix enabled
CORE.config = {"name_add_mac_suffix": True}
@@ -862,10 +848,9 @@ async def test_setup_entity_empty_name_with_mac_suffix_no_friendly_name(
await _setup_entity_impl(var, config, "sensor")
# For empty-name entities, Python passes 0 - C++ calculates hash at runtime
assert any('set_name("", 0)' in expr for expr in added_expressions), (
f"Expected set_name with hash 0, got {added_expressions}"
)
# For empty-name entities, Python stores hash 0 - C++ calculates hash at runtime
assert config.get("_entity_name") == ""
assert config.get("_entity_object_id_hash") == 0
@pytest.mark.asyncio
@@ -877,7 +862,6 @@ async def test_setup_entity_empty_name_no_mac_suffix_no_friendly_name(
For empty-name entities, Python passes 0 and C++ calculates the hash
at runtime from the device name.
"""
added_expressions = setup_test_environment
# No MAC suffix (either not set or False)
CORE.config = {}
@@ -895,10 +879,9 @@ async def test_setup_entity_empty_name_no_mac_suffix_no_friendly_name(
await _setup_entity_impl(var, config, "sensor")
# For empty-name entities, Python passes 0 - C++ calculates hash at runtime
assert any('set_name("", 0)' in expr for expr in added_expressions), (
f"Expected set_name with hash 0, got {added_expressions}"
)
# For empty-name entities, Python stores hash 0 - C++ calculates hash at runtime
assert config.get("_entity_name") == ""
assert config.get("_entity_object_id_hash") == 0
def test_register_string_overflow() -> None:
@@ -926,11 +909,27 @@ def test_register_icon_max_length() -> None:
assert register_icon("") == 0
def test_register_device_class_max_length() -> None:
"""Test register_device_class rejects device classes exceeding 47 characters."""
# 47 chars should succeed
max_dc = "a" * 47
idx = register_device_class(max_dc)
assert idx > 0
# 48 chars should fail
too_long = "a" * 48
with pytest.raises(ValueError, match="Device class string too long"):
register_device_class(too_long)
# Empty string returns 0
assert register_device_class("") == 0
@pytest.mark.asyncio
async def test_setup_entity_with_entity_category(
setup_test_environment: list[str],
) -> None:
"""Test setup_entity sets entity_category correctly."""
"""Test entity_category is packed correctly through the full setup flow."""
added_expressions = setup_test_environment
var = MockObj("sensor1")
config = {
@@ -939,9 +938,10 @@ async def test_setup_entity_with_entity_category(
CONF_ENTITY_CATEGORY: "diagnostic",
}
await _setup_entity_impl(var, config, "sensor")
assert any(
'set_entity_category("diagnostic")' in expr for expr in added_expressions
)
finalize_entity_strings(var, config)
packed = _extract_packed_value(added_expressions)
assert packed != 0
assert "category:diagnostic" in added_expressions[0]
@pytest.mark.asyncio
@@ -959,7 +959,7 @@ async def test_setup_entity_direct_call(setup_test_environment: list[str]) -> No
# Direct call mode: await setup_entity(var, config, "camera")
await setup_entity(var, config, "camera")
# Should have called set_name
# Should have emitted configure_entity_
object_id = extract_object_id_from_expressions(added_expressions)
assert object_id == "my_camera"
@@ -990,3 +990,199 @@ async def test_setup_entity_decorator_mode(setup_test_environment: list[str]) ->
assert body_called
object_id = extract_object_id_from_expressions(added_expressions)
assert object_id == "temperature"
# Tests for finalize_entity_strings packing
#
# These tests verify that flags and string indices produce non-zero packed values
# and correct inline comments. The actual bit layout correctness (Python _*_SHIFT
# matching C++ ENTITY_FIELD_*_SHIFT) is verified end-to-end by the integration
# test test_host_mode_entity_fields, which compiles firmware and checks values
# via the native API.
def _extract_packed_value(expressions: list[str]) -> int:
"""Extract the third argument (packed value) from a configure_entity_() call."""
for expr in expressions:
if "configure_entity_" in expr:
# Match the last integer argument before the closing ");"
match = re.search(r",\s*(\d+)\s*\)", expr)
if match:
return int(match.group(1))
raise AssertionError("No configure_entity_ call found")
@pytest.mark.asyncio
async def test_finalize_no_flags(setup_test_environment: list[str]) -> None:
"""Test entity with no special flags — packed value is 0, no comment."""
added_expressions = setup_test_environment
var = MockObj("sensor1")
config = {
CONF_NAME: "Test",
CONF_DISABLED_BY_DEFAULT: False,
}
await _setup_entity_impl(var, config, "sensor")
finalize_entity_strings(var, config)
packed = _extract_packed_value(added_expressions)
assert packed == 0
assert "//" not in added_expressions[0]
@pytest.mark.asyncio
async def test_finalize_internal(setup_test_environment: list[str]) -> None:
"""Test entity with internal=True packs the internal flag."""
added_expressions = setup_test_environment
var = MockObj("sensor1")
config = {
CONF_NAME: "Test",
CONF_DISABLED_BY_DEFAULT: False,
CONF_INTERNAL: True,
}
await _setup_entity_impl(var, config, "sensor")
finalize_entity_strings(var, config)
packed = _extract_packed_value(added_expressions)
assert packed != 0
assert "// internal" in added_expressions[0]
@pytest.mark.asyncio
async def test_finalize_disabled_by_default(
setup_test_environment: list[str],
) -> None:
"""Test entity with disabled_by_default=True packs the flag."""
added_expressions = setup_test_environment
var = MockObj("sensor1")
config = {
CONF_NAME: "Test",
CONF_DISABLED_BY_DEFAULT: True,
}
await _setup_entity_impl(var, config, "sensor")
finalize_entity_strings(var, config)
packed = _extract_packed_value(added_expressions)
assert packed != 0
assert "// disabled_by_default" in added_expressions[0]
@pytest.mark.asyncio
async def test_finalize_entity_category(
setup_test_environment: list[str],
) -> None:
"""Test entity_category values are packed and described in comment."""
added_expressions = setup_test_environment
var = MockObj("sensor1")
# Test diagnostic
config = {
CONF_NAME: "Test",
CONF_DISABLED_BY_DEFAULT: False,
CONF_ENTITY_CATEGORY: "diagnostic",
}
await _setup_entity_impl(var, config, "sensor")
finalize_entity_strings(var, config)
packed_diag = _extract_packed_value(added_expressions)
assert packed_diag != 0
assert "category:diagnostic" in added_expressions[0]
# Test config — different packed value
added_expressions.clear()
config2 = {
CONF_NAME: "Test2",
CONF_DISABLED_BY_DEFAULT: False,
CONF_ENTITY_CATEGORY: "config",
}
await _setup_entity_impl(var, config2, "sensor")
finalize_entity_strings(var, config2)
packed_cfg = _extract_packed_value(added_expressions)
assert packed_cfg != 0
assert packed_cfg != packed_diag
assert "category:config" in added_expressions[0]
@pytest.mark.asyncio
async def test_finalize_string_indices(
setup_test_environment: list[str],
) -> None:
"""Test device_class, unit_of_measurement, and icon produce non-zero packed value."""
added_expressions = setup_test_environment
var = MockObj("sensor1")
config = {
CONF_NAME: "Test",
CONF_DISABLED_BY_DEFAULT: False,
CONF_DEVICE_CLASS: "temperature",
CONF_UNIT_OF_MEASUREMENT: "°C",
CONF_ICON: "mdi:thermometer",
}
await _setup_entity_impl(var, config, "sensor")
setup_device_class(config)
setup_unit_of_measurement(config)
finalize_entity_strings(var, config)
packed = _extract_packed_value(added_expressions)
assert packed != 0
comment = added_expressions[0]
assert "dc:temperature" in comment
assert "uom:°C" in comment
assert "icon:mdi:thermometer" in comment
@pytest.mark.asyncio
async def test_finalize_all_fields(
setup_test_environment: list[str],
) -> None:
"""Test all fields set: flags, string indices, and comment."""
added_expressions = setup_test_environment
var = MockObj("sensor1")
config = {
CONF_NAME: "Test",
CONF_DISABLED_BY_DEFAULT: True,
CONF_INTERNAL: True,
CONF_ENTITY_CATEGORY: "diagnostic",
CONF_DEVICE_CLASS: "temperature",
CONF_UNIT_OF_MEASUREMENT: "°C",
CONF_ICON: "mdi:thermometer",
}
await _setup_entity_impl(var, config, "sensor")
setup_device_class(config)
setup_unit_of_measurement(config)
finalize_entity_strings(var, config)
packed = _extract_packed_value(added_expressions)
assert packed != 0
# Verify comment contains all flags with actual string values
comment_line = added_expressions[0]
assert (
"// internal, disabled_by_default, category:diagnostic,"
" dc:temperature, uom:°C, icon:mdi:thermometer" in comment_line
)
@pytest.mark.asyncio
async def test_finalize_comment_sanitization(
setup_test_environment: list[str],
) -> None:
"""Test that user strings in comments are sanitized against injection."""
added_expressions = setup_test_environment
var = MockObj("sensor1")
config = {
CONF_NAME: "Test",
CONF_DISABLED_BY_DEFAULT: False,
# Backslash at end would cause line splice eating next code line
CONF_ICON: "mdi:evil\\",
}
await _setup_entity_impl(var, config, "sensor")
finalize_entity_strings(var, config)
comment_line = added_expressions[0]
# Backslash must be replaced to prevent line splice
assert "\\" not in comment_line
assert "mdi:evil/" in comment_line
added_expressions.clear()
config2 = {
CONF_NAME: "Test2",
CONF_DISABLED_BY_DEFAULT: False,
CONF_ICON: "mdi:evil\nINJECTED_CODE();",
}
await _setup_entity_impl(var, config2, "sensor")
finalize_entity_strings(var, config2)
comment_line = added_expressions[0]
# Newline must be replaced to prevent breaking out of comment
assert "\n" not in comment_line
assert "INJECTED_CODE" in comment_line # still visible but safe in comment
tests/unit_tests/test_core.py
+12
View File
@@ -841,6 +841,18 @@ class TestEsphomeCore:
assert "WiFi" in target.platformio_libraries
def test_testing_ensure_platform_registered__sets_count(self, target):
"""Test testing_ensure_platform_registered sets count to 1 for new platform."""
assert target.platform_counts["sensor"] == 0
target.testing_ensure_platform_registered("sensor")
assert target.platform_counts["sensor"] == 1
def test_testing_ensure_platform_registered__does_not_overwrite(self, target):
"""Test testing_ensure_platform_registered preserves existing count."""
target.platform_counts["sensor"] = 3
target.testing_ensure_platform_registered("sensor")
assert target.platform_counts["sensor"] == 3
def test_add_library__extracts_short_name_from_path(self, target):
"""Test add_library extracts short name from library paths like owner/lib."""
target.data[const.KEY_CORE] = {