mirror of
https://github.com/esphome/esphome.git
synced 2026-09-24 05:24:14 +00:00
Merge branch 'move_icons_progmem' into move_device_class_progmem
This commit is contained in:
J. Nick Koston
@@ -66,5 +66,20 @@ def test_text_config_lamda_is_set(generate_main):
|
||||
main_cpp = generate_main("tests/component_tests/text/test_text.yaml")
|
||||
|
||||
# Then
|
||||
assert "it_4->set_template([]() -> esphome::optional<std::string> {" in main_cpp
|
||||
assert "it_4->set_template([]() -> std::optional<std::string> {" in main_cpp
|
||||
assert 'return std::string{"Hello"};' in main_cpp
|
||||
|
||||
|
||||
def test_esphome_optional_alias_works(generate_main):
|
||||
"""
|
||||
Test that esphome::optional alias compiles (backward compatibility)
|
||||
"""
|
||||
# Given
|
||||
|
||||
# When
|
||||
main_cpp = generate_main("tests/component_tests/text/test_text.yaml")
|
||||
|
||||
# Then
|
||||
# Codegen emits std::optional, but esphome::optional must also work
|
||||
# via the using alias in esphome/core/optional.h
|
||||
assert "std::optional<std::string>" in main_cpp
|
||||
|
||||
@@ -0,0 +1,9 @@
|
||||
audio_file:
|
||||
- id: test_audio
|
||||
file:
|
||||
type: local
|
||||
path: $component_dir/test.wav
|
||||
|
||||
media_source:
|
||||
- platform: audio_file
|
||||
id: audio_file_source
|
||||
@@ -0,0 +1 @@
|
||||
<<: !include common.yaml
|
||||
Binary file not shown.
@@ -0,0 +1,4 @@
|
||||
ble_nus:
|
||||
type: uart
|
||||
tx_buffer_size: 160
|
||||
rx_buffer_size: 160
|
||||
@@ -27,3 +27,14 @@ globals:
|
||||
type: bool
|
||||
restore_value: false
|
||||
initial_value: "false"
|
||||
# Test restore_value with string "false" - should be converted to bool false
|
||||
- id: glob_no_restore_string_false
|
||||
type: int
|
||||
restore_value: "false"
|
||||
initial_value: "42"
|
||||
# Test restore_value with string "true" - should be converted to bool true
|
||||
- id: glob_restore_string_true
|
||||
type: int
|
||||
restore_value: "true"
|
||||
initial_value: "99"
|
||||
update_interval: 5s
|
||||
|
||||
@@ -1,9 +1,19 @@
|
||||
esphome:
|
||||
on_boot:
|
||||
then:
|
||||
- sensor.integration.reset:
|
||||
id: integration_sensor
|
||||
- sensor.integration.set_value:
|
||||
id: integration_sensor
|
||||
value: 100.0
|
||||
|
||||
sensor:
|
||||
- platform: adc
|
||||
id: my_sensor
|
||||
pin: ${pin}
|
||||
attenuation: 12db
|
||||
- platform: integration
|
||||
id: integration_sensor
|
||||
sensor: my_sensor
|
||||
name: Integration Sensor
|
||||
time_unit: s
|
||||
|
||||
@@ -1,3 +1,5 @@
|
||||
modbus:
|
||||
id: mod_bus1
|
||||
flow_control_pin: ${flow_control_pin}
|
||||
send_wait_time: 500ms
|
||||
turnaround_time: 100ms
|
||||
|
||||
@@ -1,3 +1,9 @@
|
||||
esp32:
|
||||
board: esp32-c6-devkitc-1
|
||||
framework:
|
||||
type: esp-idf
|
||||
log_level: DEBUG
|
||||
|
||||
network:
|
||||
enable_ipv6: true
|
||||
|
||||
|
||||
@@ -0,0 +1,98 @@
|
||||
#pragma once
|
||||
#include <cstdint>
|
||||
#include <cstring>
|
||||
#include <cstdio>
|
||||
#include <vector>
|
||||
#include <gtest/gtest.h>
|
||||
#include "esphome/components/packet_transport/packet_transport.h"
|
||||
|
||||
namespace esphome::packet_transport::testing {
|
||||
|
||||
// Protocol constants mirrored from packet_transport.cpp for test packet construction.
|
||||
static constexpr uint16_t MAGIC_NUMBER = 0x4553;
|
||||
static constexpr uint16_t MAGIC_PING = 0x5048;
|
||||
|
||||
// Concrete testable implementation of PacketTransport.
|
||||
// Captures sent packets and exposes protected members for verification.
|
||||
//
|
||||
// Sensor round-trip tests require USE_SENSOR / USE_BINARY_SENSOR to be defined,
|
||||
// which happens when 'sensor' and 'binary_sensor' components are in the build.
|
||||
// Run with --all or include those components to enable the full test suite.
|
||||
class TestablePacketTransport : public PacketTransport {
|
||||
public:
|
||||
using PacketTransport::add_key_;
|
||||
using PacketTransport::data_;
|
||||
using PacketTransport::encryption_key_;
|
||||
using PacketTransport::flush_;
|
||||
using PacketTransport::header_;
|
||||
using PacketTransport::increment_code_;
|
||||
using PacketTransport::init_data_;
|
||||
using PacketTransport::is_encrypted_;
|
||||
using PacketTransport::is_provider_;
|
||||
using PacketTransport::name_;
|
||||
using PacketTransport::ping_key_;
|
||||
using PacketTransport::ping_keys_;
|
||||
using PacketTransport::ping_pong_enable_;
|
||||
using PacketTransport::ping_pong_recyle_time_;
|
||||
using PacketTransport::process_;
|
||||
using PacketTransport::providers_;
|
||||
using PacketTransport::rolling_code_;
|
||||
using PacketTransport::rolling_code_enable_;
|
||||
using PacketTransport::send_data_;
|
||||
using PacketTransport::updated_;
|
||||
#ifdef USE_SENSOR
|
||||
using PacketTransport::add_data_;
|
||||
using PacketTransport::remote_sensors_;
|
||||
using PacketTransport::sensors_;
|
||||
#endif
|
||||
#ifdef USE_BINARY_SENSOR
|
||||
using PacketTransport::add_binary_data_;
|
||||
using PacketTransport::binary_sensors_;
|
||||
using PacketTransport::remote_binary_sensors_;
|
||||
#endif
|
||||
|
||||
// NOTE: std::vector is used here for test convenience. For production code,
|
||||
// consider using StaticVector or FixedVector from esphome/core/helpers.h instead.
|
||||
mutable std::vector<std::vector<uint8_t>> sent_packets;
|
||||
size_t max_packet_size{512};
|
||||
bool send_enabled{true};
|
||||
|
||||
void send_packet(const std::vector<uint8_t> &buf) const override { this->sent_packets.push_back(buf); }
|
||||
size_t get_max_packet_size() override { return this->max_packet_size; }
|
||||
bool should_send() override { return this->send_enabled; }
|
||||
|
||||
/// Build the packet header for testing without requiring App or global_preferences.
|
||||
void init_for_test(const char *name) {
|
||||
this->name_ = name;
|
||||
this->header_.clear();
|
||||
// MAGIC_NUMBER as uint16_t little-endian
|
||||
this->header_.push_back(MAGIC_NUMBER & 0xFF);
|
||||
this->header_.push_back((MAGIC_NUMBER >> 8) & 0xFF);
|
||||
// Length-prefixed hostname
|
||||
auto len = strlen(name);
|
||||
this->header_.push_back(static_cast<uint8_t>(len));
|
||||
for (size_t i = 0; i < len; i++)
|
||||
this->header_.push_back(name[i]);
|
||||
// Pad to 4-byte boundary
|
||||
while (this->header_.size() & 0x3)
|
||||
this->header_.push_back(0);
|
||||
}
|
||||
};
|
||||
|
||||
/// Build a MAGIC_PING packet for testing add_key_ / ping-pong flows.
|
||||
inline std::vector<uint8_t> build_ping_packet(const char *hostname, uint32_t key) {
|
||||
std::vector<uint8_t> packet;
|
||||
packet.push_back(MAGIC_PING & 0xFF);
|
||||
packet.push_back((MAGIC_PING >> 8) & 0xFF);
|
||||
auto len = strlen(hostname);
|
||||
packet.push_back(static_cast<uint8_t>(len));
|
||||
for (size_t i = 0; i < len; i++)
|
||||
packet.push_back(hostname[i]);
|
||||
packet.push_back(key & 0xFF);
|
||||
packet.push_back((key >> 8) & 0xFF);
|
||||
packet.push_back((key >> 16) & 0xFF);
|
||||
packet.push_back((key >> 24) & 0xFF);
|
||||
return packet;
|
||||
}
|
||||
|
||||
} // namespace esphome::packet_transport::testing
|
||||
@@ -0,0 +1,11 @@
|
||||
# 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
|
||||
@@ -0,0 +1,445 @@
|
||||
#include "common.h"
|
||||
|
||||
namespace esphome::packet_transport::testing {
|
||||
|
||||
// --- Configuration setter tests ---
|
||||
|
||||
TEST(PacketTransportTest, SetIsProvider) {
|
||||
TestablePacketTransport transport;
|
||||
transport.set_is_provider(true);
|
||||
EXPECT_TRUE(transport.is_provider_);
|
||||
}
|
||||
|
||||
TEST(PacketTransportTest, SetEncryptionKey) {
|
||||
TestablePacketTransport transport;
|
||||
std::vector<uint8_t> key(32, 0xAB);
|
||||
transport.set_encryption_key(key);
|
||||
EXPECT_EQ(transport.encryption_key_, key);
|
||||
EXPECT_TRUE(transport.is_encrypted_());
|
||||
}
|
||||
|
||||
TEST(PacketTransportTest, NoEncryptionByDefault) {
|
||||
TestablePacketTransport transport;
|
||||
EXPECT_FALSE(transport.is_encrypted_());
|
||||
}
|
||||
|
||||
TEST(PacketTransportTest, SetRollingCodeEnable) {
|
||||
TestablePacketTransport transport;
|
||||
transport.set_rolling_code_enable(true);
|
||||
EXPECT_TRUE(transport.rolling_code_enable_);
|
||||
}
|
||||
|
||||
TEST(PacketTransportTest, SetPingPongEnable) {
|
||||
TestablePacketTransport transport;
|
||||
transport.set_ping_pong_enable(true);
|
||||
EXPECT_TRUE(transport.ping_pong_enable_);
|
||||
}
|
||||
|
||||
TEST(PacketTransportTest, SetPingPongRecycleTime) {
|
||||
TestablePacketTransport transport;
|
||||
transport.set_ping_pong_recycle_time(600);
|
||||
EXPECT_EQ(transport.ping_pong_recyle_time_, 600u);
|
||||
}
|
||||
|
||||
// --- Provider management ---
|
||||
|
||||
TEST(PacketTransportTest, AddProvider) {
|
||||
TestablePacketTransport transport;
|
||||
transport.add_provider("host1");
|
||||
EXPECT_TRUE(transport.providers_.contains("host1"));
|
||||
EXPECT_EQ(transport.providers_.size(), 1u);
|
||||
}
|
||||
|
||||
TEST(PacketTransportTest, AddProviderDuplicate) {
|
||||
TestablePacketTransport transport;
|
||||
transport.add_provider("host1");
|
||||
transport.add_provider("host1");
|
||||
EXPECT_EQ(transport.providers_.size(), 1u);
|
||||
}
|
||||
|
||||
TEST(PacketTransportTest, SetProviderEncryption) {
|
||||
TestablePacketTransport transport;
|
||||
transport.add_provider("host1");
|
||||
std::vector<uint8_t> key(32, 0xCD);
|
||||
transport.set_provider_encryption("host1", key);
|
||||
EXPECT_EQ(transport.providers_["host1"].encryption_key, key);
|
||||
}
|
||||
|
||||
// --- 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) {
|
||||
TestablePacketTransport transport;
|
||||
transport.init_for_test("receiver");
|
||||
transport.set_encryption_key(std::vector<uint8_t>(32, 0xAA));
|
||||
|
||||
auto ping = build_ping_packet("requester", 0xDEADBEEF);
|
||||
transport.process_({ping.data(), ping.size()});
|
||||
|
||||
ASSERT_EQ(transport.ping_keys_.size(), 1u);
|
||||
EXPECT_EQ(transport.ping_keys_["requester"], 0xDEADBEEFu);
|
||||
}
|
||||
|
||||
TEST(PacketTransportTest, PingKeyIgnoredWhenNotEncrypted) {
|
||||
TestablePacketTransport transport;
|
||||
transport.init_for_test("receiver");
|
||||
// No encryption key — add_key_ should be a no-op
|
||||
|
||||
auto ping = build_ping_packet("requester", 0xDEADBEEF);
|
||||
transport.process_({ping.data(), ping.size()});
|
||||
|
||||
EXPECT_TRUE(transport.ping_keys_.empty());
|
||||
}
|
||||
|
||||
TEST(PacketTransportTest, PingKeyUpdatedOnRepeat) {
|
||||
TestablePacketTransport transport;
|
||||
transport.init_for_test("receiver");
|
||||
transport.set_encryption_key(std::vector<uint8_t>(32, 0xAA));
|
||||
|
||||
auto ping1 = build_ping_packet("host1", 0x1111);
|
||||
transport.process_({ping1.data(), ping1.size()});
|
||||
EXPECT_EQ(transport.ping_keys_["host1"], 0x1111u);
|
||||
|
||||
// Same host, new key value — should update in place
|
||||
auto ping2 = build_ping_packet("host1", 0x2222);
|
||||
transport.process_({ping2.data(), ping2.size()});
|
||||
EXPECT_EQ(transport.ping_keys_.size(), 1u);
|
||||
EXPECT_EQ(transport.ping_keys_["host1"], 0x2222u);
|
||||
}
|
||||
|
||||
TEST(PacketTransportTest, PingKeyMaxLimit) {
|
||||
TestablePacketTransport transport;
|
||||
transport.init_for_test("receiver");
|
||||
transport.set_encryption_key(std::vector<uint8_t>(32, 0xAA));
|
||||
|
||||
// Fill to MAX_PING_KEYS (4)
|
||||
for (int i = 0; i < 4; i++) {
|
||||
char name[16];
|
||||
snprintf(name, sizeof(name), "host%d", i);
|
||||
auto ping = build_ping_packet(name, 0x1000 + i);
|
||||
transport.process_({ping.data(), ping.size()});
|
||||
}
|
||||
EXPECT_EQ(transport.ping_keys_.size(), 4u);
|
||||
|
||||
// 5th key should be discarded
|
||||
auto ping = build_ping_packet("host4", 0x9999);
|
||||
transport.process_({ping.data(), ping.size()});
|
||||
EXPECT_EQ(transport.ping_keys_.size(), 4u);
|
||||
EXPECT_FALSE(transport.ping_keys_.contains("host4"));
|
||||
}
|
||||
|
||||
#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) {
|
||||
TestablePacketTransport transport;
|
||||
transport.init_for_test("receiver");
|
||||
uint8_t buf[] = {0x53};
|
||||
// Too short for a magic number - should return safely
|
||||
transport.process_({buf, 1});
|
||||
}
|
||||
|
||||
TEST(PacketTransportTest, ProcessBadMagic) {
|
||||
TestablePacketTransport transport;
|
||||
transport.init_for_test("receiver");
|
||||
uint8_t buf[] = {0xFF, 0xFF, 0x00, 0x00};
|
||||
// Wrong magic - should return safely
|
||||
transport.process_({buf, sizeof(buf)});
|
||||
}
|
||||
|
||||
TEST(PacketTransportTest, ProcessOwnHostname) {
|
||||
TestablePacketTransport transport;
|
||||
transport.init_for_test("myself");
|
||||
// Build a packet from "myself" using a separate encoder
|
||||
TestablePacketTransport fake_sender;
|
||||
fake_sender.init_for_test("myself");
|
||||
fake_sender.send_data_(true);
|
||||
ASSERT_EQ(fake_sender.sent_packets.size(), 1u);
|
||||
|
||||
auto &packet = fake_sender.sent_packets[0];
|
||||
// Should be silently ignored because hostname matches our own
|
||||
transport.process_({packet.data(), packet.size()});
|
||||
}
|
||||
|
||||
TEST(PacketTransportTest, ProcessUnknownHostname) {
|
||||
TestablePacketTransport transport;
|
||||
transport.init_for_test("receiver");
|
||||
// No providers registered - "unknown" will not be found
|
||||
TestablePacketTransport sender;
|
||||
sender.init_for_test("unknown");
|
||||
sender.send_data_(true);
|
||||
ASSERT_EQ(sender.sent_packets.size(), 1u);
|
||||
|
||||
auto &packet = sender.sent_packets[0];
|
||||
// Should return safely without crash
|
||||
transport.process_({packet.data(), packet.size()});
|
||||
}
|
||||
|
||||
// --- Send disabled ---
|
||||
|
||||
TEST(PacketTransportTest, NoSendWhenDisabled) {
|
||||
TestablePacketTransport transport;
|
||||
transport.init_for_test("sender");
|
||||
transport.send_enabled = false;
|
||||
transport.send_data_(true);
|
||||
EXPECT_TRUE(transport.sent_packets.empty());
|
||||
}
|
||||
|
||||
} // namespace esphome::packet_transport::testing
|
||||
@@ -28,9 +28,14 @@ esphome:
|
||||
# Test C++ API: set_template() with stateless lambda (no captures)
|
||||
# NOTE: set_template() is not intended to be a public API, but we test it to ensure it doesn't break.
|
||||
- lambda: |-
|
||||
id(template_sens).set_template([]() -> esphome::optional<float> {
|
||||
id(template_sens).set_template([]() -> std::optional<float> {
|
||||
return 123.0f;
|
||||
});
|
||||
# Test that esphome::optional alias still works for backward compatibility
|
||||
- lambda: |-
|
||||
id(template_sens).set_template([]() -> esphome::optional<float> {
|
||||
return 42.0f;
|
||||
});
|
||||
|
||||
- datetime.date.set:
|
||||
id: test_date
|
||||
|
||||
@@ -73,11 +73,6 @@ def shared_platformio_cache() -> Generator[Path]:
|
||||
test_cache_dir = Path.home() / ".esphome-integration-tests"
|
||||
cache_dir = test_cache_dir / "platformio"
|
||||
|
||||
# Create the temp directory that PlatformIO uses to avoid race conditions
|
||||
# This ensures it exists and won't be deleted by parallel processes
|
||||
platformio_tmp_dir = cache_dir / ".cache" / "tmp"
|
||||
platformio_tmp_dir.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
# Use a lock file in the home directory to ensure only one process initializes the cache
|
||||
# This is needed when running with pytest-xdist
|
||||
# The lock file must be in a directory that already exists to avoid race conditions
|
||||
@@ -87,8 +82,9 @@ def shared_platformio_cache() -> Generator[Path]:
|
||||
with open(lock_file, "w") as lock_fd:
|
||||
fcntl.flock(lock_fd.fileno(), fcntl.LOCK_EX)
|
||||
|
||||
# Check if cache needs initialization while holding the lock
|
||||
if not cache_dir.exists() or not any(cache_dir.iterdir()):
|
||||
# Check if the native platform is installed (the actual indicator of a populated cache)
|
||||
native_platform = cache_dir / "platforms" / "native"
|
||||
if not native_platform.exists():
|
||||
# Create the test cache directory if it doesn't exist
|
||||
test_cache_dir.mkdir(exist_ok=True)
|
||||
|
||||
|
||||
@@ -43,6 +43,7 @@ CONF_INJECT_RX = "inject_rx"
|
||||
CONF_EXPECT_TX = "expect_tx"
|
||||
CONF_PERIODIC_RX = "periodic_rx"
|
||||
CONF_ON_TX = "on_tx"
|
||||
CONF_AUTO_START = "auto_start"
|
||||
|
||||
UART_PARITY_OPTIONS = {
|
||||
"NONE": uart.UARTParityOptions.UART_CONFIG_PARITY_NONE,
|
||||
@@ -70,6 +71,7 @@ RESPONSE_SCHEMA = cv.Schema(
|
||||
{
|
||||
cv.Required(CONF_EXPECT_TX): [cv.hex_uint8_t],
|
||||
cv.Required(CONF_INJECT_RX): [cv.hex_uint8_t],
|
||||
cv.Optional(CONF_DELAY, default="0ms"): cv.positive_time_period_milliseconds,
|
||||
}
|
||||
)
|
||||
|
||||
@@ -95,6 +97,7 @@ CONFIG_SCHEMA = cv.Schema(
|
||||
cv.Optional(CONF_INJECTIONS, default=[]): cv.ensure_list(INJECTION_SCHEMA),
|
||||
cv.Optional(CONF_RESPONSES, default=[]): cv.ensure_list(RESPONSE_SCHEMA),
|
||||
cv.Optional(CONF_PERIODIC_RX, default=[]): cv.ensure_list(PERIODIC_RX_SCHEMA),
|
||||
cv.Optional(CONF_AUTO_START, default=True): cv.boolean,
|
||||
cv.Optional(CONF_ON_TX): automation.validate_automation(
|
||||
{
|
||||
cv.GenerateID(CONF_TRIGGER_ID): cv.declare_id(MockUartTXTrigger),
|
||||
@@ -138,6 +141,9 @@ async def to_code(config):
|
||||
cg.add(var.set_data_bits(config[CONF_DATA_BITS]))
|
||||
cg.add(var.set_parity(config[CONF_PARITY]))
|
||||
|
||||
if not config[CONF_AUTO_START]:
|
||||
cg.add(var.set_auto_start(False))
|
||||
|
||||
for injection in config[CONF_INJECTIONS]:
|
||||
rx_data = injection[CONF_INJECT_RX]
|
||||
delay_ms = injection[CONF_DELAY]
|
||||
@@ -146,7 +152,8 @@ async def to_code(config):
|
||||
for response in config[CONF_RESPONSES]:
|
||||
tx_data = response[CONF_EXPECT_TX]
|
||||
rx_data = response[CONF_INJECT_RX]
|
||||
cg.add(var.add_response(tx_data, rx_data))
|
||||
delay_ms = response[CONF_DELAY]
|
||||
cg.add(var.add_response(tx_data, rx_data, delay_ms))
|
||||
|
||||
for periodic in config[CONF_PERIODIC_RX]:
|
||||
data = periodic[CONF_DATA]
|
||||
|
||||
@@ -16,24 +16,28 @@ void MockUartComponent::setup() {
|
||||
}
|
||||
|
||||
void MockUartComponent::loop() {
|
||||
uint32_t now = App.get_loop_component_start_time();
|
||||
|
||||
// Initialize scenario start time on first loop() call, after all components have
|
||||
// finished setup(). This prevents injection delays from being consumed during setup.
|
||||
if (!this->loop_started_) {
|
||||
this->loop_started_ = true;
|
||||
this->scenario_start_ms_ = now;
|
||||
this->cumulative_delay_ms_ = 0;
|
||||
ESP_LOGD(TAG, "Scenario started at %u ms", now);
|
||||
if (this->auto_start_) {
|
||||
this->start_scenario();
|
||||
} else {
|
||||
ESP_LOGD(TAG, "Scenario waiting for manual start");
|
||||
}
|
||||
}
|
||||
|
||||
if (!this->scenario_active_) {
|
||||
return;
|
||||
}
|
||||
|
||||
uint32_t now = App.get_loop_component_start_time();
|
||||
|
||||
// Process at most ONE timed injection per loop iteration.
|
||||
// This ensures each injection is in a separate loop cycle, giving the consuming
|
||||
// component (e.g., LD2410) a chance to process each batch independently.
|
||||
if (this->injection_index_ < this->injections_.size()) {
|
||||
auto &injection = this->injections_[this->injection_index_];
|
||||
uint32_t target_time = this->scenario_start_ms_ + this->cumulative_delay_ms_ + injection.delay_ms;
|
||||
if (now >= target_time) {
|
||||
uint32_t total_delay = this->cumulative_delay_ms_ + injection.delay_ms;
|
||||
if (now - this->scenario_start_ms_ >= total_delay) {
|
||||
ESP_LOGD(TAG, "Injecting %zu RX bytes (injection %u)", injection.rx_data.size(), this->injection_index_);
|
||||
this->inject_to_rx_buffer(injection.rx_data);
|
||||
this->cumulative_delay_ms_ += injection.delay_ms;
|
||||
@@ -48,6 +52,28 @@ void MockUartComponent::loop() {
|
||||
periodic.last_inject_ms = now;
|
||||
}
|
||||
}
|
||||
|
||||
// Process delayed responses
|
||||
for (auto &response : this->responses_) {
|
||||
if (response.delay_ms > 0 && response.last_match_ms > 0 && now - response.last_match_ms >= response.delay_ms) {
|
||||
ESP_LOGD(TAG, "Injecting %zu RX bytes for delayed response", response.inject_rx.size());
|
||||
this->inject_to_rx_buffer(response.inject_rx);
|
||||
response.last_match_ms = 0; // Reset to prevent repeated injection
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
void MockUartComponent::start_scenario() {
|
||||
uint32_t now = App.get_loop_component_start_time();
|
||||
this->scenario_active_ = true;
|
||||
this->scenario_start_ms_ = now;
|
||||
this->cumulative_delay_ms_ = 0;
|
||||
this->injection_index_ = 0;
|
||||
this->tx_buffer_.clear();
|
||||
for (auto &periodic : this->periodic_rx_) {
|
||||
periodic.last_inject_ms = now;
|
||||
}
|
||||
ESP_LOGD(TAG, "Scenario started at %u ms", now);
|
||||
}
|
||||
|
||||
void MockUartComponent::dump_config() {
|
||||
@@ -78,10 +104,12 @@ void MockUartComponent::write_array(const uint8_t *data, size_t len) {
|
||||
}
|
||||
#endif
|
||||
|
||||
this->try_match_response_();
|
||||
if (this->scenario_active_) {
|
||||
this->try_match_response_();
|
||||
}
|
||||
|
||||
// This directly calls a tx_hook (lambda) as an alternative to the simpler match_response mechanism.
|
||||
if (this->tx_hook_) {
|
||||
if (this->tx_hook_ && this->scenario_active_) {
|
||||
std::vector<uint8_t> buf(data, data + len);
|
||||
this->tx_hook_(buf);
|
||||
}
|
||||
@@ -130,8 +158,9 @@ void MockUartComponent::add_injection(const std::vector<uint8_t> &rx_data, uint3
|
||||
this->injections_.push_back({rx_data, delay_ms});
|
||||
}
|
||||
|
||||
void MockUartComponent::add_response(const std::vector<uint8_t> &expect_tx, const std::vector<uint8_t> &inject_rx) {
|
||||
this->responses_.push_back({expect_tx, inject_rx});
|
||||
void MockUartComponent::add_response(const std::vector<uint8_t> &expect_tx, const std::vector<uint8_t> &inject_rx,
|
||||
uint32_t delay_ms) {
|
||||
this->responses_.push_back({expect_tx, inject_rx, delay_ms, 0});
|
||||
}
|
||||
|
||||
void MockUartComponent::add_periodic_rx(const std::vector<uint8_t> &data, uint32_t interval_ms) {
|
||||
@@ -147,7 +176,13 @@ void MockUartComponent::try_match_response_() {
|
||||
size_t offset = this->tx_buffer_.size() - response.expect_tx.size();
|
||||
if (std::equal(response.expect_tx.begin(), response.expect_tx.end(), this->tx_buffer_.begin() + offset)) {
|
||||
ESP_LOGD(TAG, "TX match found, injecting %zu RX bytes", response.inject_rx.size());
|
||||
this->inject_to_rx_buffer(response.inject_rx);
|
||||
if (response.delay_ms > 0) {
|
||||
ESP_LOGD(TAG, "Delaying response by %u ms", response.delay_ms);
|
||||
// Schedule the response injection as a future injection
|
||||
response.last_match_ms = App.get_loop_component_start_time();
|
||||
} else {
|
||||
this->inject_to_rx_buffer(response.inject_rx);
|
||||
}
|
||||
this->tx_buffer_.clear();
|
||||
return;
|
||||
}
|
||||
|
||||
@@ -34,9 +34,12 @@ class MockUartComponent : public uart::UARTComponent, public Component {
|
||||
|
||||
// Scenario configuration - called from generated code
|
||||
void add_injection(const std::vector<uint8_t> &rx_data, uint32_t delay_ms);
|
||||
void add_response(const std::vector<uint8_t> &expect_tx, const std::vector<uint8_t> &inject_rx);
|
||||
void add_response(const std::vector<uint8_t> &expect_tx, const std::vector<uint8_t> &inject_rx,
|
||||
uint32_t delay_ms = 0);
|
||||
void add_periodic_rx(const std::vector<uint8_t> &data, uint32_t interval_ms);
|
||||
|
||||
void start_scenario();
|
||||
void set_auto_start(bool auto_start) { this->auto_start_ = auto_start; }
|
||||
void set_tx_hook(std::function<void(const std::vector<uint8_t> &)> &&cb) { this->tx_hook_ = std::move(cb); }
|
||||
void inject_to_rx_buffer(const std::vector<uint8_t> &data);
|
||||
void inject_to_rx_buffer(const uint8_t *data, size_t len);
|
||||
@@ -55,11 +58,15 @@ class MockUartComponent : public uart::UARTComponent, public Component {
|
||||
uint32_t scenario_start_ms_{0};
|
||||
uint32_t cumulative_delay_ms_{0};
|
||||
bool loop_started_{false};
|
||||
bool auto_start_{true};
|
||||
bool scenario_active_{false};
|
||||
|
||||
// TX-triggered responses
|
||||
struct Response {
|
||||
std::vector<uint8_t> expect_tx;
|
||||
std::vector<uint8_t> inject_rx;
|
||||
uint32_t delay_ms;
|
||||
uint32_t last_match_ms{0};
|
||||
};
|
||||
std::vector<Response> responses_;
|
||||
std::vector<uint8_t> tx_buffer_;
|
||||
|
||||
@@ -20,6 +20,7 @@ uart:
|
||||
uart_mock:
|
||||
id: mock_uart
|
||||
baud_rate: 256000
|
||||
auto_start: false
|
||||
injections:
|
||||
# Phase 1 (t=100ms): Valid LD2410 normal mode data frame - happy path
|
||||
# The buffer is clean at this point, so this frame should parse correctly.
|
||||
@@ -143,3 +144,10 @@ binary_sensor:
|
||||
name: "Has Moving Target"
|
||||
has_still_target:
|
||||
name: "Has Still Target"
|
||||
|
||||
button:
|
||||
- platform: template
|
||||
name: "Start Scenario"
|
||||
id: start_scenario_btn
|
||||
on_press:
|
||||
- lambda: 'id(mock_uart).start_scenario();'
|
||||
|
||||
@@ -19,6 +19,7 @@ uart:
|
||||
uart_mock:
|
||||
id: mock_uart
|
||||
baud_rate: 256000
|
||||
auto_start: false
|
||||
injections:
|
||||
# Phase 1 (t=100ms): Valid LD2410 engineering mode data frame
|
||||
# Captured from a real Screek Human Presence Sensor 1U with LD2410 firmware 2.4.x
|
||||
@@ -154,3 +155,10 @@ binary_sensor:
|
||||
name: "Has Still Target"
|
||||
out_pin_presence_status:
|
||||
name: "Out Pin Presence"
|
||||
|
||||
button:
|
||||
- platform: template
|
||||
name: "Start Scenario"
|
||||
id: start_scenario_btn
|
||||
on_press:
|
||||
- lambda: 'id(mock_uart).start_scenario();'
|
||||
|
||||
@@ -0,0 +1,179 @@
|
||||
esphome:
|
||||
name: uart-mock-ld2412-test
|
||||
|
||||
host:
|
||||
api:
|
||||
logger:
|
||||
level: VERBOSE
|
||||
|
||||
external_components:
|
||||
- source:
|
||||
type: local
|
||||
path: EXTERNAL_COMPONENT_PATH
|
||||
|
||||
# Dummy uart entry to satisfy ld2412's DEPENDENCIES = ["uart"]
|
||||
# The actual UART bus used is the uart_mock component below
|
||||
uart:
|
||||
baud_rate: 115200
|
||||
port: /dev/null
|
||||
|
||||
uart_mock:
|
||||
id: mock_uart
|
||||
baud_rate: 256000
|
||||
auto_start: false
|
||||
injections:
|
||||
# Phase 1 (t=100ms): Valid LD2412 normal mode data frame - happy path
|
||||
# The buffer is clean at this point, so this frame should parse correctly.
|
||||
# Moving target: 100cm, energy 50
|
||||
# Still target: 120cm, energy 25
|
||||
# Target state: 0x03 (moving + still)
|
||||
# detection_distance = 100 (LD2412 computes from moving target when MOVE_BITMASK set)
|
||||
#
|
||||
# Frame layout (24 bytes):
|
||||
# [0-3] F4 F3 F2 F1 = data frame header
|
||||
# [4-5] 0D 00 = length 13
|
||||
# [6] 02 = data type (normal)
|
||||
# [7] AA = data header marker
|
||||
# [8] 03 = target states (moving+still)
|
||||
# [9-10] 64 00 = moving distance 100 (0x0064)
|
||||
# [11] 32 = moving energy 50
|
||||
# [12-13] 78 00 = still distance 120 (0x0078)
|
||||
# [14] 19 = still energy 25
|
||||
# [15-16] 64 00 = detect distance bytes (ignored by LD2412 code)
|
||||
# [17] 00 = padding
|
||||
# [18] 55 = data footer marker
|
||||
# [19] 00 = CRC/check
|
||||
# [20-23] F8 F7 F6 F5 = data frame footer
|
||||
- delay: 100ms
|
||||
inject_rx:
|
||||
[
|
||||
0xF4, 0xF3, 0xF2, 0xF1,
|
||||
0x0D, 0x00,
|
||||
0x02, 0xAA,
|
||||
0x03,
|
||||
0x64, 0x00,
|
||||
0x32,
|
||||
0x78, 0x00,
|
||||
0x19,
|
||||
0x64, 0x00,
|
||||
0x00,
|
||||
0x55, 0x00,
|
||||
0xF8, 0xF7, 0xF6, 0xF5,
|
||||
]
|
||||
|
||||
# Phase 2 (t=300ms): Garbage bytes
|
||||
# LD2412's parser rejects bytes that don't match the frame header at
|
||||
# position 0 (must start with F4 or FD), so buffer stays empty.
|
||||
- delay: 200ms
|
||||
inject_rx: [0xDE, 0xAD, 0xBE, 0xEF, 0x00, 0x11, 0x22]
|
||||
|
||||
# Phase 3 (t=400ms): Truncated frame (header + partial data, no footer)
|
||||
# Starts with valid data frame header so parser accepts it.
|
||||
# After this, buffer_pos_ = 8.
|
||||
- delay: 100ms
|
||||
inject_rx: [0xF4, 0xF3, 0xF2, 0xF1, 0x0D, 0x00, 0x02, 0xAA]
|
||||
|
||||
# Phase 4 (t=600ms): Overflow - inject 60 bytes of 0xFF (MAX_LINE_LENGTH=54)
|
||||
# Buffer has 8 bytes from phase 3 (garbage in phase 2 was rejected).
|
||||
# Overflow math: buffer_pos_ starts at 8, overflow triggers when
|
||||
# buffer_pos_ reaches 53 (MAX_LINE_LENGTH - 1). Need 45 more bytes to
|
||||
# fill positions 8-52, then byte 46 triggers overflow. After overflow,
|
||||
# buffer_pos_ = 0 and remaining 0xFF bytes are rejected (don't match header).
|
||||
- delay: 200ms
|
||||
inject_rx:
|
||||
[
|
||||
0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF,
|
||||
0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF,
|
||||
0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF,
|
||||
0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF,
|
||||
0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF,
|
||||
0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF,
|
||||
]
|
||||
|
||||
# Phase 5 (t=700ms): Valid frame after overflow - recovery test
|
||||
# Buffer was reset by overflow. This valid frame should parse correctly.
|
||||
# Moving target: 50cm, energy 100
|
||||
# Still target: 75cm, energy 80
|
||||
# detection_distance = 50 (moving target distance, since MOVE_BITMASK set)
|
||||
- delay: 100ms
|
||||
inject_rx:
|
||||
[
|
||||
0xF4, 0xF3, 0xF2, 0xF1,
|
||||
0x0D, 0x00,
|
||||
0x02, 0xAA,
|
||||
0x03,
|
||||
0x32, 0x00,
|
||||
0x64,
|
||||
0x4B, 0x00,
|
||||
0x50,
|
||||
0x32, 0x00,
|
||||
0x00,
|
||||
0x55, 0x00,
|
||||
0xF8, 0xF7, 0xF6, 0xF5,
|
||||
]
|
||||
|
||||
ld2412:
|
||||
id: ld2412_dev
|
||||
uart_id: mock_uart
|
||||
|
||||
sensor:
|
||||
- platform: ld2412
|
||||
ld2412_id: ld2412_dev
|
||||
moving_distance:
|
||||
name: "Moving Distance"
|
||||
filters:
|
||||
- timeout:
|
||||
timeout: 50ms
|
||||
value: last
|
||||
- throttle_with_priority: 50ms
|
||||
still_distance:
|
||||
name: "Still Distance"
|
||||
filters:
|
||||
- timeout:
|
||||
timeout: 50ms
|
||||
value: last
|
||||
- throttle_with_priority: 50ms
|
||||
moving_energy:
|
||||
name: "Moving Energy"
|
||||
filters:
|
||||
- timeout:
|
||||
timeout: 50ms
|
||||
value: last
|
||||
- throttle_with_priority: 50ms
|
||||
still_energy:
|
||||
name: "Still Energy"
|
||||
filters:
|
||||
- timeout:
|
||||
timeout: 50ms
|
||||
value: last
|
||||
- throttle_with_priority: 50ms
|
||||
detection_distance:
|
||||
name: "Detection Distance"
|
||||
filters:
|
||||
- timeout:
|
||||
timeout: 50ms
|
||||
value: last
|
||||
- throttle_with_priority: 50ms
|
||||
|
||||
binary_sensor:
|
||||
- platform: ld2412
|
||||
ld2412_id: ld2412_dev
|
||||
has_target:
|
||||
name: "Has Target"
|
||||
filters:
|
||||
- settle: 50ms
|
||||
has_moving_target:
|
||||
name: "Has Moving Target"
|
||||
filters:
|
||||
- settle: 50ms
|
||||
has_still_target:
|
||||
name: "Has Still Target"
|
||||
filters:
|
||||
- settle: 50ms
|
||||
|
||||
button:
|
||||
- platform: template
|
||||
name: "Start Scenario"
|
||||
id: start_scenario_btn
|
||||
on_press:
|
||||
- lambda: 'id(mock_uart).start_scenario();'
|
||||
@@ -0,0 +1,221 @@
|
||||
esphome:
|
||||
name: uart-mock-ld2412-eng-test
|
||||
|
||||
host:
|
||||
api:
|
||||
logger:
|
||||
level: VERBOSE
|
||||
|
||||
external_components:
|
||||
- source:
|
||||
type: local
|
||||
path: EXTERNAL_COMPONENT_PATH
|
||||
|
||||
# Dummy uart entry to satisfy ld2412's DEPENDENCIES = ["uart"]
|
||||
uart:
|
||||
baud_rate: 115200
|
||||
port: /dev/null
|
||||
|
||||
uart_mock:
|
||||
id: mock_uart
|
||||
baud_rate: 256000
|
||||
auto_start: false
|
||||
injections:
|
||||
# Phase 1 (t=100ms): Valid LD2412 engineering mode data frame
|
||||
#
|
||||
# Engineering mode frame layout (52 bytes):
|
||||
# [0-3] F4 F3 F2 F1 = data frame header
|
||||
# [4-5] 2A 00 = length 42
|
||||
# [6] 01 = data type (engineering mode)
|
||||
# [7] AA = data header marker
|
||||
# [8] 03 = target states (moving+still)
|
||||
# [9-10] 1E 00 = moving distance 30 (0x001E)
|
||||
# [11] 64 = moving energy 100
|
||||
# [12-13] 1E 00 = still distance 30 (0x001E)
|
||||
# [14] 64 = still energy 100
|
||||
# [15-16] 00 00 = detection distance bytes (ignored)
|
||||
# [17-30] gate moving energies (14 gates)
|
||||
# [31-44] gate still energies (14 gates)
|
||||
# [45] 57 = light sensor value 87
|
||||
# [46] 55 = data footer marker
|
||||
# [47] 00 = check
|
||||
# [48-51] F8 F7 F6 F5 = data frame footer
|
||||
- delay: 100ms
|
||||
inject_rx:
|
||||
[
|
||||
0xF4, 0xF3, 0xF2, 0xF1,
|
||||
0x2A, 0x00,
|
||||
0x01, 0xAA,
|
||||
0x03,
|
||||
0x1E, 0x00,
|
||||
0x64,
|
||||
0x1E, 0x00,
|
||||
0x64,
|
||||
0x00, 0x00,
|
||||
0x64, 0x41, 0x06, 0x0E, 0x2B, 0x16, 0x03, 0x03, 0x07, 0x05, 0x09, 0x08, 0x07, 0x06,
|
||||
0x00, 0x00, 0x64, 0x64, 0x64, 0x64, 0x64, 0x64, 0x64, 0x50, 0x40, 0x30, 0x20, 0x10,
|
||||
0x57,
|
||||
0x55, 0x00,
|
||||
0xF8, 0xF7, 0xF6, 0xF5,
|
||||
]
|
||||
|
||||
# Phase 2 (t=200ms): Second engineering mode frame with different values
|
||||
# Moving at 73cm, still at 30cm
|
||||
- delay: 100ms
|
||||
inject_rx:
|
||||
[
|
||||
0xF4, 0xF3, 0xF2, 0xF1,
|
||||
0x2A, 0x00,
|
||||
0x01, 0xAA,
|
||||
0x03,
|
||||
0x49, 0x00,
|
||||
0x64,
|
||||
0x1E, 0x00,
|
||||
0x64,
|
||||
0x21, 0x00,
|
||||
0x11, 0x64, 0x05, 0x29, 0x39, 0x10, 0x03, 0x11, 0x0E, 0x08, 0x06, 0x04, 0x03, 0x02,
|
||||
0x00, 0x00, 0x64, 0x64, 0x64, 0x64, 0x64, 0x64, 0x64, 0x50, 0x40, 0x30, 0x20, 0x10,
|
||||
0x57,
|
||||
0x55, 0x00,
|
||||
0xF8, 0xF7, 0xF6, 0xF5,
|
||||
]
|
||||
|
||||
# Phase 3 (t=300ms): Frame with still target at 291cm (multi-byte distance)
|
||||
# This tests encode_uint16 with high byte > 0
|
||||
# Target state: 0x02 (still only) -> detection_distance = still distance = 291
|
||||
- delay: 100ms
|
||||
inject_rx:
|
||||
[
|
||||
0xF4, 0xF3, 0xF2, 0xF1,
|
||||
0x2A, 0x00,
|
||||
0x01, 0xAA,
|
||||
0x02,
|
||||
0x2F, 0x00,
|
||||
0x36,
|
||||
0x23, 0x01,
|
||||
0x64,
|
||||
0x21, 0x00,
|
||||
0x2F, 0x36, 0x09, 0x0D, 0x15, 0x0B, 0x06, 0x06, 0x08, 0x09, 0x08, 0x07, 0x06, 0x05,
|
||||
0x00, 0x00, 0x64, 0x64, 0x64, 0x64, 0x64, 0x5A, 0x3D, 0x30, 0x20, 0x10, 0x08, 0x04,
|
||||
0x57,
|
||||
0x55, 0x00,
|
||||
0xF8, 0xF7, 0xF6, 0xF5,
|
||||
]
|
||||
|
||||
ld2412:
|
||||
id: ld2412_dev
|
||||
uart_id: mock_uart
|
||||
|
||||
sensor:
|
||||
- platform: ld2412
|
||||
ld2412_id: ld2412_dev
|
||||
moving_distance:
|
||||
name: "Moving Distance"
|
||||
filters:
|
||||
- timeout:
|
||||
timeout: 50ms
|
||||
value: last
|
||||
- throttle_with_priority: 50ms
|
||||
still_distance:
|
||||
name: "Still Distance"
|
||||
filters:
|
||||
- timeout:
|
||||
timeout: 50ms
|
||||
value: last
|
||||
- throttle_with_priority: 50ms
|
||||
moving_energy:
|
||||
name: "Moving Energy"
|
||||
filters:
|
||||
- timeout:
|
||||
timeout: 50ms
|
||||
value: last
|
||||
- throttle_with_priority: 50ms
|
||||
still_energy:
|
||||
name: "Still Energy"
|
||||
filters:
|
||||
- timeout:
|
||||
timeout: 50ms
|
||||
value: last
|
||||
- throttle_with_priority: 50ms
|
||||
detection_distance:
|
||||
name: "Detection Distance"
|
||||
filters:
|
||||
- timeout:
|
||||
timeout: 50ms
|
||||
value: last
|
||||
- throttle_with_priority: 50ms
|
||||
light:
|
||||
name: "Light"
|
||||
filters:
|
||||
- timeout:
|
||||
timeout: 50ms
|
||||
value: last
|
||||
- throttle_with_priority: 50ms
|
||||
gate_0:
|
||||
move_energy:
|
||||
name: "Gate 0 Move Energy"
|
||||
filters:
|
||||
- timeout:
|
||||
timeout: 50ms
|
||||
value: last
|
||||
- throttle_with_priority: 50ms
|
||||
still_energy:
|
||||
name: "Gate 0 Still Energy"
|
||||
filters:
|
||||
- timeout:
|
||||
timeout: 50ms
|
||||
value: last
|
||||
- throttle_with_priority: 50ms
|
||||
gate_1:
|
||||
move_energy:
|
||||
name: "Gate 1 Move Energy"
|
||||
filters:
|
||||
- timeout:
|
||||
timeout: 50ms
|
||||
value: last
|
||||
- throttle_with_priority: 50ms
|
||||
still_energy:
|
||||
name: "Gate 1 Still Energy"
|
||||
filters:
|
||||
- timeout:
|
||||
timeout: 50ms
|
||||
value: last
|
||||
- throttle_with_priority: 50ms
|
||||
gate_2:
|
||||
move_energy:
|
||||
name: "Gate 2 Move Energy"
|
||||
filters:
|
||||
- timeout:
|
||||
timeout: 50ms
|
||||
value: last
|
||||
- throttle_with_priority: 50ms
|
||||
still_energy:
|
||||
name: "Gate 2 Still Energy"
|
||||
filters:
|
||||
- timeout:
|
||||
timeout: 50ms
|
||||
value: last
|
||||
- throttle_with_priority: 50ms
|
||||
|
||||
binary_sensor:
|
||||
- platform: ld2412
|
||||
ld2412_id: ld2412_dev
|
||||
has_target:
|
||||
name: "Has Target"
|
||||
filters:
|
||||
- settle: 50ms
|
||||
has_moving_target:
|
||||
name: "Has Moving Target"
|
||||
filters:
|
||||
- settle: 50ms
|
||||
has_still_target:
|
||||
name: "Has Still Target"
|
||||
filters:
|
||||
- settle: 50ms
|
||||
|
||||
button:
|
||||
- platform: template
|
||||
name: "Start Scenario"
|
||||
id: start_scenario_btn
|
||||
on_press:
|
||||
- lambda: 'id(mock_uart).start_scenario();'
|
||||
@@ -22,19 +22,71 @@ uart_mock:
|
||||
baud_rate: 9600
|
||||
rx_full_threshold: 120
|
||||
rx_timeout: 2
|
||||
auto_start: false
|
||||
debug:
|
||||
responses:
|
||||
- expect_tx: [0x01, 0x03, 0x00, 0x03, 0x00, 0x01, 0x74, 0x0A] # Read holding register 1 on device 1
|
||||
- expect_tx: [0x01, 0x03, 0x00, 0x03, 0x00, 0x01, 0x74, 0x0A] # Read holding register 3 on device 1 (basic_register)
|
||||
inject_rx: [0x01, 0x03, 0x02, 0x01, 0x03, 0xF9, 0xD5] # Return value 0x0103 (hex) = 259 (dec)
|
||||
- expect_tx: [0x01, 0x03, 0x00, 0x05, 0x00, 0x01, 0x94, 0x0B] # Read holding register 5 on device 1 (delayed_response)
|
||||
delay: 100ms # Shorter than modbus send_wait_time of 200ms, should succeed
|
||||
inject_rx: [0x01, 0x03, 0x02, 0x00, 0xFF, 0xF8, 0x04] # Return value 0x00FF (hex) = 255 (dec)
|
||||
- expect_tx: [0x02, 0x03, 0x00, 0x07, 0x00, 0x01, 0x35, 0xF8] # Read holding register 7 on device 2 (late_response)
|
||||
delay: 300ms # Longer than modbus send_wait_time of 200ms, should cause timeout
|
||||
inject_rx: [0x02, 0x03, 0x02, 0x00, 0xF0, 0xFC, 0x00] # Return value 0x00F0 (hex) = 240 (dec)
|
||||
- expect_tx: [0x03, 0x03, 0x00, 0x09, 0x00, 0x01, 0x55, 0xEA] # Read holding register 9 on device 3 (no_response)
|
||||
inject_rx: [] # No response, should cause timeout
|
||||
- expect_tx: [0x01, 0x03, 0x00, 0x0A, 0x00, 0x01, 0xA4, 0x08] # Read holding register A on device 1 (exception_response)
|
||||
inject_rx: [0x01, 0x83, 0x02, 0xC0, 0xF1] # Exception response with code 2 (illegal data address)
|
||||
|
||||
modbus:
|
||||
uart_id: virtual_uart_dev
|
||||
send_wait_time: 200ms
|
||||
turnaround_time: 10ms
|
||||
|
||||
modbus_controller:
|
||||
address: 1
|
||||
- address: 1
|
||||
id: modbus_controller_ok
|
||||
max_cmd_retries: 0
|
||||
update_interval: 1s
|
||||
- address: 2
|
||||
id: modbus_controller_slow
|
||||
max_cmd_retries: 0
|
||||
update_interval: 1s
|
||||
- address: 3
|
||||
id: modbus_controller_offline
|
||||
max_cmd_retries: 0
|
||||
update_interval: 1s
|
||||
|
||||
sensor:
|
||||
- platform: modbus_controller
|
||||
name: "basic_register"
|
||||
address: 0x03
|
||||
register_type: holding
|
||||
modbus_controller_id: modbus_controller_ok
|
||||
- platform: modbus_controller
|
||||
name: "delayed_response"
|
||||
address: 0x05
|
||||
register_type: holding
|
||||
modbus_controller_id: modbus_controller_ok
|
||||
- platform: modbus_controller
|
||||
name: "late_response"
|
||||
address: 0x07
|
||||
register_type: holding
|
||||
modbus_controller_id: modbus_controller_slow
|
||||
- platform: modbus_controller
|
||||
name: "no_response"
|
||||
address: 0x09
|
||||
register_type: holding
|
||||
modbus_controller_id: modbus_controller_offline
|
||||
- platform: modbus_controller
|
||||
name: "exception_response"
|
||||
address: 0x0A
|
||||
register_type: holding
|
||||
modbus_controller_id: modbus_controller_ok
|
||||
|
||||
button:
|
||||
- platform: template
|
||||
name: "Start Scenario"
|
||||
id: start_scenario_btn
|
||||
on_press:
|
||||
- lambda: 'id(virtual_uart_dev).start_scenario();'
|
||||
|
||||
@@ -22,6 +22,7 @@ uart_mock:
|
||||
baud_rate: 9600
|
||||
rx_full_threshold: 120
|
||||
rx_timeout: 2
|
||||
auto_start: false
|
||||
debug:
|
||||
on_tx:
|
||||
- then:
|
||||
@@ -45,10 +46,19 @@ uart_mock:
|
||||
|
||||
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();'
|
||||
|
||||
@@ -3,10 +3,17 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
from collections.abc import Callable
|
||||
import logging
|
||||
from typing import TypeVar
|
||||
|
||||
from aioesphomeapi import ButtonInfo, EntityInfo, EntityState
|
||||
from aioesphomeapi import (
|
||||
BinarySensorState,
|
||||
ButtonInfo,
|
||||
EntityInfo,
|
||||
EntityState,
|
||||
SensorState,
|
||||
)
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
@@ -234,3 +241,90 @@ class InitialStateHelper:
|
||||
asyncio.TimeoutError: If initial states aren't received within timeout
|
||||
"""
|
||||
await asyncio.wait_for(self._initial_states_received, timeout=timeout)
|
||||
|
||||
|
||||
class SensorStateCollector:
|
||||
"""Collects sensor and binary sensor state updates and provides wait helpers.
|
||||
|
||||
Usage:
|
||||
collector = SensorStateCollector(
|
||||
sensor_names=["moving_distance", "still_distance"],
|
||||
binary_sensor_names=["has_target"],
|
||||
)
|
||||
# Use collector.on_state as the callback (or wrap it)
|
||||
client.subscribe_states(helper.on_state_wrapper(collector.on_state))
|
||||
|
||||
# Wait for all sensors to have at least one value
|
||||
await collector.wait_for_all(timeout=3.0)
|
||||
|
||||
# Access collected states
|
||||
assert collector.sensor_states["moving_distance"][0] == approx(100.0)
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
sensor_names: list[str],
|
||||
binary_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._key_to_sensor: dict[int, str] = {}
|
||||
self._waiters: list[tuple[Callable[[], bool], asyncio.Future[bool]]] = []
|
||||
|
||||
if entities is not None:
|
||||
self.build_key_mapping(entities)
|
||||
|
||||
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.sort(key=len, reverse=True)
|
||||
self._key_to_sensor = build_key_to_entity_mapping(entities, all_names)
|
||||
|
||||
def on_state(self, state: EntityState) -> None:
|
||||
"""Process a state update."""
|
||||
if isinstance(state, SensorState) and not state.missing_state:
|
||||
sensor_name = self._key_to_sensor.get(state.key)
|
||||
if sensor_name and sensor_name in self.sensor_states:
|
||||
self.sensor_states[sensor_name].append(state.state)
|
||||
self._check_waiters()
|
||||
elif isinstance(state, BinarySensorState):
|
||||
sensor_name = self._key_to_sensor.get(state.key)
|
||||
if sensor_name and sensor_name in self.binary_states:
|
||||
self.binary_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."""
|
||||
for condition, future in self._waiters:
|
||||
if not future.done() and condition():
|
||||
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()
|
||||
)
|
||||
|
||||
async def wait_for_all(self, timeout: float = 3.0) -> None:
|
||||
"""Wait until all sensors and binary sensors have at least one value."""
|
||||
if self._all_have_values():
|
||||
return
|
||||
future: asyncio.Future[bool] = asyncio.get_running_loop().create_future()
|
||||
self._waiters.append((self._all_have_values, future))
|
||||
await asyncio.wait_for(future, timeout=timeout)
|
||||
|
||||
def add_waiter(self, condition: Callable[[], bool]) -> asyncio.Future[bool]:
|
||||
"""Add a custom waiter that resolves when condition returns True.
|
||||
|
||||
Returns:
|
||||
A future that resolves when the condition is met.
|
||||
"""
|
||||
future: asyncio.Future[bool] = asyncio.get_running_loop().create_future()
|
||||
if condition():
|
||||
future.set_result(True)
|
||||
else:
|
||||
self._waiters.append((condition, future))
|
||||
return future
|
||||
|
||||
@@ -21,16 +21,10 @@ from __future__ import annotations
|
||||
import asyncio
|
||||
from pathlib import Path
|
||||
|
||||
from aioesphomeapi import (
|
||||
BinarySensorInfo,
|
||||
BinarySensorState,
|
||||
EntityState,
|
||||
SensorInfo,
|
||||
SensorState,
|
||||
)
|
||||
from aioesphomeapi import ButtonInfo
|
||||
import pytest
|
||||
|
||||
from .state_utils import InitialStateHelper, build_key_to_entity_mapping, find_entity
|
||||
from .state_utils import InitialStateHelper, SensorStateCollector, find_entity
|
||||
from .types import APIClientConnectedFactory, RunCompiledFunction
|
||||
|
||||
|
||||
@@ -64,100 +58,65 @@ async def test_uart_mock_ld2410(
|
||||
if "uart_mock" in line and "TX " in line:
|
||||
tx_log_lines.append(line)
|
||||
|
||||
# Track sensor state updates (after initial state is swallowed)
|
||||
sensor_states: dict[str, list[float]] = {
|
||||
"moving_distance": [],
|
||||
"still_distance": [],
|
||||
"moving_energy": [],
|
||||
"still_energy": [],
|
||||
"detection_distance": [],
|
||||
}
|
||||
binary_states: dict[str, list[bool]] = {
|
||||
"has_target": [],
|
||||
"has_moving_target": [],
|
||||
"has_still_target": [],
|
||||
}
|
||||
collector = SensorStateCollector(
|
||||
sensor_names=[
|
||||
"moving_distance",
|
||||
"still_distance",
|
||||
"moving_energy",
|
||||
"still_energy",
|
||||
"detection_distance",
|
||||
],
|
||||
binary_sensor_names=[
|
||||
"has_target",
|
||||
"has_moving_target",
|
||||
"has_still_target",
|
||||
],
|
||||
)
|
||||
|
||||
# Signal when we see recovery frame values
|
||||
recovery_received = 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 the recovery frame (moving_distance = 50)
|
||||
if (
|
||||
sensor_name == "moving_distance"
|
||||
and state.state == pytest.approx(50.0)
|
||||
and not recovery_received.done()
|
||||
):
|
||||
recovery_received.set_result(True)
|
||||
elif isinstance(state, BinarySensorState):
|
||||
sensor_name = key_to_sensor.get(state.key)
|
||||
if sensor_name and sensor_name in binary_states:
|
||||
binary_states[sensor_name].append(state.state)
|
||||
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()
|
||||
|
||||
# Build key mappings for all sensor types
|
||||
all_names = list(sensor_states.keys()) + list(binary_states.keys())
|
||||
key_to_sensor = build_key_to_entity_mapping(entities, all_names)
|
||||
collector.build_key_mapping(entities)
|
||||
|
||||
# Set up initial state helper
|
||||
initial_state_helper = InitialStateHelper(entities)
|
||||
client.subscribe_states(initial_state_helper.on_state_wrapper(on_state))
|
||||
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")
|
||||
|
||||
# Phase 1 values are in the initial states (swallowed by InitialStateHelper).
|
||||
# Verify them via initial_states dict.
|
||||
moving_dist_entity = find_entity(entities, "moving_distance", SensorInfo)
|
||||
assert moving_dist_entity is not None
|
||||
initial_moving = initial_state_helper.initial_states.get(moving_dist_entity.key)
|
||||
assert initial_moving is not None and isinstance(initial_moving, SensorState)
|
||||
assert initial_moving.state == pytest.approx(100.0), (
|
||||
f"Initial moving distance should be 100, got {initial_moving.state}"
|
||||
)
|
||||
# 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)
|
||||
|
||||
still_dist_entity = find_entity(entities, "still_distance", SensorInfo)
|
||||
assert still_dist_entity is not None
|
||||
initial_still = initial_state_helper.initial_states.get(still_dist_entity.key)
|
||||
assert initial_still is not None and isinstance(initial_still, SensorState)
|
||||
assert initial_still.state == pytest.approx(120.0), (
|
||||
f"Initial still distance should be 120, got {initial_still.state}"
|
||||
)
|
||||
# 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}"
|
||||
)
|
||||
|
||||
moving_energy_entity = find_entity(entities, "moving_energy", SensorInfo)
|
||||
assert moving_energy_entity is not None
|
||||
initial_me = initial_state_helper.initial_states.get(moving_energy_entity.key)
|
||||
assert initial_me is not None and isinstance(initial_me, SensorState)
|
||||
assert initial_me.state == pytest.approx(50.0), (
|
||||
f"Initial moving energy should be 50, got {initial_me.state}"
|
||||
)
|
||||
|
||||
still_energy_entity = find_entity(entities, "still_energy", SensorInfo)
|
||||
assert still_energy_entity is not None
|
||||
initial_se = initial_state_helper.initial_states.get(still_energy_entity.key)
|
||||
assert initial_se is not None and isinstance(initial_se, SensorState)
|
||||
assert initial_se.state == pytest.approx(25.0), (
|
||||
f"Initial still energy should be 25, got {initial_se.state}"
|
||||
)
|
||||
|
||||
detect_dist_entity = find_entity(entities, "detection_distance", SensorInfo)
|
||||
assert detect_dist_entity is not None
|
||||
initial_dd = initial_state_helper.initial_states.get(detect_dist_entity.key)
|
||||
assert initial_dd is not None and isinstance(initial_dd, SensorState)
|
||||
assert initial_dd.state == pytest.approx(300.0), (
|
||||
f"Initial detection distance should be 300, got {initial_dd.state}"
|
||||
)
|
||||
# Phase 1 values: moving=100, still=120, energy=50/25, detect=300
|
||||
assert collector.sensor_states["moving_distance"][0] == pytest.approx(100.0)
|
||||
assert collector.sensor_states["still_distance"][0] == pytest.approx(120.0)
|
||||
assert collector.sensor_states["moving_energy"][0] == pytest.approx(50.0)
|
||||
assert collector.sensor_states["still_energy"][0] == pytest.approx(25.0)
|
||||
assert collector.sensor_states["detection_distance"][0] == pytest.approx(300.0)
|
||||
|
||||
# Wait for the recovery frame (Phase 5) to be parsed
|
||||
# This proves the component survived garbage + truncated + overflow
|
||||
@@ -165,12 +124,8 @@ async def test_uart_mock_ld2410(
|
||||
await asyncio.wait_for(recovery_received, timeout=15.0)
|
||||
except TimeoutError:
|
||||
pytest.fail(
|
||||
f"Timeout waiting for recovery frame. Received sensor states:\n"
|
||||
f" moving_distance: {sensor_states['moving_distance']}\n"
|
||||
f" still_distance: {sensor_states['still_distance']}\n"
|
||||
f" moving_energy: {sensor_states['moving_energy']}\n"
|
||||
f" still_energy: {sensor_states['still_energy']}\n"
|
||||
f" detection_distance: {sensor_states['detection_distance']}"
|
||||
f"Timeout waiting for recovery frame. Received:\n"
|
||||
f" sensor_states: {collector.sensor_states}"
|
||||
)
|
||||
|
||||
# Verify overflow warning was logged
|
||||
@@ -183,67 +138,36 @@ async def test_uart_mock_ld2410(
|
||||
# A5 (MAC), AB (distance res), AE (light), 61 (params), FE (config off)
|
||||
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 LD2410 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 LD2410 command frame footer 04:03:02:01 in TX log"
|
||||
)
|
||||
|
||||
# Recovery frame values (Phase 5, after overflow)
|
||||
assert len(sensor_states["moving_distance"]) >= 1, (
|
||||
f"Expected recovery moving_distance, got: {sensor_states['moving_distance']}"
|
||||
)
|
||||
# Find the recovery value (moving_distance = 50)
|
||||
recovery_values = [
|
||||
v for v in sensor_states["moving_distance"] if v == pytest.approx(50.0)
|
||||
]
|
||||
assert len(recovery_values) >= 1, (
|
||||
f"Expected moving_distance=50 in recovery, got: {sensor_states['moving_distance']}"
|
||||
)
|
||||
|
||||
# Recovery frame: moving=50, still=75, energy=100/80, detect=127
|
||||
recovery_idx = next(
|
||||
i
|
||||
for i, v in enumerate(sensor_states["moving_distance"])
|
||||
for i, v in enumerate(collector.sensor_states["moving_distance"])
|
||||
if v == pytest.approx(50.0)
|
||||
)
|
||||
assert sensor_states["still_distance"][recovery_idx] == pytest.approx(75.0), (
|
||||
f"Recovery still distance should be 75, got {sensor_states['still_distance'][recovery_idx]}"
|
||||
assert collector.sensor_states["still_distance"][recovery_idx] == pytest.approx(
|
||||
75.0
|
||||
)
|
||||
assert sensor_states["moving_energy"][recovery_idx] == pytest.approx(100.0), (
|
||||
f"Recovery moving energy should be 100, got {sensor_states['moving_energy'][recovery_idx]}"
|
||||
assert collector.sensor_states["moving_energy"][recovery_idx] == pytest.approx(
|
||||
100.0
|
||||
)
|
||||
assert sensor_states["still_energy"][recovery_idx] == pytest.approx(80.0), (
|
||||
f"Recovery still energy should be 80, got {sensor_states['still_energy'][recovery_idx]}"
|
||||
)
|
||||
assert sensor_states["detection_distance"][recovery_idx] == pytest.approx(
|
||||
127.0
|
||||
), (
|
||||
f"Recovery detection distance should be 127, got {sensor_states['detection_distance'][recovery_idx]}"
|
||||
assert collector.sensor_states["still_energy"][recovery_idx] == pytest.approx(
|
||||
80.0
|
||||
)
|
||||
assert collector.sensor_states["detection_distance"][
|
||||
recovery_idx
|
||||
] == pytest.approx(127.0)
|
||||
|
||||
# Verify binary sensors detected targets
|
||||
# Binary sensors could be in initial states or forwarded states
|
||||
has_target_entity = find_entity(entities, "has_target", BinarySensorInfo)
|
||||
assert has_target_entity is not None
|
||||
initial_ht = initial_state_helper.initial_states.get(has_target_entity.key)
|
||||
assert initial_ht is not None and isinstance(initial_ht, BinarySensorState)
|
||||
assert initial_ht.state is True, "Has target should be True"
|
||||
|
||||
has_moving_entity = find_entity(entities, "has_moving_target", BinarySensorInfo)
|
||||
assert has_moving_entity is not None
|
||||
initial_hm = initial_state_helper.initial_states.get(has_moving_entity.key)
|
||||
assert initial_hm is not None and isinstance(initial_hm, BinarySensorState)
|
||||
assert initial_hm.state is True, "Has moving target should be True"
|
||||
|
||||
has_still_entity = find_entity(entities, "has_still_target", BinarySensorInfo)
|
||||
assert has_still_entity is not None
|
||||
initial_hs = initial_state_helper.initial_states.get(has_still_entity.key)
|
||||
assert initial_hs is not None and isinstance(initial_hs, BinarySensorState)
|
||||
assert initial_hs.state is True, "Has still target should be True"
|
||||
# Verify binary sensors detected targets (from Phase 1 frame)
|
||||
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
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@@ -260,133 +184,82 @@ async def test_uart_mock_ld2410_engineering(
|
||||
"EXTERNAL_COMPONENT_PATH", external_components_path
|
||||
)
|
||||
|
||||
loop = asyncio.get_running_loop()
|
||||
collector = SensorStateCollector(
|
||||
sensor_names=[
|
||||
"moving_distance",
|
||||
"still_distance",
|
||||
"moving_energy",
|
||||
"still_energy",
|
||||
"detection_distance",
|
||||
"light",
|
||||
"gate_0_move_energy",
|
||||
"gate_1_move_energy",
|
||||
"gate_2_move_energy",
|
||||
"gate_0_still_energy",
|
||||
"gate_1_still_energy",
|
||||
"gate_2_still_energy",
|
||||
],
|
||||
binary_sensor_names=[
|
||||
"has_target",
|
||||
"has_moving_target",
|
||||
"has_still_target",
|
||||
"out_pin_presence",
|
||||
],
|
||||
)
|
||||
|
||||
# Track sensor state updates (after initial state is swallowed)
|
||||
sensor_states: dict[str, list[float]] = {
|
||||
"moving_distance": [],
|
||||
"still_distance": [],
|
||||
"moving_energy": [],
|
||||
"still_energy": [],
|
||||
"detection_distance": [],
|
||||
"light": [],
|
||||
"gate_0_move_energy": [],
|
||||
"gate_1_move_energy": [],
|
||||
"gate_2_move_energy": [],
|
||||
"gate_0_still_energy": [],
|
||||
"gate_1_still_energy": [],
|
||||
"gate_2_still_energy": [],
|
||||
}
|
||||
binary_states: dict[str, list[bool]] = {
|
||||
"has_target": [],
|
||||
"has_moving_target": [],
|
||||
"has_still_target": [],
|
||||
"out_pin_presence": [],
|
||||
}
|
||||
|
||||
# Signal when we see Phase 3 frame (still_distance = 291)
|
||||
phase3_received = 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)
|
||||
if (
|
||||
sensor_name == "still_distance"
|
||||
and state.state == pytest.approx(291.0)
|
||||
and not phase3_received.done()
|
||||
):
|
||||
phase3_received.set_result(True)
|
||||
elif isinstance(state, BinarySensorState):
|
||||
sensor_name = key_to_sensor.get(state.key)
|
||||
if sensor_name and sensor_name in binary_states:
|
||||
binary_states[sensor_name].append(state.state)
|
||||
# Signal when we see Phase 3 frame values
|
||||
phase3_received = collector.add_waiter(
|
||||
lambda: pytest.approx(291.0) in collector.sensor_states["still_distance"]
|
||||
)
|
||||
|
||||
async with (
|
||||
run_compiled(yaml_config),
|
||||
api_client_connected() as client,
|
||||
):
|
||||
entities, _ = await client.list_entities_services()
|
||||
|
||||
all_names = list(sensor_states.keys()) + list(binary_states.keys())
|
||||
key_to_sensor = build_key_to_entity_mapping(entities, all_names)
|
||||
collector.build_key_mapping(entities)
|
||||
|
||||
initial_state_helper = InitialStateHelper(entities)
|
||||
client.subscribe_states(initial_state_helper.on_state_wrapper(on_state))
|
||||
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")
|
||||
|
||||
# Phase 1 initial values (engineering mode frame):
|
||||
# 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 (engineering mode frame):
|
||||
# moving=30, energy=100, still=30, energy=100, detect=0
|
||||
moving_dist_entity = find_entity(entities, "moving_distance", SensorInfo)
|
||||
assert moving_dist_entity is not None
|
||||
initial_moving = initial_state_helper.initial_states.get(moving_dist_entity.key)
|
||||
assert initial_moving is not None and isinstance(initial_moving, SensorState)
|
||||
assert initial_moving.state == pytest.approx(30.0), (
|
||||
f"Initial moving distance should be 30, got {initial_moving.state}"
|
||||
)
|
||||
|
||||
still_dist_entity = find_entity(entities, "still_distance", SensorInfo)
|
||||
assert still_dist_entity is not None
|
||||
initial_still = initial_state_helper.initial_states.get(still_dist_entity.key)
|
||||
assert initial_still is not None and isinstance(initial_still, SensorState)
|
||||
assert initial_still.state == pytest.approx(30.0), (
|
||||
f"Initial still distance should be 30, got {initial_still.state}"
|
||||
)
|
||||
|
||||
# Verify engineering mode sensors from initial state
|
||||
# Gate 0 moving energy = 0x64 = 100
|
||||
gate0_move_entity = find_entity(entities, "gate_0_move_energy", SensorInfo)
|
||||
assert gate0_move_entity is not None
|
||||
initial_g0m = initial_state_helper.initial_states.get(gate0_move_entity.key)
|
||||
assert initial_g0m is not None and isinstance(initial_g0m, SensorState)
|
||||
assert initial_g0m.state == pytest.approx(100.0), (
|
||||
f"Gate 0 move energy should be 100, got {initial_g0m.state}"
|
||||
)
|
||||
|
||||
# Gate 1 moving energy = 0x41 = 65
|
||||
gate1_move_entity = find_entity(entities, "gate_1_move_energy", SensorInfo)
|
||||
assert gate1_move_entity is not None
|
||||
initial_g1m = initial_state_helper.initial_states.get(gate1_move_entity.key)
|
||||
assert initial_g1m is not None and isinstance(initial_g1m, SensorState)
|
||||
assert initial_g1m.state == pytest.approx(65.0), (
|
||||
f"Gate 1 move energy should be 65, got {initial_g1m.state}"
|
||||
)
|
||||
|
||||
# Light sensor = 0x57 = 87
|
||||
light_entity = find_entity(entities, "light", SensorInfo)
|
||||
assert light_entity is not None
|
||||
initial_light = initial_state_helper.initial_states.get(light_entity.key)
|
||||
assert initial_light is not None and isinstance(initial_light, SensorState)
|
||||
assert initial_light.state == pytest.approx(87.0), (
|
||||
f"Light sensor should be 87, got {initial_light.state}"
|
||||
)
|
||||
|
||||
# Out pin presence = 0x01 = True
|
||||
out_pin_entity = find_entity(entities, "out_pin_presence", BinarySensorInfo)
|
||||
assert out_pin_entity is not None
|
||||
initial_out = initial_state_helper.initial_states.get(out_pin_entity.key)
|
||||
assert initial_out is not None and isinstance(initial_out, BinarySensorState)
|
||||
assert initial_out.state is True, "Out pin presence should be True"
|
||||
assert collector.sensor_states["moving_distance"][0] == pytest.approx(30.0)
|
||||
assert collector.sensor_states["still_distance"][0] == pytest.approx(30.0)
|
||||
assert collector.sensor_states["gate_0_move_energy"][0] == pytest.approx(100.0)
|
||||
assert collector.sensor_states["gate_1_move_energy"][0] == pytest.approx(65.0)
|
||||
assert collector.sensor_states["light"][0] == pytest.approx(87.0)
|
||||
assert collector.binary_states["out_pin_presence"][0] is True
|
||||
|
||||
# Wait for Phase 3 frame (still_distance = 291cm, multi-byte)
|
||||
try:
|
||||
await asyncio.wait_for(phase3_received, timeout=15.0)
|
||||
except TimeoutError:
|
||||
pytest.fail(
|
||||
f"Timeout waiting for Phase 3 frame. Received sensor states:\n"
|
||||
f" still_distance: {sensor_states['still_distance']}\n"
|
||||
f" moving_distance: {sensor_states['moving_distance']}"
|
||||
f"Timeout waiting for Phase 3 frame. Received:\n"
|
||||
f" still_distance: {collector.sensor_states['still_distance']}"
|
||||
)
|
||||
|
||||
# Phase 3: still distance = 0x0123 = 291cm (multi-byte distance test)
|
||||
phase3_still = [
|
||||
v for v in sensor_states["still_distance"] if v == pytest.approx(291.0)
|
||||
]
|
||||
assert len(phase3_still) >= 1, (
|
||||
f"Expected still_distance=291, got: {sensor_states['still_distance']}"
|
||||
)
|
||||
assert pytest.approx(291.0) in collector.sensor_states["still_distance"]
|
||||
|
||||
@@ -0,0 +1,275 @@
|
||||
"""Integration test for LD2412 component with mock UART.
|
||||
|
||||
Tests:
|
||||
test_uart_mock_ld2412 (normal mode):
|
||||
1. Happy path - valid data frame publishes correct sensor values
|
||||
2. Garbage resilience - random bytes don't crash the component
|
||||
3. Truncated frame handling - partial frame doesn't corrupt state
|
||||
4. Buffer overflow recovery - overflow resets the parser
|
||||
5. Post-overflow parsing - next valid frame after overflow is parsed correctly
|
||||
6. TX logging - verifies LD2412 sends expected setup commands
|
||||
|
||||
test_uart_mock_ld2412_engineering (engineering mode):
|
||||
1. Engineering mode frames with per-gate energy data and light sensor
|
||||
2. Multi-byte still distance (291cm) using high byte > 0
|
||||
3. Gate energy sensor values
|
||||
4. Detection distance computed from target state
|
||||
"""
|
||||
|
||||
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_ld2412(
|
||||
yaml_config: str,
|
||||
run_compiled: RunCompiledFunction,
|
||||
api_client_connected: APIClientConnectedFactory,
|
||||
) -> None:
|
||||
"""Test LD2412 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=[
|
||||
"moving_distance",
|
||||
"still_distance",
|
||||
"moving_energy",
|
||||
"still_energy",
|
||||
"detection_distance",
|
||||
],
|
||||
binary_sensor_names=[
|
||||
"has_target",
|
||||
"has_moving_target",
|
||||
"has_still_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, still=120, energy=50/25, detect=100
|
||||
assert collector.sensor_states["moving_distance"][0] == pytest.approx(100.0)
|
||||
assert collector.sensor_states["still_distance"][0] == pytest.approx(120.0)
|
||||
assert collector.sensor_states["moving_energy"][0] == pytest.approx(50.0)
|
||||
assert collector.sensor_states["still_energy"][0] == pytest.approx(25.0)
|
||||
assert collector.sensor_states["detection_distance"][0] == pytest.approx(100.0)
|
||||
|
||||
# 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=3.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 LD2412 sent setup commands (TX logging)
|
||||
assert len(tx_log_lines) > 0, "Expected TX log lines from uart_mock"
|
||||
tx_data = " ".join(tx_log_lines)
|
||||
assert "FD:FC:FB:FA" in tx_data, (
|
||||
"Expected LD2412 command frame header FD:FC:FB:FA in TX log"
|
||||
)
|
||||
assert "04:03:02:01" in tx_data, (
|
||||
"Expected LD2412 command frame footer 04:03:02:01 in TX log"
|
||||
)
|
||||
|
||||
# Recovery frame: moving=50, still=75, energy=100/80, detect=50
|
||||
recovery_idx = next(
|
||||
i
|
||||
for i, v in enumerate(collector.sensor_states["moving_distance"])
|
||||
if v == pytest.approx(50.0)
|
||||
)
|
||||
assert collector.sensor_states["still_distance"][recovery_idx] == pytest.approx(
|
||||
75.0
|
||||
)
|
||||
assert collector.sensor_states["moving_energy"][recovery_idx] == pytest.approx(
|
||||
100.0
|
||||
)
|
||||
assert collector.sensor_states["still_energy"][recovery_idx] == pytest.approx(
|
||||
80.0
|
||||
)
|
||||
assert collector.sensor_states["detection_distance"][
|
||||
recovery_idx
|
||||
] == pytest.approx(50.0)
|
||||
|
||||
# Verify binary sensors detected targets (from Phase 1 frame)
|
||||
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
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_uart_mock_ld2412_engineering(
|
||||
yaml_config: str,
|
||||
run_compiled: RunCompiledFunction,
|
||||
api_client_connected: APIClientConnectedFactory,
|
||||
) -> None:
|
||||
"""Test LD2412 engineering mode with per-gate energy, light, and multi-byte distance."""
|
||||
external_components_path = str(
|
||||
Path(__file__).parent / "fixtures" / "external_components"
|
||||
)
|
||||
yaml_config = yaml_config.replace(
|
||||
"EXTERNAL_COMPONENT_PATH", external_components_path
|
||||
)
|
||||
|
||||
collector = SensorStateCollector(
|
||||
sensor_names=[
|
||||
"moving_distance",
|
||||
"still_distance",
|
||||
"moving_energy",
|
||||
"still_energy",
|
||||
"detection_distance",
|
||||
"light",
|
||||
"gate_0_move_energy",
|
||||
"gate_1_move_energy",
|
||||
"gate_2_move_energy",
|
||||
"gate_0_still_energy",
|
||||
"gate_1_still_energy",
|
||||
"gate_2_still_energy",
|
||||
],
|
||||
binary_sensor_names=[
|
||||
"has_target",
|
||||
"has_moving_target",
|
||||
"has_still_target",
|
||||
],
|
||||
)
|
||||
|
||||
# Signal when we see Phase 3 frame values
|
||||
phase3_still_received = collector.add_waiter(
|
||||
lambda: pytest.approx(291.0) in collector.sensor_states["still_distance"]
|
||||
)
|
||||
phase3_detect_received = collector.add_waiter(
|
||||
lambda: pytest.approx(291.0) in collector.sensor_states["detection_distance"]
|
||||
)
|
||||
|
||||
async with (
|
||||
run_compiled(yaml_config),
|
||||
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 values (engineering mode frame):
|
||||
# moving=30, energy=100, still=30, energy=100, detect=30
|
||||
assert collector.sensor_states["moving_distance"][0] == pytest.approx(30.0)
|
||||
assert collector.sensor_states["still_distance"][0] == pytest.approx(30.0)
|
||||
assert collector.sensor_states["gate_0_move_energy"][0] == pytest.approx(100.0)
|
||||
assert collector.sensor_states["gate_1_move_energy"][0] == pytest.approx(65.0)
|
||||
assert collector.sensor_states["light"][0] == pytest.approx(87.0)
|
||||
|
||||
# Wait for Phase 3 frame: still_distance = 291cm (multi-byte)
|
||||
try:
|
||||
await asyncio.wait_for(phase3_still_received, timeout=3.0)
|
||||
except TimeoutError:
|
||||
pytest.fail(
|
||||
f"Timeout waiting for Phase 3 still_distance. Received:\n"
|
||||
f" still_distance: {collector.sensor_states['still_distance']}"
|
||||
)
|
||||
|
||||
assert pytest.approx(291.0) in collector.sensor_states["still_distance"]
|
||||
|
||||
# Wait for Phase 3: detection_distance = 291 (still-only target)
|
||||
try:
|
||||
await asyncio.wait_for(phase3_detect_received, timeout=3.0)
|
||||
except TimeoutError:
|
||||
pytest.fail(
|
||||
f"Timeout waiting for detection_distance=291. "
|
||||
f"Received: {collector.sensor_states['detection_distance']}"
|
||||
)
|
||||
|
||||
assert pytest.approx(291.0) in collector.sensor_states["detection_distance"]
|
||||
@@ -12,10 +12,10 @@ from __future__ import annotations
|
||||
import asyncio
|
||||
from pathlib import Path
|
||||
|
||||
from aioesphomeapi import EntityState, SensorState
|
||||
from aioesphomeapi import ButtonInfo, EntityState, SensorState
|
||||
import pytest
|
||||
|
||||
from .state_utils import InitialStateHelper, build_key_to_entity_mapping
|
||||
from .state_utils import InitialStateHelper, build_key_to_entity_mapping, find_entity
|
||||
from .types import APIClientConnectedFactory, RunCompiledFunction
|
||||
|
||||
|
||||
@@ -39,9 +39,17 @@ async def test_uart_mock_modbus(
|
||||
# Track sensor state updates (after initial state is swallowed)
|
||||
sensor_states: dict[str, list[float]] = {
|
||||
"basic_register": [],
|
||||
"delayed_response": [],
|
||||
"late_response": [],
|
||||
"no_response": [],
|
||||
"exception_response": [],
|
||||
}
|
||||
|
||||
basic_register_changed = loop.create_future()
|
||||
delayed_response_changed = loop.create_future()
|
||||
late_response_changed = loop.create_future()
|
||||
no_response_changed = loop.create_future()
|
||||
exception_response_changed = loop.create_future()
|
||||
|
||||
def on_state(state: EntityState) -> None:
|
||||
if isinstance(state, SensorState) and not state.missing_state:
|
||||
@@ -54,6 +62,23 @@ async def test_uart_mock_modbus(
|
||||
and not basic_register_changed.done()
|
||||
):
|
||||
basic_register_changed.set_result(True)
|
||||
elif (
|
||||
sensor_name == "delayed_response"
|
||||
and state.state == 255.0
|
||||
and not delayed_response_changed.done()
|
||||
):
|
||||
delayed_response_changed.set_result(True)
|
||||
elif (
|
||||
sensor_name == "late_response" and not late_response_changed.done()
|
||||
):
|
||||
late_response_changed.set_result(True)
|
||||
elif sensor_name == "no_response" and not no_response_changed.done():
|
||||
no_response_changed.set_result(True)
|
||||
elif (
|
||||
sensor_name == "exception_response"
|
||||
and not exception_response_changed.done()
|
||||
):
|
||||
exception_response_changed.set_result(True)
|
||||
|
||||
async with (
|
||||
run_compiled(yaml_config),
|
||||
@@ -74,20 +99,57 @@ async def test_uart_mock_modbus(
|
||||
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)
|
||||
|
||||
try:
|
||||
await asyncio.wait_for(delayed_response_changed, timeout=2.0)
|
||||
except TimeoutError:
|
||||
pytest.fail(
|
||||
f"Timeout waiting for delayed_response change. Received sensor states:\n"
|
||||
f" delayed_response: {sensor_states['delayed_response']}\n"
|
||||
)
|
||||
|
||||
try:
|
||||
await asyncio.wait_for(late_response_changed, timeout=2.0)
|
||||
pytest.fail(
|
||||
f"late_response change should not have been triggered, but was. Received sensor states:\n"
|
||||
f" late_response: {sensor_states['late_response']}\n"
|
||||
)
|
||||
except TimeoutError:
|
||||
pass # Expected timeout since we never inject a response for late_response
|
||||
|
||||
try:
|
||||
await asyncio.wait_for(no_response_changed, timeout=2.0)
|
||||
pytest.fail(
|
||||
f"no_response change should not have been triggered, but was. Received sensor states:\n"
|
||||
f" no_response: {sensor_states['no_response']}\n"
|
||||
)
|
||||
except TimeoutError:
|
||||
pass # Expected timeout since we never inject a response for no_response
|
||||
|
||||
# Wait for basic register to be updated with successful parse
|
||||
try:
|
||||
await asyncio.wait_for(basic_register_changed, timeout=15.0)
|
||||
await asyncio.wait_for(basic_register_changed, timeout=2.0)
|
||||
except TimeoutError:
|
||||
pytest.fail(
|
||||
f"Timeout waiting for Basic Register change. Received sensor states:\n"
|
||||
f" basic_register: {sensor_states['basic_register']}\n"
|
||||
)
|
||||
|
||||
try:
|
||||
await asyncio.wait_for(exception_response_changed, timeout=2.0)
|
||||
pytest.fail(
|
||||
f"exception_response change should not have been triggered, but was. Received sensor states:\n"
|
||||
f" exception_response: {sensor_states['exception_response']}\n"
|
||||
)
|
||||
except TimeoutError:
|
||||
pass
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.xfail(
|
||||
reason="There is a bug in UART which will timeout for long responses."
|
||||
)
|
||||
async def test_uart_mock_modbus_timing(
|
||||
yaml_config: str,
|
||||
run_compiled: RunCompiledFunction,
|
||||
@@ -143,9 +205,14 @@ async def test_uart_mock_modbus_timing(
|
||||
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=15.0)
|
||||
await asyncio.wait_for(voltage_changed, timeout=2.0)
|
||||
except TimeoutError:
|
||||
pytest.fail(
|
||||
f"Timeout waiting for SDM voltage change. Received sensor states:\n"
|
||||
|
||||
Reference in New Issue
Block a user