Merge branch 'esp32-pio-pch' into nrf52-pch

This commit is contained in:
J. Nick Koston
2026-08-27 12:38:39 -05:00
47 changed files with 1451 additions and 123 deletions
@@ -0,0 +1,74 @@
"""Config validation for the byte offset on holding-register write entities.
A 16-bit register write cannot target half a register, so an odd offset (or byte_offset) is
rejected for holding-register switches and outputs; even offsets and coil offsets pass.
"""
import pytest
from voluptuous import Invalid, MultipleInvalid
from esphome.components.modbus_controller.output import (
CONFIG_SCHEMA as OUTPUT_CONFIG_SCHEMA,
)
from esphome.components.modbus_controller.switch import (
CONFIG_SCHEMA as SWITCH_CONFIG_SCHEMA,
)
from esphome.const import CONF_ADDRESS, CONF_ID, CONF_NAME, CONF_OFFSET
def _switch_config(register_type: str, offset: int) -> dict:
return {
CONF_NAME: "test switch",
CONF_ADDRESS: 0x10,
"register_type": register_type,
CONF_OFFSET: offset,
}
def _output_config(register_type: str, offset: int) -> dict:
return {
CONF_ID: "test_output",
CONF_ADDRESS: 0x10,
"register_type": register_type,
CONF_OFFSET: offset,
}
def test_odd_offset_on_holding_switch_rejected() -> None:
with pytest.raises((Invalid, MultipleInvalid), match="odd"):
SWITCH_CONFIG_SCHEMA(_switch_config("holding", 3))
def test_even_offset_on_holding_switch_accepted() -> None:
config = SWITCH_CONFIG_SCHEMA(_switch_config("holding", 2))
assert config[CONF_OFFSET] == 2
def test_odd_offset_on_coil_switch_accepted() -> None:
"""A coil offset is a coil count, so odd values are fine."""
config = SWITCH_CONFIG_SCHEMA(_switch_config("coil", 3))
assert config[CONF_OFFSET] == 3
def test_odd_byte_offset_on_holding_switch_rejected() -> None:
"""byte_offset is the alias the validator must also catch."""
config = _switch_config("holding", 0)
del config[CONF_OFFSET]
config["byte_offset"] = 3
with pytest.raises((Invalid, MultipleInvalid), match="byte_offset"):
SWITCH_CONFIG_SCHEMA(config)
def test_odd_offset_on_holding_output_rejected() -> None:
with pytest.raises((Invalid, MultipleInvalid), match="odd"):
OUTPUT_CONFIG_SCHEMA(_output_config("holding", 3))
def test_even_offset_on_holding_output_accepted() -> None:
config = OUTPUT_CONFIG_SCHEMA(_output_config("holding", 2))
assert config[CONF_OFFSET] == 2
def test_odd_offset_on_coil_output_accepted() -> None:
config = OUTPUT_CONFIG_SCHEMA(_output_config("coil", 3))
assert config[CONF_OFFSET] == 3
@@ -11,6 +11,7 @@ from esphome.components.provisioning import (
CONFIG_SCHEMA,
FINAL_VALIDATE_SCHEMA,
register_source,
report_ap_without_sta,
report_hardcoded_credentials,
)
from esphome.const import CONF_TIMEOUT, PlatformFramework
@@ -66,6 +67,32 @@ def test_provisioning_no_warning_without_hardcoded_credentials(
assert "credentials" not in caplog.text
def test_provisioning_warns_on_ap_without_sta(
set_core_config: SetCoreConfigCallable,
caplog: pytest.LogCaptureFixture,
) -> None:
"""An access point with no station credentials triggers a reachability warning."""
set_core_config(PlatformFramework.ESP32_IDF)
register_source("network")
report_ap_without_sta()
with caplog.at_level(logging.WARNING):
FINAL_VALIDATE_SCHEMA({})
assert "access point" in caplog.text
assert "unreachable" in caplog.text
def test_provisioning_no_warning_without_ap(
set_core_config: SetCoreConfigCallable,
caplog: pytest.LogCaptureFixture,
) -> None:
"""No reachability warning when no AP-without-station setup is reported."""
set_core_config(PlatformFramework.ESP32_IDF)
register_source("network")
with caplog.at_level(logging.WARNING):
FINAL_VALIDATE_SCHEMA({})
assert "access point" not in caplog.text
def test_provisioning_rejects_zero_timeout(
set_core_config: SetCoreConfigCallable,
) -> None:
@@ -0,0 +1,6 @@
# The builder test compares against the Improv library's build_rpc_response,
# so the library must be part of the unit test build.
# Keep the version in sync with the pin in esphome/components/improv_base/__init__.py.
esphome:
libraries:
- improv/Improv@1.2.7
@@ -0,0 +1,102 @@
#include <gtest/gtest.h>
#include <array>
#include <cstdint>
#include <cstring>
#include <string>
#include <vector>
#include <improv.h>
namespace esphome::improv_base::testing {
namespace {
std::vector<uint8_t> build_with_builder(improv::Command command, const std::vector<std::string> &datum,
bool add_checksum) {
std::array<uint8_t, improv::RPC_RESPONSE_MAX_SIZE> buf;
improv::RpcResponseBuilder builder(buf, command);
for (const auto &str : datum) {
EXPECT_TRUE(builder.add_string(str.c_str(), str.size()));
}
auto out = builder.finish(add_checksum);
return {out.begin(), out.end()};
}
} // namespace
// The serial path sends builder output where build_rpc_response bytes went before,
// so the two must match exactly, including the trailing 0x00 when checksums are off.
TEST(RpcResponseBuilder, ByteIdenticalToBuildRpcResponse) {
const std::vector<std::string> device_info = {"ESPHome", "2026.9.0", "ESP32", "test-device"};
const std::vector<std::string> network = {"MySSID", "-67", "YES"};
const std::vector<std::string> empty = {};
const std::vector<std::string> max_payload = {std::string(254, 'x')};
for (bool add_checksum : {false, true}) {
for (const auto *datum : {&device_info, &network, &empty, &max_payload}) {
EXPECT_EQ(build_with_builder(improv::GET_DEVICE_INFO, *datum, add_checksum),
improv::build_rpc_response(improv::GET_DEVICE_INFO, *datum, add_checksum));
}
}
}
// Golden bytes independent of the library: command, data length, string entries,
// then the trailing byte (0x00 without checksum, additive checksum with).
TEST(RpcResponseBuilder, GoldenBytes) {
EXPECT_EQ(build_with_builder(improv::GET_WIFI_NETWORKS, {}, false), (std::vector<uint8_t>{0x04, 0x00, 0x00}));
EXPECT_EQ(build_with_builder(improv::GET_WIFI_NETWORKS, {"ab"}, false),
(std::vector<uint8_t>{0x04, 0x03, 0x02, 'a', 'b', 0x00}));
// Checksum: 0x04 + 0x03 + 0x02 + 'a' + 'b' = 0xCC
EXPECT_EQ(build_with_builder(improv::GET_WIFI_NETWORKS, {"ab"}, true),
(std::vector<uint8_t>{0x04, 0x03, 0x02, 'a', 'b', 0xCC}));
}
// esp32_improv calls finish() and build_rpc_response() with no checksum flag,
// so the two defaults must agree
TEST(RpcResponseBuilder, DefaultChecksumFlagMatches) {
const std::vector<std::string> urls = {"https://example.com"};
std::array<uint8_t, improv::RPC_RESPONSE_MAX_SIZE> buf;
improv::RpcResponseBuilder builder(buf, improv::WIFI_SETTINGS);
for (const auto &str : urls) {
EXPECT_TRUE(builder.add_string(str.c_str(), str.size()));
}
auto out = builder.finish();
EXPECT_EQ(std::vector<uint8_t>(out.begin(), out.end()), improv::build_rpc_response(improv::WIFI_SETTINGS, urls));
}
TEST(RpcResponseBuilder, PayloadBudget) {
std::array<uint8_t, improv::RPC_RESPONSE_MAX_SIZE> buf;
// 254 byte string fills the payload exactly; a second entry no longer fits
improv::RpcResponseBuilder full(buf, improv::GET_DEVICE_INFO);
const std::string big(254, 'x');
EXPECT_TRUE(full.add_string(big.c_str(), big.size()));
EXPECT_FALSE(full.add_string("y", 1));
// 255 byte string can never fit (its length byte would exceed the budget)
improv::RpcResponseBuilder over(buf, improv::GET_DEVICE_INFO);
const std::string too_big(255, 'y');
EXPECT_FALSE(over.add_string(too_big.c_str(), too_big.size()));
// A wildly out of range length must not wrap the position arithmetic
EXPECT_FALSE(over.add_string("z", static_cast<size_t>(-1)));
auto out = over.finish(false);
EXPECT_EQ(std::vector<uint8_t>(out.begin(), out.end()), (std::vector<uint8_t>{0x03, 0x00, 0x00}));
}
TEST(RpcResponseBuilder, FinishIsIdempotent) {
std::array<uint8_t, improv::RPC_RESPONSE_MAX_SIZE> buf;
improv::RpcResponseBuilder builder(buf, improv::GET_DEVICE_INFO);
EXPECT_TRUE(builder.add_string("abc", 3));
auto first = builder.finish(true);
const std::vector<uint8_t> expected(first.begin(), first.end());
EXPECT_FALSE(builder.add_string("late", 4));
auto again = builder.finish(true);
EXPECT_EQ(std::vector<uint8_t>(again.begin(), again.end()), expected);
// The checksum flag on a later call is ignored
auto no_checksum = builder.finish(false);
EXPECT_EQ(std::vector<uint8_t>(no_checksum.begin(), no_checksum.end()), expected);
}
} // namespace esphome::improv_base::testing
@@ -5,4 +5,6 @@ wifi:
logger:
hardware_uart: UART0
# next_url compiles the USE_IMPROV_SERIAL_NEXT_URL branch and add_next_url_
improv_serial:
next_url: https://example.com/?device_name={{device_name}}&ip_address={{ip_address}}
@@ -62,6 +62,12 @@ image:
url: http://www.faqs.org/images/library.jpg
format: AUTO
type: RGB565
- platform: online_image
id: online_qoi_image
url: https://www.example.org/image.qoi
format: QOI
type: RGB
transparency: alpha_channel
# Check the set_url action
esphome:
@@ -1,6 +1,7 @@
# Exercises the provisioning window: api registers as a provisioning source
# (encryption enabled, no key), the on_timeout automation, and the wifi +
# esp32_improv cross-component guards. improv_serial is intentionally NOT gated.
# (encryption enabled, no key), the on_timeout automation, and the wifi (AP +
# captive portal) and esp32_improv cross-component guards. improv_serial is
# intentionally NOT gated.
provisioning:
timeout: 1min
on_timeout:
@@ -13,6 +14,10 @@ api:
wifi:
ssid: MySSID
password: password1
ap:
ssid: MyAP
captive_portal:
improv_serial:
@@ -1,5 +1,6 @@
# Provisioning window on ESP8266 (no BLE Improv): api as a provisioning source
# and the wifi reboot guard. improv_serial is present and intentionally NOT gated.
# and the wifi (AP + captive portal) guards. improv_serial is present and
# intentionally NOT gated.
provisioning:
timeout: 1min
on_timeout:
@@ -12,5 +13,9 @@ api:
wifi:
ssid: MySSID
password: password1
ap:
ssid: MyAP
captive_portal:
improv_serial:
+2 -1
View File
@@ -9,7 +9,8 @@ def override_manifest(manifest: ComponentManifestOverride) -> None:
# tests have two decoder types and every retained decoder is under test.
async def to_code_testing(config: ConfigType) -> None:
enable_format("BMP")
enable_format("PNG")
enable_format("JPEG")
enable_format("PNG")
enable_format("QOI")
manifest.to_code = to_code_testing
@@ -70,11 +70,36 @@ static const uint8_t PNG_RGB_EXPECTED[4][4][3] = {
{{0x12, 0x34, 0x56}, {0x65, 0x43, 0x21}, {0xFE, 0xDC, 0xBA}, {0xAB, 0xCD, 0xEF}},
};
// 3x3 QOI, exercising all possible chunk types
static const uint8_t QOI_RGBA[] = {
0x71, 0x6F, 0x69, 0x66, // Header: 'qoif'
0x00, 0x00, 0x00, 0x03, // Width: 3
0x00, 0x00, 0x00, 0x03, // Height: 3
0x04, // Channels: 4 (RGBA)
0x00, // Colorspace: 0 (SRGB)
0xC1, // 1. QOI_OP_RUN
0x79, // 2. QOI_OP_DIFF
0xAA, 0x79, // 3. QOI_OP_LUMA
0xFE, 0xC8, 0x64, 0x32, // 4. QOI_OP_RGB
0xFF, 0x78, 0x50, 0x28,
0x64, // 5. QOI_OP_RGBA
0x31, // 6. QOI_OP_INDEX
0xC1, // 7. QOI_OP_RUN
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x01 // End Marker
};
static const uint8_t QOI_EXPECTED_RGBA[3][3][4] = {
{{0x00, 0x00, 0x00, 0xFF}, {0x00, 0x00, 0x00, 0xFF}, {0x01, 0x00, 0xFF, 0xFF}},
{{0x0A, 0x0A, 0x0A, 0xFF}, {0xC8, 0x64, 0x32, 0xFF}, {0x78, 0x50, 0x28, 0x64}},
{{0x01, 0x00, 0xFF, 0xFF}, {0x01, 0x00, 0xFF, 0xFF}, {0x01, 0x00, 0xFF, 0xFF}}
};
/// Exposes the protected decoder machinery so reuse and eviction can be observed directly.
class TestableRuntimeImage : public RuntimeImage {
public:
explicit TestableRuntimeImage(ImageFormat format)
: RuntimeImage(format, image::IMAGE_TYPE_RGB, image::TRANSPARENCY_OPAQUE, nullptr, false, 0, 0) {}
explicit TestableRuntimeImage(ImageFormat format, image::Transparency transparency = image::TRANSPARENCY_OPAQUE)
: RuntimeImage(format, image::IMAGE_TYPE_RGB, transparency, nullptr, false, 0, 0) {}
ImageDecoder *decoder() { return this->decoder_.get(); }
};
@@ -132,6 +157,19 @@ template<size_t H, size_t W> static void expect_pixels(TestableRuntimeImage &img
}
}
template<size_t H, size_t W>
static void expect_pixels_rgba(TestableRuntimeImage &img, const uint8_t (&expected)[H][W][4]) {
ASSERT_EQ(img.get_width(), static_cast<int>(W));
ASSERT_EQ(img.get_height(), static_cast<int>(H));
for (size_t y = 0; y < H; y++) {
for (size_t x = 0; x < W; x++) {
SCOPED_TRACE(::testing::Message() << "pixel (" << x << "," << y << ")");
Color color = img.get_pixel(x, y);
EXPECT_THAT((std::array<uint8_t, 4>{color.r, color.g, color.b, color.w}),
::testing::ElementsAreArray(expected[y][x]));
}
}
}
TEST(RuntimeImageDecoder, DecoderStaysWarmAcrossDecodes) {
TestableRuntimeImage img(BMP);
@@ -337,6 +375,33 @@ TEST(RuntimeImageDecoder, JpegDecoderStaysWarmAcrossDecodes) {
}
#endif // USE_RUNTIME_IMAGE_JPEG
TEST(RuntimeImageDecoder, QoiDecoderStaysWarmAcrossDecodes) {
TestableRuntimeImage img(QOI, image::TRANSPARENCY_ALPHA_CHANNEL);
ASSERT_TRUE(decode_all(img, QOI_RGBA, sizeof(QOI_RGBA)));
expect_pixels_rgba(img, QOI_EXPECTED_RGBA);
ImageDecoder *first = img.decoder();
ASSERT_NE(first, nullptr);
ASSERT_TRUE(decode_all(img, QOI_RGBA, sizeof(QOI_RGBA)));
expect_pixels_rgba(img, QOI_EXPECTED_RGBA);
EXPECT_EQ(img.decoder(), first) << "decoder must be reused, not reallocated";
}
TEST(RuntimeImageDecoder, QoiChunkedFeedDecodesLikeDownloadLoop) {
TestableRuntimeImage img(QOI, image::TRANSPARENCY_ALPHA_CHANNEL);
ASSERT_TRUE(decode_chunked(img, QOI_RGBA, sizeof(QOI_RGBA), 10));
expect_pixels_rgba(img, QOI_EXPECTED_RGBA);
ImageDecoder *first = img.decoder();
// Chunked again on the warm decoder: the cross-call resume state
// (current_index_ / paint_index_) must have been fully reset.
ASSERT_TRUE(decode_chunked(img, QOI_RGBA, sizeof(QOI_RGBA), 10));
expect_pixels_rgba(img, QOI_EXPECTED_RGBA);
EXPECT_EQ(img.decoder(), first);
}
TEST(RuntimeImageDecoder, SessionFlagsTrackLifecycle) {
TestableRuntimeImage img(BMP);
std::vector<uint8_t> buffer(BMP_24BPP, BMP_24BPP + sizeof(BMP_24BPP));
@@ -10,12 +10,14 @@ TEST(RuntimeImageMime, FormatForKnownMimeTypes) {
EXPECT_EQ(get_format_for_mime_type("image/bmp"), BMP);
EXPECT_EQ(get_format_for_mime_type("image/x-ms-bmp"), BMP);
EXPECT_EQ(get_format_for_mime_type("image/x-bmp"), BMP);
EXPECT_EQ(get_format_for_mime_type("image/png"), PNG);
EXPECT_EQ(get_format_for_mime_type("image/x-png"), PNG);
#ifdef USE_RUNTIME_IMAGE_JPEG
EXPECT_EQ(get_format_for_mime_type("image/jpeg"), JPEG);
EXPECT_EQ(get_format_for_mime_type("image/jpg"), JPEG);
#endif // USE_RUNTIME_IMAGE_JPEG
EXPECT_EQ(get_format_for_mime_type("image/png"), PNG);
EXPECT_EQ(get_format_for_mime_type("image/x-png"), PNG);
EXPECT_EQ(get_format_for_mime_type("image/qoi"), QOI);
EXPECT_EQ(get_format_for_mime_type("image/x-qoi"), QOI);
}
TEST(RuntimeImageMime, FormatMatchingIsCaseInsensitive) {
@@ -39,20 +41,22 @@ TEST(RuntimeImageMime, UnknownMimeTypeHasNoFormat) {
TEST(RuntimeImageMime, MimeTypeForFormatRoundTrip) {
EXPECT_STREQ(get_mime_type_for_format(BMP), "image/bmp");
EXPECT_STREQ(get_mime_type_for_format(PNG), "image/png");
#ifdef USE_RUNTIME_IMAGE_JPEG
EXPECT_STREQ(get_mime_type_for_format(JPEG), "image/jpeg");
#endif // USE_RUNTIME_IMAGE_JPEG
EXPECT_STREQ(get_mime_type_for_format(PNG), "image/png");
EXPECT_STREQ(get_mime_type_for_format(QOI), "image/qoi");
// AUTO has no single MIME type and falls back to the wildcard
EXPECT_STREQ(get_mime_type_for_format(AUTO), "image/*");
// Every decodable format must resolve back to itself through its MIME type
for (ImageFormat format : {
BMP,
PNG,
#ifdef USE_RUNTIME_IMAGE_JPEG
JPEG,
#endif // USE_RUNTIME_IMAGE_JPEG
PNG,
QOI,
}) {
EXPECT_EQ(get_format_for_mime_type(get_mime_type_for_format(format)), format) << format;
}
+21
View File
@@ -80,3 +80,24 @@ switch:
- platform: tuya
id: tuya_switch
switch_datapoint: 1
water_heater:
- platform: tuya
id: tuya_water_heater
name: Tuya Water Heater
switch_datapoint: 1
current_temperature_datapoint: 3
target_temperature_datapoint: 2
current_temperature_multiplier: 0.5
target_temperature_multiplier: 0.5
mode_datapoint: 4
eco_value: 0
electric_value: 2
supported_modes:
- "OFF"
- ECO
- ELECTRIC
visual:
min_temperature: 30
max_temperature: 75
target_temperature_step: 1
@@ -38,3 +38,5 @@ uart_mock:
improv_serial:
uart_id: mock_uart
# Deterministic on host: only the device name placeholder is used
next_url: https://example.com/?device={{device_name}}
@@ -0,0 +1,95 @@
esphome:
name: uart-mock-modbus-lambda-invert
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
uart_mock:
- id: virtual_uart_server
baud_rate: 9600
auto_start: true
debug:
on_tx:
- then:
- uart_mock.inject_rx:
id: virtual_uart_controller
data: !lambda return data;
- id: virtual_uart_controller
baud_rate: 9600
auto_start: true
debug:
on_tx:
- then:
- uart_mock.inject_rx:
id: virtual_uart_server
data: !lambda return data;
globals:
- id: reg40
type: uint16_t
initial_value: "5"
modbus:
- uart_id: virtual_uart_server
id: virtual_modbus_server
role: server
- uart_id: virtual_uart_controller
id: virtual_modbus_controller
role: client
turnaround_time: 10ms
modbus_controller:
- address: 1
modbus_id: virtual_modbus_controller
id: modbus_controller_1
update_interval: 1s
modbus_server:
- address: 1
modbus_id: virtual_modbus_server
id: modbus_server_1
registers:
- address: 0x40
value_type: U_WORD
read_lambda: return id(reg40);
write_lambda: id(reg40) = x; return true;
# An active-low holding switch: the write_lambda inverts the wire value, but the entity must still
# report the REQUESTED state. assumed_state keeps the register unpolled, so the published state comes
# only from write_state() - turning ON writes 0x0000 yet the switch shows ON.
switch:
- platform: modbus_controller
modbus_controller_id: modbus_controller_1
name: "invert_switch"
register_type: holding
address: 0x40
assumed_state: true
write_lambda: |-
return !x;
sensor:
- platform: modbus_controller
modbus_controller_id: modbus_controller_1
name: "reg_40"
address: 0x40
register_type: holding
value_type: U_WORD
button:
- platform: template
name: "Start Scenario"
id: start_scenario_btn
# This test does not have anything to start (mock is autostart)
@@ -100,7 +100,7 @@ switch:
offset: 2
assumed_state: true
# A holding-register switch that READS its state. Byte offset 6 -> register 0x10 + 6/2 = 0x13. Post-fix
# the switch itself resolves to 0x13 (whole registers fold into the address, residual byte stays) and
# the switch itself resolves to 0x13 (the even byte offset folds into the address as whole registers) and
# joins the 0x10..0x13 range, so no separate 0x13 sensor is needed. Pre-fix the whole byte offset folds
# into the address (0x16), where the server answers ILLEGAL_DATA_ADDRESS and the switch never publishes.
- platform: modbus_controller
+11 -4
View File
@@ -133,8 +133,15 @@ async def test_improv_serial_uart(
)
await waiter.wait_for("save_wifi_sta ssid=NewNet")
await waiter.wait_for("uart_mock", f"TX 12 bytes: {state_frame_hex(0x04)}")
# Settings RPC response with no URLs: payload [0x01, 0x00, 0x00] and footer
await waiter.wait_for("uart_mock", "TX 3 bytes: 01:00:00")
await waiter.wait_for(
"uart_mock", f"TX 2 bytes: {rpc_footer_hex(bytes([0x01, 0x00, 0x00]))}"
# Settings RPC response carries the formatted next_url and its footer
next_url = b"https://example.com/?device=improv-uart"
payload = (
bytes([CMD_WIFI_SETTINGS, len(next_url) + 1, len(next_url)])
+ next_url
+ b"\x00"
)
await waiter.wait_for(
"uart_mock",
f"TX {len(payload)} bytes: " + ":".join(f"{b:02X}" for b in payload),
)
await waiter.wait_for("uart_mock", f"TX 2 bytes: {rpc_footer_hex(payload)}")
+55 -7
View File
@@ -967,13 +967,6 @@ async def test_uart_mock_modbus_client_read_write(
_assert_no_modbus_errors(error_log_lines, warning_log_lines)
@pytest.mark.xfail(
strict=True,
reason="Byte-accurate register-offset writes land in the follow-up offset fix; "
"until then the byte offset is folded into the address (writes 0x12 instead of "
"0x11). The write and read assertions both flip via the same switch-constructor "
"fold. Remove this marker when that change merges.",
)
@pytest.mark.asyncio
async def test_uart_mock_modbus_register_offset(
yaml_config: str,
@@ -1065,6 +1058,61 @@ async def test_uart_mock_modbus_lambda_write(
await tracker.await_change(wrote_30, "reg_30", timeout=4.0)
@pytest.mark.asyncio
async def test_uart_mock_modbus_lambda_invert(
yaml_config: str,
run_compiled: RunCompiledFunction,
api_client_connected: APIClientConnectedFactory,
) -> None:
"""Test that a write_lambda's return value is the wire value only.
`invert_switch` is an active-low holding switch whose write_lambda returns !x. Turning it ON must
write 0x0000 to the register (observed through the independent reg_40 sensor) while the entity
reports ON - the requested state, not the inverted wire value. Turning it OFF writes 0xFFFF and
reports OFF. The switch is assumed_state, so the published state comes only from write_state().
"""
tracker = SensorTracker(["reg_40"])
initial = tracker.expect("reg_40", 5)
wrote_on = tracker.expect("reg_40", 0)
wrote_off = tracker.expect("reg_40", 65535)
async with (
run_compiled(yaml_config),
api_client_connected() as client,
):
entities = await tracker.setup_and_start_scenario(client)
await tracker.await_change(initial, "reg_40", timeout=4.0)
switch = find_entity(entities, "invert_switch", SwitchInfo)
assert switch is not None, "invert_switch not found"
client.switch_command(switch.key, True)
# The wire byte carries the inverted value...
await tracker.await_change(wrote_on, "reg_40", timeout=4.0)
# ...while the entity reports the requested state. Switch states are deduped, so this relies on
# wait_for_state's fresh subscribe_states re-dumping every entity's current state.
await wait_for_state(
client,
lambda s: (
getattr(s, "key", None) == switch.key
and getattr(s, "state", None) is True
),
timeout=6.0,
)
client.switch_command(switch.key, False)
await tracker.await_change(wrote_off, "reg_40", timeout=4.0)
await wait_for_state(
client,
lambda s: (
getattr(s, "key", None) == switch.key
and getattr(s, "state", None) is False
),
timeout=6.0,
)
@pytest.mark.asyncio
async def test_uart_mock_modbus_deprecated_write_buffer(
yaml_config: str,
+25
View File
@@ -665,6 +665,31 @@ def test_pch_compile_command_variants(tmp_path: Path) -> None:
assert cmd_dir == build
@pytest.mark.skipif(os.name == "nt", reason="symlinks need privileges on Windows")
def test_pch_compile_command_matches_src_through_symlink(tmp_path: Path) -> None:
"""Find the src TU when CMake spells paths through a different symlink (macOS /tmp)."""
from esphome.build_helpers.pch import pch_compile_command
real = tmp_path / "real"
(real / "src" / "esphome").mkdir(parents=True)
link = tmp_path / "link"
link.symlink_to(real, target_is_directory=True)
CORE.build_path = str(real)
build = real / "build"
build.mkdir()
header = build / "esphome_pch.h"
gch = build / "esphome_pch.h.gch"
src_file = str(link / "src" / "esphome" / "a.cpp")
(build / "compile_commands.json").write_text(
json.dumps([{"command": f"g++ -DX=1 -o a.obj -c {src_file}", "file": src_file}])
)
cmd, cmd_dir = pch_compile_command(build, header, gch)
assert cmd[:2] == ["g++", "-DX=1"]
assert cmd_dir == build
def test_pch_compile_command_rejects_unusable_entries(tmp_path: Path) -> None:
"""Malformed DB shapes and command-less entries skip cleanly instead of
producing a compiler-less argv retried every build."""