Merge branch 'esp8266-native-pch' into esp32-idf-pch

tests/benchmarks/components/api/bench_list_entities.cpp

@@ -49,7 +49,7 @@ static void Encode_ListEntitiesSensorResponse(benchmark::State &state) {
   auto msg = make_sensor_response();
   APIBuffer buffer;
   uint32_t size = msg.calculate_size();
-  buffer.resize(size);
+  (void) buffer.resize(size);
 
   for (auto _ : state) {
     for (int i = 0; i < kInnerIterations; i++) {
@@ -69,7 +69,7 @@ static void CalcAndEncode_ListEntitiesSensorResponse(benchmark::State &state) {
   for (auto _ : state) {
     for (int i = 0; i < kInnerIterations; i++) {
       uint32_t size = msg.calculate_size();
-      buffer.resize(size);
+      (void) buffer.resize(size);
       ProtoWriteBuffer writer(&buffer, 0);
       msg.encode(writer);
     }
@@ -117,7 +117,7 @@ static void Encode_ListEntitiesBinarySensorResponse(benchmark::State &state) {
   auto msg = make_binary_sensor_response();
   APIBuffer buffer;
   uint32_t size = msg.calculate_size();
-  buffer.resize(size);
+  (void) buffer.resize(size);
 
   for (auto _ : state) {
     for (int i = 0; i < kInnerIterations; i++) {
@@ -137,7 +137,7 @@ static void CalcAndEncode_ListEntitiesBinarySensorResponse(benchmark::State &sta
   for (auto _ : state) {
     for (int i = 0; i < kInnerIterations; i++) {
       uint32_t size = msg.calculate_size();
-      buffer.resize(size);
+      (void) buffer.resize(size);
       ProtoWriteBuffer writer(&buffer, 0);
       msg.encode(writer);
     }
@@ -202,7 +202,7 @@ static void Encode_ListEntitiesLightResponse(benchmark::State &state) {
   auto msg = make_light_response();
   APIBuffer buffer;
   uint32_t size = msg.calculate_size();
-  buffer.resize(size);
+  (void) buffer.resize(size);
 
   for (auto _ : state) {
     for (int i = 0; i < kInnerIterations; i++) {
@@ -222,7 +222,7 @@ static void CalcAndEncode_ListEntitiesLightResponse(benchmark::State &state) {
   for (auto _ : state) {
     for (int i = 0; i < kInnerIterations; i++) {
       uint32_t size = msg.calculate_size();
-      buffer.resize(size);
+      (void) buffer.resize(size);
       ProtoWriteBuffer writer(&buffer, 0);
       msg.encode(writer);
     }

tests/benchmarks/components/api/bench_log_response.cpp

@@ -23,7 +23,7 @@ static void Encode_LogResponse_Typical(benchmark::State &state) {
   msg.level = enums::LOG_LEVEL_DEBUG;
   msg.set_message(reinterpret_cast<const uint8_t *>(kTypicalLogLine), strlen(kTypicalLogLine));
   uint32_t size = msg.calculate_size();
-  buffer.resize(size);
+  (void) buffer.resize(size);
 
   for (auto _ : state) {
     for (int i = 0; i < kInnerIterations; i++) {
@@ -42,7 +42,7 @@ static void Encode_LogResponse_Short(benchmark::State &state) {
   msg.level = enums::LOG_LEVEL_INFO;
   msg.set_message(reinterpret_cast<const uint8_t *>(kShortLogLine), strlen(kShortLogLine));
   uint32_t size = msg.calculate_size();
-  buffer.resize(size);
+  (void) buffer.resize(size);
 
   for (auto _ : state) {
     for (int i = 0; i < kInnerIterations; i++) {
@@ -84,7 +84,7 @@ static void CalcAndEncode_LogResponse_Typical(benchmark::State &state) {
   for (auto _ : state) {
     for (int i = 0; i < kInnerIterations; i++) {
       uint32_t size = msg.calculate_size();
-      buffer.resize(size);
+      (void) buffer.resize(size);
       ProtoWriteBuffer writer(&buffer, 0);
       msg.encode(writer);
     }
@@ -105,7 +105,7 @@ static void CalcAndEncode_LogResponse_Typical_Fresh(benchmark::State &state) {
     for (int i = 0; i < kInnerIterations; i++) {
       APIBuffer buffer;
       uint32_t size = msg.calculate_size();
-      buffer.resize(size);
+      (void) buffer.resize(size);
       ProtoWriteBuffer writer(&buffer, 0);
       msg.encode(writer);
       benchmark::DoNotOptimize(buffer.data());

tests/benchmarks/components/api/bench_plaintext_frame.cpp

@@ -33,7 +33,7 @@ static void PlaintextFrame_WriteSensorState(benchmark::State &state) {
   // Pre-init buffer to typical TCP MSS size to avoid benchmarking
   // heap allocation — in real use the buffer is reused across writes.
   APIBuffer buffer;
-  buffer.reserve(1460);
+  (void) buffer.reserve(1460);
 
   for (auto _ : state) {
     for (int i = 0; i < kInnerIterations; i++) {
@@ -44,7 +44,7 @@ static void PlaintextFrame_WriteSensorState(benchmark::State &state) {
       msg.missing_state = false;
 
       uint32_t size = msg.calculate_size();
-      buffer.resize(padding + size);
+      (void) buffer.resize(padding + size);
       ProtoWriteBuffer writer(&buffer, padding);
       msg.encode(writer);
 
@@ -70,7 +70,7 @@ static void PlaintextFrame_WriteBatch5(benchmark::State &state) {
   // Pre-init buffer to typical TCP MSS size to avoid benchmarking
   // heap allocation — in real use the buffer is reused across writes.
   APIBuffer buffer;
-  buffer.reserve(1460);
+  (void) buffer.reserve(1460);
 
   for (auto _ : state) {
     for (int i = 0; i < kInnerIterations; i++) {
@@ -85,7 +85,7 @@ static void PlaintextFrame_WriteBatch5(benchmark::State &state) {
         msg.missing_state = false;
 
         uint32_t size = msg.calculate_size();
-        buffer.resize(offset + padding + size + footer);
+        (void) buffer.resize(offset + padding + size + footer);
         ProtoWriteBuffer writer(&buffer, offset + padding);
         msg.encode(writer);
 

tests/benchmarks/components/api/bench_proto_decode.cpp

@@ -16,7 +16,7 @@ static constexpr int kInnerIterations = 2000;
 template<typename T> static APIBuffer encode_message(const T &msg) {
   APIBuffer buffer;
   uint32_t size = msg.calculate_size();
-  buffer.resize(size);
+  (void) buffer.resize(size);
   ProtoWriteBuffer writer(&buffer, 0);
   msg.encode(writer);
   return buffer;

tests/benchmarks/components/api/bench_proto_encode.cpp

@@ -19,7 +19,7 @@ static void Encode_SensorStateResponse(benchmark::State &state) {
   msg.state = 23.5f;
   msg.missing_state = false;
   uint32_t size = msg.calculate_size();
-  buffer.resize(size);
+  (void) buffer.resize(size);
 
   for (auto _ : state) {
     for (int i = 0; i < kInnerIterations; i++) {
@@ -60,7 +60,7 @@ static void CalcAndEncode_SensorStateResponse(benchmark::State &state) {
   for (auto _ : state) {
     for (int i = 0; i < kInnerIterations; i++) {
       uint32_t size = msg.calculate_size();
-      buffer.resize(size);
+      (void) buffer.resize(size);
       ProtoWriteBuffer writer(&buffer, 0);
       msg.encode(writer);
     }
@@ -84,7 +84,7 @@ static void CalcAndEncode_SensorStateResponse_Fresh(benchmark::State &state) {
     for (int i = 0; i < kInnerIterations; i++) {
       APIBuffer buffer;
       uint32_t size = msg.calculate_size();
-      buffer.resize(size);
+      (void) buffer.resize(size);
       ProtoWriteBuffer writer(&buffer, 0);
       msg.encode(writer);
       benchmark::DoNotOptimize(buffer.data());
@@ -103,7 +103,7 @@ static void Encode_BinarySensorStateResponse(benchmark::State &state) {
   msg.state = true;
   msg.missing_state = false;
   uint32_t size = msg.calculate_size();
-  buffer.resize(size);
+  (void) buffer.resize(size);
 
   for (auto _ : state) {
     for (int i = 0; i < kInnerIterations; i++) {
@@ -126,7 +126,7 @@ static void Encode_HelloResponse(benchmark::State &state) {
   msg.server_info = StringRef::from_lit("esphome v2026.3.0");
   msg.name = StringRef::from_lit("living-room-sensor");
   uint32_t size = msg.calculate_size();
-  buffer.resize(size);
+  (void) buffer.resize(size);
 
   for (auto _ : state) {
     for (int i = 0; i < kInnerIterations; i++) {
@@ -158,7 +158,7 @@ static void Encode_LightStateResponse(benchmark::State &state) {
   msg.warm_white = 0.0f;
   msg.effect = StringRef::from_lit("rainbow");
   uint32_t size = msg.calculate_size();
-  buffer.resize(size);
+  (void) buffer.resize(size);
 
   for (auto _ : state) {
     for (int i = 0; i < kInnerIterations; i++) {
@@ -243,7 +243,7 @@ static void Encode_DeviceInfoResponse(benchmark::State &state) {
   auto msg = make_device_info_response();
   APIBuffer buffer;
   uint32_t total_size = msg.calculate_size();
-  buffer.resize(total_size);
+  (void) buffer.resize(total_size);
 
   for (auto _ : state) {
     for (int i = 0; i < kInnerIterations; i++) {
@@ -264,7 +264,7 @@ static void CalcAndEncode_DeviceInfoResponse(benchmark::State &state) {
   for (auto _ : state) {
     for (int i = 0; i < kInnerIterations; i++) {
       uint32_t size = msg.calculate_size();
-      buffer.resize(size);
+      (void) buffer.resize(size);
       ProtoWriteBuffer writer(&buffer, 0);
       msg.encode(writer);
     }
@@ -285,7 +285,7 @@ static void CalcAndEncode_DeviceInfoResponse_Fresh(benchmark::State &state) {
     for (int i = 0; i < kInnerIterations; i++) {
       APIBuffer buffer;
       uint32_t size = msg.calculate_size();
-      buffer.resize(size);
+      (void) buffer.resize(size);
       ProtoWriteBuffer writer(&buffer, 0);
       msg.encode(writer);
       benchmark::DoNotOptimize(buffer.data());
@@ -335,7 +335,7 @@ static void Encode_BLERawAdvs12(benchmark::State &state) {
   auto msg = make_ble_raw_advs_12();
   APIBuffer buffer;
   uint32_t total_size = msg.calculate_size();
-  buffer.resize(total_size);
+  (void) buffer.resize(total_size);
 
   for (auto _ : state) {
     for (int i = 0; i < kInnerIterations; i++) {
@@ -355,7 +355,7 @@ static void CalcAndEncode_BLERawAdvs12(benchmark::State &state) {
   for (auto _ : state) {
     for (int i = 0; i < kInnerIterations; i++) {
       uint32_t size = msg.calculate_size();
-      buffer.resize(size);
+      (void) buffer.resize(size);
       ProtoWriteBuffer writer(&buffer, 0);
       msg.encode(writer);
     }
@@ -372,7 +372,7 @@ static void CalcAndEncode_BLERawAdvs12_Fresh(benchmark::State &state) {
     for (int i = 0; i < kInnerIterations; i++) {
       APIBuffer buffer;
       uint32_t size = msg.calculate_size();
-      buffer.resize(size);
+      (void) buffer.resize(size);
       ProtoWriteBuffer writer(&buffer, 0);
       msg.encode(writer);
       benchmark::DoNotOptimize(buffer.data());

tests/benchmarks/components/api/bench_proto_proxy.cpp

@@ -16,7 +16,7 @@ static constexpr int kInnerIterations = 2000;
 // Encodes `src` into `out`. Caller owns `out` and must keep it alive across
 // the decode loop (decoded messages may store pointers back into its bytes).
 template<typename T> static void encode_into(APIBuffer &out, const T &src) {
-  out.resize(src.calculate_size());
+  (void) out.resize(src.calculate_size());
   ProtoWriteBuffer writer(&out, 0);
   src.encode(writer);
 }
@@ -33,7 +33,7 @@ static void Encode_ZWaveProxyFrame(benchmark::State &state) {
   msg.data = kZWaveFrameData;
   msg.data_len = sizeof(kZWaveFrameData);
   APIBuffer buffer;
-  buffer.resize(msg.calculate_size());
+  (void) buffer.resize(msg.calculate_size());
 
   for (auto _ : state) {
     for (int i = 0; i < kInnerIterations; i++) {
@@ -111,7 +111,7 @@ static void Encode_SerialProxyDataReceived(benchmark::State &state) {
   msg.instance = 0;
   msg.set_data(kSerialPayload, kSerialPayloadSize);
   APIBuffer buffer;
-  buffer.resize(msg.calculate_size());
+  (void) buffer.resize(msg.calculate_size());
 
   for (auto _ : state) {
     for (int i = 0; i < kInnerIterations; i++) {
@@ -171,7 +171,7 @@ static void Encode_InfraredRFReceiveEvent(benchmark::State &state) {
   msg.key = 0xDEADBEEF;
   msg.timings = &get_ir_timings_100();
   APIBuffer buffer;
-  buffer.resize(msg.calculate_size());
+  (void) buffer.resize(msg.calculate_size());
 
   for (auto _ : state) {
     for (int i = 0; i < kInnerIterations; i++) {
@@ -254,7 +254,7 @@ static APIBuffer build_infrared_rf_transmit_wire() {
   put_varint(1);
 
   APIBuffer buf;
-  buf.resize(len);
+  (void) buf.resize(len);
   std::memcpy(buf.data(), bytes, len);
   return buf;
 }

tests/benchmarks/components/api/bench_proto_varint.cpp

@@ -58,7 +58,7 @@ BENCHMARK(ProtoVarInt_Parse_FiveByte);
 
 static void Encode_Varint_Small(benchmark::State &state) {
   APIBuffer buffer;
-  buffer.resize(16);
+  (void) buffer.resize(16);
 
   for (auto _ : state) {
     for (int i = 0; i < kInnerIterations; i++) {
@@ -73,7 +73,7 @@ BENCHMARK(Encode_Varint_Small);
 
 static void Encode_Varint_Large(benchmark::State &state) {
   APIBuffer buffer;
-  buffer.resize(16);
+  (void) buffer.resize(16);
 
   for (auto _ : state) {
     for (int i = 0; i < kInnerIterations; i++) {
@@ -88,7 +88,7 @@ BENCHMARK(Encode_Varint_Large);
 
 static void Encode_Varint_MaxUint32(benchmark::State &state) {
   APIBuffer buffer;
-  buffer.resize(16);
+  (void) buffer.resize(16);
 
   for (auto _ : state) {
     for (int i = 0; i < kInnerIterations; i++) {

tests/component_tests/api/test_homeassistant_variables.py

@@ -0,0 +1,63 @@
+"""Tests for variables handling in homeassistant.event and homeassistant.action."""
+
+from collections.abc import Callable
+import logging
+from pathlib import Path
+
+import pytest
+
+CONFIG = "tests/component_tests/api/test_homeassistant_variables.yaml"
+
+
+def test_plain_string_with_return_is_compiled_as_lambda_with_warning(
+    generate_main: Callable[[str | Path], str],
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """A plain string with a return statement compiles as a lambda and warns."""
+    with caplog.at_level(logging.WARNING):
+        main_cpp = generate_main(CONFIG)
+
+    assert main_cpp.count('add_variable(ESPHOME_F("lambda_var"), []() {') == 2
+    assert "return millis();" in main_cpp
+    # The source text must not be sent as a static string value.
+    assert '"return millis();"' not in main_cpp
+    assert "missing the !lambda tag" in caplog.text
+
+
+def test_static_string_is_kept_as_static_value(
+    generate_main: Callable[[str | Path], str],
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """A static string stays static, PROGMEM wrapped, with no warning."""
+    with caplog.at_level(logging.WARNING):
+        main_cpp = generate_main(CONFIG)
+
+    assert (
+        main_cpp.count(
+            'add_variable(ESPHOME_F("static_var"), ESPHOME_F("static value"));'
+        )
+        == 2
+    )
+    assert "static value" not in caplog.text
+
+
+def test_static_id_value_stays_literal_with_hint(
+    generate_main: Callable[[str | Path], str],
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """Lambda source without a return stays literal text but warns."""
+    with caplog.at_level(logging.WARNING):
+        main_cpp = generate_main(CONFIG)
+
+    assert 'ESPHOME_F("id(test_sensor).state")' in main_cpp
+    assert "sent as literal text" in caplog.text
+
+
+def test_explicit_lambda_tag_is_compiled_as_lambda(
+    generate_main: Callable[[str | Path], str],
+) -> None:
+    """A !lambda value keeps working unchanged."""
+    main_cpp = generate_main(CONFIG)
+
+    assert 'add_variable(ESPHOME_F("tagged_var"), []() {' in main_cpp
+    assert "return App.get_name();" in main_cpp

tests/component_tests/api/test_homeassistant_variables.yaml

@@ -0,0 +1,32 @@
+esphome:
+  name: test
+  on_boot:
+    then:
+      # Plain strings with a return statement compile as lambdas
+      - homeassistant.event:
+          event: esphome.test_event
+          data_template:
+            message: "{{ lambda_var }} {{ static_var }} {{ tagged_var }}"
+          variables:
+            lambda_var: |-
+              return millis();
+            static_var: static value
+            tagged_var: !lambda return App.get_name();
+            hint_var: id(test_sensor).state
+      - homeassistant.action:
+          action: notify.notify
+          data_template:
+            message: "{{ lambda_var }} {{ static_var }}"
+          variables:
+            lambda_var: |-
+              return millis();
+            static_var: static value
+
+esp32:
+  board: esp32dev
+
+wifi:
+  ssid: SomeNetwork
+  password: SomePassword
+
+api:

tests/component_tests/external_components/test_init.py

@@ -1,16 +1,21 @@
-"""Tests for the external_components skip-update behavior driven by CORE.skip_external_update."""
+"""Tests for the external_components config pass."""
 
+import logging
 from pathlib import Path
 from typing import Any
 from unittest.mock import MagicMock
 
+import pytest
+
 from esphome.components.external_components import do_external_components_pass
 from esphome.const import (
     CONF_EXTERNAL_COMPONENTS,
+    CONF_PATH,
     CONF_REFRESH,
     CONF_SOURCE,
     CONF_URL,
     TYPE_GIT,
+    TYPE_LOCAL,
 )
 from esphome.core import CORE, TimePeriodSeconds
 
@@ -69,3 +74,112 @@ def test_external_components_normal_refresh(
     mock_clone_or_update.assert_called_once()
     call_args = mock_clone_or_update.call_args
     assert call_args.kwargs["refresh"] == TimePeriodSeconds(days=1)
+
+
+def test_external_components_logs_built_in_override(
+    tmp_path: Path,
+    mock_clone_or_update: MagicMock,
+    mock_install_meta_finder: MagicMock,
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """A source that provides a component with the same name as a built-in one logs an info message."""
+    mock_clone_or_update.return_value = (tmp_path, None)
+    config = _make_config(tmp_path)
+
+    for name in ("gpio", "some_custom_component"):
+        component_dir = tmp_path / "components" / name
+        component_dir.mkdir()
+        (component_dir / "__init__.py").write_text("# Test component")
+
+    with caplog.at_level(logging.INFO):
+        do_external_components_pass(config)
+
+    assert (
+        "External components are overriding built-in components:\n"
+        "  source: https://github.com/test/components\n"
+        "  components: gpio" in caplog.text
+    )
+    assert "some_custom_component" not in caplog.text
+
+
+def test_external_components_override_log_includes_ref(
+    tmp_path: Path,
+    mock_clone_or_update: MagicMock,
+    mock_install_meta_finder: MagicMock,
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """A git source with a ref logs the ref appended to the url."""
+    mock_clone_or_update.return_value = (tmp_path, None)
+    config = _make_config(tmp_path)
+    config[CONF_EXTERNAL_COMPONENTS][0][CONF_SOURCE] = "github://test/components@main"
+
+    component_dir = tmp_path / "components" / "gpio"
+    component_dir.mkdir()
+    (component_dir / "__init__.py").write_text("# Test component")
+
+    with caplog.at_level(logging.INFO):
+        do_external_components_pass(config)
+
+    assert "  source: https://github.com/test/components.git@main\n" in caplog.text
+
+
+def test_external_components_override_log_includes_git_path(
+    tmp_path: Path,
+    mock_clone_or_update: MagicMock,
+    mock_install_meta_finder: MagicMock,
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """A git source with a subdirectory path logs the path after the url."""
+    mock_clone_or_update.return_value = (tmp_path, None)
+    config = _make_config(tmp_path)
+    config[CONF_EXTERNAL_COMPONENTS][0][CONF_SOURCE][CONF_PATH] = "components"
+
+    component_dir = tmp_path / "components" / "gpio"
+    component_dir.mkdir()
+    (component_dir / "__init__.py").write_text("# Test component")
+
+    with caplog.at_level(logging.INFO):
+        do_external_components_pass(config)
+
+    assert "  source: https://github.com/test/components (components)\n" in caplog.text
+
+
+def test_external_components_override_log_local_source(
+    tmp_path: Path,
+    mock_install_meta_finder: MagicMock,
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """A local source logs its resolved path."""
+    components_dir = tmp_path / "my_components"
+    gpio_dir = components_dir / "gpio"
+    gpio_dir.mkdir(parents=True)
+    (gpio_dir / "__init__.py").write_text("# Test component")
+
+    CORE.config_path = tmp_path / "dummy.yaml"
+    config = {
+        CONF_EXTERNAL_COMPONENTS: [
+            {CONF_SOURCE: {"type": TYPE_LOCAL, CONF_PATH: "my_components"}}
+        ]
+    }
+
+    with caplog.at_level(logging.INFO):
+        do_external_components_pass(config)
+
+    assert f"  source: {components_dir}\n" in caplog.text
+    assert "  components: gpio" in caplog.text
+
+
+def test_external_components_no_override_no_log(
+    tmp_path: Path,
+    mock_clone_or_update: MagicMock,
+    mock_install_meta_finder: MagicMock,
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """A source that only provides components not shipped with ESPHome logs nothing."""
+    mock_clone_or_update.return_value = (tmp_path, None)
+    config = _make_config(tmp_path)
+
+    with caplog.at_level(logging.INFO):
+        do_external_components_pass(config)
+
+    assert "are overriding built-in components" not in caplog.text

tests/component_tests/http_request/config/test_esp32_default.yaml

@@ -0,0 +1,12 @@
+esphome:
+  name: test
+
+esp32:
+  board: nodemcu-32s
+
+wifi:
+  ssid: test
+  password: testtest
+
+http_request:
+  timeout: 10s

tests/component_tests/http_request/config/test_esp32_explicit.yaml

@@ -0,0 +1,13 @@
+esphome:
+  name: test
+
+esp32:
+  board: nodemcu-32s
+
+wifi:
+  ssid: test
+  password: testtest
+
+http_request:
+  timeout: 10s
+  watchdog_timeout: 20s

tests/component_tests/http_request/config/test_esp32_platform_wider.yaml

@@ -0,0 +1,13 @@
+esphome:
+  name: test
+
+esp32:
+  board: nodemcu-32s
+  watchdog_timeout: 60s
+
+wifi:
+  ssid: test
+  password: testtest
+
+http_request:
+  timeout: 10s

tests/component_tests/http_request/config/test_esp32_stock.yaml

@@ -0,0 +1,11 @@
+esphome:
+  name: test
+
+esp32:
+  board: nodemcu-32s
+
+wifi:
+  ssid: test
+  password: testtest
+
+http_request:

tests/component_tests/http_request/config/test_esp8266.yaml

@@ -0,0 +1,13 @@
+esphome:
+  name: test
+
+esp8266:
+  board: d1_mini
+
+wifi:
+  ssid: test
+  password: testtest
+
+http_request:
+  timeout: 10s
+  verify_ssl: false

tests/component_tests/http_request/config/test_rp2040.yaml

@@ -0,0 +1,13 @@
+esphome:
+  name: test
+
+rp2:
+  board: rpipicow
+
+wifi:
+  ssid: test
+  password: testtest
+
+http_request:
+  timeout: 10s
+  verify_ssl: false

tests/component_tests/http_request/test_init.py

@@ -0,0 +1,42 @@
+"""Tests for the http_request watchdog timeout default."""
+
+from collections.abc import Callable
+from pathlib import Path
+
+import pytest
+
+from esphome.config import read_config
+from esphome.const import CONF_WATCHDOG_TIMEOUT
+from esphome.core import CORE, TimePeriodMilliseconds
+
+
+@pytest.mark.parametrize(
+    ("yaml_file", "expected_ms"),
+    [
+        # stock 4.5s timeout: 3 x 4.5s plus 1s margin
+        ("test_esp32_stock.yaml", 14500),
+        # 3 x 10s plus 1s margin
+        ("test_esp32_default.yaml", 31000),
+        # esp32.watchdog_timeout: 60s is wider than the derived value and wins
+        ("test_esp32_platform_wider.yaml", 60000),
+        # explicit value is kept as is
+        ("test_esp32_explicit.yaml", 20000),
+    ],
+)
+def test_esp32_watchdog_timeout(
+    component_config_path: Callable[[str], Path], yaml_file: str, expected_ms: int
+) -> None:
+    CORE.config_path = component_config_path(yaml_file)
+    config = read_config({})
+    assert config["http_request"][CONF_WATCHDOG_TIMEOUT] == TimePeriodMilliseconds(
+        milliseconds=expected_ms
+    )
+
+
+@pytest.mark.parametrize("yaml_file", ["test_esp8266.yaml", "test_rp2040.yaml"])
+def test_other_platforms_leave_watchdog_unset(
+    component_config_path: Callable[[str], Path], yaml_file: str
+) -> None:
+    CORE.config_path = component_config_path(yaml_file)
+    config = read_config({})
+    assert CONF_WATCHDOG_TIMEOUT not in config["http_request"]

tests/component_tests/lvgl/config/dropdown_update_fires_event_test.yaml

@@ -0,0 +1,36 @@
+esphome:
+  name: test-dropdown-update-event
+  on_boot:
+    - lvgl.dropdown.update:
+        id: test_dropdown
+        selected_index: 2
+
+esp32:
+  board: lolin_c3_mini
+
+spi:
+  mosi_pin:
+    number: GPIO2
+    ignore_strapping_warning: true
+  clk_pin: GPIO1
+
+display:
+  - platform: mipi_spi
+    data_rate: 20MHz
+    model: st7735
+    cs_pin:
+      number: GPIO8
+      ignore_strapping_warning: true
+    dc_pin: GPIO3
+
+lvgl:
+  widgets:
+    - dropdown:
+        id: test_dropdown
+        options:
+          - First
+          - Second
+          - Third
+        on_update:
+          - lambda: |-
+              ESP_LOGD("test", "dropdown updated");

tests/component_tests/lvgl/test_dropdown_update_fires_event.py

@@ -0,0 +1,41 @@
+"""Regression test: lvgl.dropdown.update with selected_index must fire on_value/on_update.
+
+LvSelect (backing both dropdown and roller) did not set `value_property`, so the generic
+update-action machinery in automation.py never sent the synthetic update event for a
+`selected_index:` change made via `lvgl.dropdown.update`/`lvgl.roller.update`, unlike `value:`
+on number widgets or `text:` on text widgets. Fixed by setting `LvSelect.value_property` to
+`CONF_SELECTED_INDEX`.
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+import pytest
+
+from esphome.__main__ import generate_cpp_contents
+from esphome.config import read_config
+from esphome.core import CORE
+
+
+@pytest.fixture(scope="module")
+def main_cpp(request: pytest.FixtureRequest) -> str:
+    config_path = (
+        Path(request.fspath).parent / "config" / "dropdown_update_fires_event_test.yaml"
+    )
+    original_path = CORE.config_path
+    try:
+        CORE.config_path = config_path
+        CORE.config = read_config({})
+        generate_cpp_contents(CORE.config)
+        return CORE.cpp_main_section
+    finally:
+        CORE.config_path = original_path
+        CORE.reset()
+
+
+def test_dropdown_update_sends_update_event(main_cpp: str) -> None:
+    assert (
+        "lv_obj_send_event(test_dropdown->obj, lvgl::lv_update_event, nullptr)"
+        in main_cpp
+    )

tests/component_tests/remote_transmitter/test_non_blocking_gate.py

@@ -0,0 +1,48 @@
+"""non_blocking is family-gated at config validation; the CI build boards never compile
+the ISR paths, so this gate is the only CI-reachable coverage for the platform matrix."""
+
+import pytest
+
+from esphome.components.libretiny.const import (
+    FAMILY_BK7231N,
+    FAMILY_BK7231T,
+    FAMILY_BK7238,
+    FAMILY_RTL8710B,
+    FAMILY_RTL8720C,
+    KEY_FAMILY,
+    KEY_LIBRETINY,
+)
+from esphome.components.remote_transmitter import _validate_non_blocking_platform
+import esphome.config_validation as cv
+from esphome.const import PlatformFramework
+from esphome.core import CORE
+
+from ..types import SetCoreConfigCallable
+
+
+@pytest.mark.parametrize(
+    ("platform_framework", "family", "accepted"),
+    [
+        (PlatformFramework.ESP32_IDF, None, True),
+        (PlatformFramework.RTL87XX_ARDUINO, FAMILY_RTL8720C, True),
+        (PlatformFramework.RTL87XX_ARDUINO, FAMILY_RTL8710B, False),
+        (PlatformFramework.BK72XX_ARDUINO, FAMILY_BK7231N, True),
+        (PlatformFramework.BK72XX_ARDUINO, FAMILY_BK7238, True),
+        (PlatformFramework.BK72XX_ARDUINO, FAMILY_BK7231T, False),
+        (PlatformFramework.ESP8266_ARDUINO, None, False),
+    ],
+)
+def test_non_blocking_platform_gate(
+    set_core_config: SetCoreConfigCallable,
+    platform_framework: PlatformFramework,
+    family: str | None,
+    accepted: bool,
+) -> None:
+    set_core_config(platform_framework)
+    if family is not None:
+        CORE.data[KEY_LIBRETINY] = {KEY_FAMILY: family}
+    if accepted:
+        assert _validate_non_blocking_platform(True) is True
+    else:
+        with pytest.raises(cv.Invalid, match="non_blocking is only supported on"):
+            _validate_non_blocking_platform(True)

tests/component_tests/spi/test_psram_dma.py

@@ -0,0 +1,227 @@
+"""Tests for SPI PSRAM DMA configuration validation."""
+
+import pytest
+
+from esphome import config_validation as cv
+from esphome.components.esp32 import (
+    KEY_BOARD,
+    KEY_VARIANT,
+    VARIANT_ESP32,
+    VARIANT_ESP32S3,
+    VARIANT_ESP32S31,
+)
+from esphome.components.spi import (
+    CONF_INTERFACE_INDEX,
+    CONF_PSRAM_DMA,
+    _final_validate,
+    spi_device_schema,
+)
+from esphome.config import Config
+from esphome.const import CONF_ID, CONF_SPI_ID, KEY_FRAMEWORK_VERSION, PlatformFramework
+from esphome.core import CORE, ID
+from tests.component_tests.types import SetCoreConfigCallable
+
+
+def _schema() -> cv.Schema:
+    return spi_device_schema(
+        cs_pin_required=False,
+        default_data_rate="1MHz",
+        default_mode="MODE0",
+    )
+
+
+def _stage(
+    set_core_config: SetCoreConfigCallable,
+    platform_framework: PlatformFramework,
+    variant: str,
+    version: cv.Version,
+) -> None:
+    set_core_config(
+        platform_framework,
+        core_data={KEY_FRAMEWORK_VERSION: version},
+        platform_data={KEY_BOARD: "test-board", KEY_VARIANT: variant},
+    )
+    CORE.loaded_integrations.add("psram")
+
+
+def test_psram_dma_accepts_supported_idf_target(
+    set_core_config: SetCoreConfigCallable,
+) -> None:
+    _stage(
+        set_core_config,
+        PlatformFramework.ESP32_IDF,
+        VARIANT_ESP32S3,
+        cv.Version(5, 5, 3),
+    )
+
+    config = _schema()({CONF_PSRAM_DMA: True})
+
+    assert config[CONF_PSRAM_DMA] is True
+
+
+def test_psram_dma_accepts_esp32s31(
+    set_core_config: SetCoreConfigCallable,
+) -> None:
+    _stage(
+        set_core_config,
+        PlatformFramework.ESP32_IDF,
+        VARIANT_ESP32S31,
+        cv.Version(6, 0, 0),
+    )
+
+    config = _schema()({CONF_PSRAM_DMA: True})
+
+    assert config[CONF_PSRAM_DMA] is True
+
+
+def test_psram_dma_rejects_arduino(
+    set_core_config: SetCoreConfigCallable,
+) -> None:
+    _stage(
+        set_core_config,
+        PlatformFramework.ESP32_ARDUINO,
+        VARIANT_ESP32S3,
+        cv.Version(5, 5, 3),
+    )
+
+    with pytest.raises(cv.Invalid, match="only available with framework"):
+        _schema()({CONF_PSRAM_DMA: True})
+
+
+def test_psram_dma_rejects_target_without_capability(
+    set_core_config: SetCoreConfigCallable,
+) -> None:
+    _stage(
+        set_core_config,
+        PlatformFramework.ESP32_IDF,
+        VARIANT_ESP32,
+        cv.Version(5, 5, 3),
+    )
+
+    with pytest.raises(cv.Invalid, match="PSRAM DMA is only available"):
+        _schema()({CONF_PSRAM_DMA: True})
+
+
+def test_psram_dma_false_is_portable(
+    set_core_config: SetCoreConfigCallable,
+) -> None:
+    _stage(
+        set_core_config,
+        PlatformFramework.ESP32_ARDUINO,
+        VARIANT_ESP32,
+        cv.Version(5, 5, 2),
+    )
+
+    config = _schema()({CONF_PSRAM_DMA: False})
+
+    assert config[CONF_PSRAM_DMA] is False
+
+
+def test_psram_dma_rejects_older_idf(
+    set_core_config: SetCoreConfigCallable,
+) -> None:
+    _stage(
+        set_core_config,
+        PlatformFramework.ESP32_IDF,
+        VARIANT_ESP32S3,
+        cv.Version(5, 5, 2),
+    )
+
+    with pytest.raises(cv.Invalid, match="requires at least framework version 5.5.3"):
+        _schema()({CONF_PSRAM_DMA: True})
+
+
+def test_psram_dma_requires_psram_component(
+    set_core_config: SetCoreConfigCallable,
+) -> None:
+    _stage(
+        set_core_config,
+        PlatformFramework.ESP32_IDF,
+        VARIANT_ESP32S3,
+        cv.Version(5, 5, 3),
+    )
+    CORE.loaded_integrations.remove("psram")
+
+    with pytest.raises(cv.Invalid, match="requires component psram"):
+        _schema()({CONF_PSRAM_DMA: True})
+
+
+def _full_spi_config(*, hardware: bool, with_device: bool = True) -> tuple[Config, ID]:
+    bus_id = ID("spi_bus", is_declaration=True, type="SPIComponent")
+    bus = {CONF_ID: bus_id}
+    if hardware:
+        bus[CONF_INTERFACE_INDEX] = 0
+    full = Config()
+    full["spi"] = [bus]
+    if with_device:
+        full["spi_device_test"] = {
+            CONF_SPI_ID: ID("spi_bus"),
+            CONF_PSRAM_DMA: True,
+        }
+    full.declare_ids.append((bus_id, ["spi", 0, CONF_ID]))
+    return full, ID("spi_bus", is_declaration=False, type="SPIComponent")
+
+
+def test_psram_dma_accepts_hardware_spi(
+    set_core_config: SetCoreConfigCallable,
+) -> None:
+    full_config, _ = _full_spi_config(hardware=True)
+    set_core_config(PlatformFramework.ESP32_IDF, full_config=full_config)
+
+    _final_validate(full_config["spi"])
+
+
+def test_psram_dma_rejects_software_spi(
+    set_core_config: SetCoreConfigCallable,
+) -> None:
+    full_config, _ = _full_spi_config(hardware=False)
+    set_core_config(PlatformFramework.ESP32_IDF, full_config=full_config)
+
+    with pytest.raises(cv.Invalid, match="psram_dma requires a hardware SPI") as error:
+        _final_validate(full_config["spi"])
+
+    assert error.value.path[-2:] == ["spi_device_test", CONF_PSRAM_DMA]
+
+
+def test_spi_bus_rejects_psram_dma_device_without_component_final_validation(
+    set_core_config: SetCoreConfigCallable,
+) -> None:
+    full_config, bus_id = _full_spi_config(hardware=False, with_device=False)
+    full_config["device_without_final_validation"] = {
+        CONF_SPI_ID: bus_id,
+        CONF_PSRAM_DMA: True,
+    }
+    set_core_config(PlatformFramework.ESP32_IDF, full_config=full_config)
+
+    with pytest.raises(cv.Invalid, match="psram_dma requires a hardware SPI") as error:
+        _final_validate(full_config["spi"])
+
+    assert error.value.path[-2:] == [
+        "device_without_final_validation",
+        CONF_PSRAM_DMA,
+    ]
+
+
+def test_psram_dma_accepts_hardware_device_with_mixed_buses(
+    set_core_config: SetCoreConfigCallable,
+) -> None:
+    software_bus_id = ID("software_bus", is_declaration=True, type="SPIComponent")
+    hardware_bus_id = ID("hardware_bus", is_declaration=True, type="SPIComponent")
+    full_config = Config()
+    full_config["spi"] = [
+        {CONF_ID: software_bus_id},
+        {CONF_ID: hardware_bus_id, CONF_INTERFACE_INDEX: 0},
+    ]
+    full_config["spi_device_test"] = {
+        CONF_SPI_ID: ID("hardware_bus"),
+        CONF_PSRAM_DMA: True,
+    }
+    full_config.declare_ids.extend(
+        (
+            (software_bus_id, ["spi", 0, CONF_ID]),
+            (hardware_bus_id, ["spi", 1, CONF_ID]),
+        )
+    )
+    set_core_config(PlatformFramework.ESP32_IDF, full_config=full_config)
+
+    _final_validate(full_config["spi"])

tests/components/api/common-base.yaml

@@ -9,6 +9,14 @@ esphome:
           event: esphome.button_pressed
           data:
             message: Button was pressed
+      - homeassistant.event:
+          event: esphome.button_pressed_with_variables
+          data_template:
+            message: Button {{ button_name }} ({{ button_index }}) was pressed from {{ button_source }}
+          variables:
+            button_name: !lambda 'return std::string("test_button");'
+            button_index: !lambda 'return 1;'
+            button_source: static_value
       - homeassistant.action:
           action: notify.html5
           data:

tests/components/api/test_proto_mac_varint.cpp

@@ -54,7 +54,7 @@ static void verify_mac(uint64_t mac, size_t expected_bytes) {
   size_t ref_len = reference_encode(mac, ref_buf);
 
   APIBuffer api_buf;
-  api_buf.resize(16);
+  ASSERT_TRUE(api_buf.resize(16));
   uint8_t *pos = api_buf.data();
 #ifdef ESPHOME_DEBUG_API
   uint8_t *proto_debug_end_ = api_buf.data() + api_buf.size();

tests/components/climate/climate_test.cpp

@@ -0,0 +1,73 @@
+#include <gtest/gtest.h>
+#include "esphome/components/climate/climate.h"
+
+namespace esphome::climate::testing {
+
+// Minimal concrete Climate that offers a fixed set of modes, so the restore path can be exercised
+// without any hardware or platform component.
+class TestClimate : public Climate {
+ public:
+  ClimateTraits traits() override {
+    auto traits = ClimateTraits();
+    traits.set_supported_modes({CLIMATE_MODE_OFF, CLIMATE_MODE_COOL});
+    traits.set_supported_fan_modes({CLIMATE_FAN_LOW, CLIMATE_FAN_HIGH});
+    return traits;
+  }
+
+ protected:
+  void control(const ClimateCall &call) override {}
+};
+
+TEST(ClimateRestoreStateTest, RestoresASupportedMode) {
+  TestClimate climate;
+  // Value-initialized: several members (mode, swing_mode, the temperature union) have no default
+  // member initializer, so leaving the {} off would read indeterminate values.
+  ClimateDeviceRestoreState state{};
+  state.mode = CLIMATE_MODE_COOL;
+
+  state.apply(&climate);
+
+  EXPECT_EQ(climate.mode, CLIMATE_MODE_COOL);
+}
+
+TEST(ClimateRestoreStateTest, DoesNotRestoreAnUnsupportedMode) {
+  TestClimate climate;
+  ClimateDeviceRestoreState state{};
+  state.mode = CLIMATE_MODE_HEAT;
+
+  state.apply(&climate);
+
+  // The device never advertised HEAT, so the mode stays where it was.
+  EXPECT_EQ(climate.mode, CLIMATE_MODE_OFF);
+}
+
+TEST(ClimateRestoreStateTest, LeavesTheCurrentModeAloneRatherThanForcingOff) {
+  TestClimate climate;
+  // apply() is public and nothing restricts it to setup(), so the entity is not necessarily off
+  // when an unsupported mode is dropped. It keeps what it had rather than being forced to OFF.
+  climate.mode = CLIMATE_MODE_COOL;
+  ClimateDeviceRestoreState state{};
+  state.mode = CLIMATE_MODE_HEAT;
+
+  state.apply(&climate);
+
+  EXPECT_EQ(climate.mode, CLIMATE_MODE_COOL);
+}
+
+TEST(ClimateRestoreStateTest, KeepsRestoringTheOtherFieldsWhenTheModeIsDropped) {
+  TestClimate climate;
+  ClimateDeviceRestoreState state{};
+  state.mode = CLIMATE_MODE_HEAT;
+  state.target_temperature = 21.0f;
+  state.uses_custom_fan_mode = false;
+  state.fan_mode = CLIMATE_FAN_HIGH;
+
+  state.apply(&climate);
+
+  EXPECT_EQ(climate.mode, CLIMATE_MODE_OFF);
+  EXPECT_FLOAT_EQ(climate.target_temperature, 21.0f);
+  // Compared as an optional: this asserts both that the fan mode was restored and what it holds.
+  EXPECT_EQ(climate.fan_mode, CLIMATE_FAN_HIGH);
+}
+
+}  // namespace esphome::climate::testing

tests/components/climate_ir_lg/common.yaml

@@ -12,5 +12,8 @@ climate:
   - platform: climate_ir_lg
     name: LG Climate
     transmitter_id: xmitr
+    header_high: 3300us
+    header_low: 9840us
+    advanced_commands_support: true
     sensor: climate_ir_lg_temp_sensor
     humidity_sensor: humidity_sensor

tests/components/hoermann_hcp/button/hoermann_hcp_button_test.cpp

@@ -0,0 +1,72 @@
+#include <gtest/gtest.h>
+
+#include "esphome/components/hoermann_hcp/button/hoermann_hcp_button.h"
+
+#include "../common.h"
+
+namespace esphome::hoermann_hcp::testing {
+
+// The intermediate positions are named in the second register, which repeats that name on release.
+TEST(HoermannHcpButtonTest, VentButtonSendsTheVentCommand) {
+  TestableHoermannHcp door;
+  HoermannHcpVentButton vent(&door);
+  connect_controller(door);
+
+  vent.press();
+
+  auto [pressed, pressed_2] = poll_command(door);
+  EXPECT_EQ(pressed, 0x0200);
+  EXPECT_EQ(pressed_2, 0x4000);
+  std::this_thread::sleep_for(KEY_PRESS_ELAPSED);
+  auto [released, released_2] = poll_command(door);
+  EXPECT_EQ(released, 0x0100);
+  EXPECT_EQ(released_2, 0x4000);
+}
+
+TEST(HoermannHcpButtonTest, HalfOpenButtonSendsTheHalfOpenCommand) {
+  TestableHoermannHcp door;
+  HoermannHcpHalfOpenButton half_open(&door);
+  connect_controller(door);
+
+  half_open.press();
+
+  auto [pressed, pressed_2] = poll_command(door);
+  EXPECT_EQ(pressed, 0x0200);
+  EXPECT_EQ(pressed_2, 0x0400);
+  std::this_thread::sleep_for(KEY_PRESS_ELAPSED);
+  auto [released, released_2] = poll_command(door);
+  EXPECT_EQ(released, 0x0100);
+  EXPECT_EQ(released_2, 0x0400);
+}
+
+// The door drives to the vent position on its own, so a position the cover was still travelling to must not
+// stop it on the way there.
+TEST(HoermannHcpButtonTest, VentAbandonsAnArmedTarget) {
+  TestableHoermannHcp door;  // starts out fully closed
+  HoermannHcpVentButton vent(&door);
+  connect_controller(door);
+  door.set_position(0.5f);
+  consume_command(door);
+  door.on_write_registers(BROADCAST_REG, make_registers({0x0000, 0x0014, 0x0100}));
+  ASSERT_EQ(door.get_door_state(), DoorState::OPENING);
+
+  vent.press();
+  consume_command(door);
+
+  // Position 120/200 = 0.6 is past the abandoned target, which must no longer stop the door.
+  door.on_write_registers(BROADCAST_REG, make_registers({0x0000, 0x0078, 0x0100}));
+  EXPECT_EQ(poll_command(door).first, 0x0000);
+}
+
+// A button carries no state, so a refused press is simply dropped rather than fired once the controller
+// turns up, which could be much later.
+TEST(HoermannHcpButtonTest, PressWithoutABusControllerSendsNothing) {
+  HoermannHcp door;  // never contacted by a bus controller
+  HoermannHcpVentButton vent(&door);
+
+  vent.press();
+
+  EXPECT_EQ(poll_command(door).first, 0x0000);
+}
+
+}  // namespace esphome::hoermann_hcp::testing

tests/components/hoermann_hcp/common.yaml

@@ -12,6 +12,13 @@ binary_sensor:
     is_connected:
       name: Garage Connected
 
+button:
+  - platform: hoermann_hcp
+    vent:
+      name: Garage Vent
+    half_open:
+      name: Garage Half Open
+
 light:
   - platform: hoermann_hcp
     name: Garage Light

tests/components/homeassistant/common.yaml

@@ -12,7 +12,7 @@ esphome:
           data_template:
             message: The humidity is {{ my_variable }}%.
           variables:
-            my_variable: "return id(ha_hello_world_temperature).state;"
+            my_variable: !lambda "return id(ha_hello_world_temperature).state;"
       - homeassistant.action:
           action: notify.html5
           data:
@@ -24,7 +24,7 @@ esphome:
           data_template:
             message: The humidity is {{ my_variable }}%.
           variables:
-            my_variable: "return id(ha_hello_world_temperature).state;"
+            my_variable: !lambda "return id(ha_hello_world_temperature).state;"
 
 wifi:
   ssid: MySSID

tests/components/improv_serial/common-uart-bus.yaml

@@ -0,0 +1,11 @@
+wifi:
+  ssid: MySSID
+  password: password1
+
+# Serial logging off; on a dedicated UART bus improv_serial must not
+# require the logger's serial settings
+logger:
+  baud_rate: 0
+
+improv_serial:
+  uart_id: uart_bus

tests/components/improv_serial/test-uart-bus.esp32-idf.yaml

@@ -0,0 +1,3 @@
+packages:
+  uart: !include ../../test_build_components/common/uart/esp32-idf.yaml
+  improv_serial: !include common-uart-bus.yaml

tests/components/improv_serial/test-uart-bus.esp8266-ard.yaml

@@ -0,0 +1,3 @@
+packages:
+  uart: !include ../../test_build_components/common/uart/esp8266-ard.yaml
+  improv_serial: !include common-uart-bus.yaml

tests/components/modbus_controller/command_payload_test.cpp

@@ -13,7 +13,7 @@ namespace esphome::modbus_controller::testing {
 // malformed. Built at its true byte count, the oversize frame is refused by the hub's size check with
 // a log instead.
 TEST(ModbusCommandPayload, CoilWritePayloadIsExactSizedNotTruncated) {
-  ModbusController controller;
+  ModbusController controller(nullptr, 1);
   std::vector<bool> coils(modbus::MAX_NUM_OF_COILS_TO_WRITE + 1, true);
   auto cmd = ModbusCommandItem::create_write_multiple_coils(&controller, 0x10, coils);
   EXPECT_EQ(cmd.payload.size(), modbus::packed_bit_bytes(coils.size()));
@@ -21,7 +21,7 @@ TEST(ModbusCommandPayload, CoilWritePayloadIsExactSizedNotTruncated) {
 
 // LSB-first packing with zeroed pad bits, matching the wire layout the PDU builders produce.
 TEST(ModbusCommandPayload, CoilWritePacksLsbFirstWithZeroPad) {
-  ModbusController controller;
+  ModbusController controller(nullptr, 1);
   const std::vector<bool> coils{true, false, true, true};
   auto cmd = ModbusCommandItem::create_write_multiple_coils(&controller, 0x10, coils);
   ASSERT_EQ(cmd.payload.size(), 1u);

tests/components/remote_transmitter/test.bk72xx-ard.yaml

@@ -2,6 +2,7 @@ remote_transmitter:
   id: xmitr
   pin: GPIO26
   carrier_duty_percent: 50%
+  # non_blocking is bk7231n/bk7238-only; the CI board is a BK7252
 
 packages:
   buttons: !include common-buttons.yaml

tests/components/remote_transmitter/test.rtl87xx-ard.yaml

@@ -2,6 +2,7 @@ remote_transmitter:
   id: xmitr
   pin: GPIO12
   carrier_duty_percent: 50%
+  # non_blocking is rtl8720c-only; the CI board is an RTL8710B
 
 packages:
   buttons: !include common-buttons.yaml

tests/components/spi_device/test.esp32-s3-idf.yaml

@@ -0,0 +1,10 @@
+packages:
+  spi: !include ../../test_build_components/common/spi/esp32-s3-idf.yaml
+  common: !include common.yaml
+psram:
+  mode: octal
+spi_device:
+  - id: spi_device_psram_dma_test
+    psram_dma: true
+    data_rate: 1MHz
+    spi_mode: 0

tests/integration/fixtures/external_components/wifi/__init__.py

@@ -0,0 +1,39 @@
+"""Host-only stub of the wifi component for integration tests.
+
+HOST-ONLY TEST COMPONENT: this shadows the real wifi component for EVERY
+fixture that uses the shared external_components directory. Any host fixture
+with a wifi block gets this stub, not the real component: fixed scan results,
+is_connected() hardwired true, and save_wifi_sta that only logs. See
+wifi_component.h for the full behavior.
+"""
+
+import esphome.codegen as cg
+import esphome.config_validation as cv
+from esphome.const import CONF_ID, CONF_PASSWORD, CONF_SSID, CONF_USE_ADDRESS
+from esphome.types import ConfigType
+
+CODEOWNERS = ["@esphome/tests"]
+
+wifi_ns = cg.esphome_ns.namespace("wifi")
+WiFiComponent = wifi_ns.class_("WiFiComponent", cg.Component)
+
+CONFIG_SCHEMA = cv.Schema(
+    {
+        cv.GenerateID(): cv.declare_id(WiFiComponent),
+        # Accepted for fixture realism; the stub ignores them
+        cv.Optional(CONF_SSID): cv.string,
+        cv.Optional(CONF_PASSWORD): cv.string,
+        # Read by StorageJSON via CORE.address whenever a wifi block exists
+        cv.Optional(CONF_USE_ADDRESS, default="localhost"): cv.string,
+    }
+).extend(cv.COMPONENT_SCHEMA)
+
+
+def check_placeholder_credentials(config: ConfigType) -> None:
+    """Compile-time hook the esphome CLI imports from the wifi module; no-op here."""
+
+
+async def to_code(config: ConfigType) -> None:
+    var = cg.new_Pvariable(config[CONF_ID])
+    await cg.register_component(var, config)
+    cg.add_define("USE_WIFI")

tests/integration/fixtures/external_components/wifi/scan_list.h

@@ -0,0 +1 @@
+../../../../../esphome/components/wifi/scan_list.h

tests/integration/fixtures/external_components/wifi/wifi_component.cpp

@@ -0,0 +1,40 @@
+#include "wifi_component.h"
+
+#include "esphome/core/log.h"
+
+namespace esphome::wifi {
+
+static const char *const TAG = "wifi_stub";
+
+WiFiComponent *global_wifi_component = nullptr;  // NOLINT(cppcoreguidelines-avoid-non-const-global-variables)
+
+WiFiComponent::WiFiComponent() { global_wifi_component = this; }
+
+void WiFiComponent::setup() { ESP_LOGI(TAG, "Stub wifi ready"); }
+
+void WiFiComponent::dump_config() { ESP_LOGCONFIG(TAG, "Stub wifi"); }
+
+void WiFiComponent::start_scanning() {
+  // Duplicate TestNet entry (weaker) and a hidden entry exercise the
+  // should_show_scan_entry dedup and filtering logic
+  this->scan_result_.clear();
+  this->scan_result_.emplace_back("TestNet", -50, true, false);
+  this->scan_result_.emplace_back("TestNet", -60, true, false);
+  this->scan_result_.emplace_back("OpenNet", -70, false, false);
+  this->scan_result_.emplace_back("", -40, false, true);
+  ESP_LOGI(TAG, "Scan complete with %zu results", this->scan_result_.size());
+}
+
+void WiFiComponent::set_sta(const WiFiAP &ap) { ESP_LOGI(TAG, "set_sta ssid=%s", ap.get_ssid().c_str()); }
+
+void WiFiComponent::start_connecting(const WiFiAP &ap) {
+  ESP_LOGI(TAG, "start_connecting ssid=%s", ap.get_ssid().c_str());
+}
+
+void WiFiComponent::clear_sta() { ESP_LOGI(TAG, "clear_sta"); }
+
+void WiFiComponent::save_wifi_sta(StringRef ssid, StringRef password) {
+  ESP_LOGI(TAG, "save_wifi_sta ssid=%s password_len=%zu", ssid.c_str(), password.size());
+}
+
+}  // namespace esphome::wifi

tests/integration/fixtures/external_components/wifi/wifi_component.h

@@ -0,0 +1,77 @@
+#pragma once
+
+// ============================================================================
+// HOST-ONLY TEST COMPONENT — DO NOT COPY TO PRODUCTION CODE
+//
+// Stub of the real wifi component with just enough API surface for
+// improv_serial to build and run on the host platform. Scan results are
+// fixed, "connecting" succeeds immediately, and save_wifi_sta only logs so
+// tests can assert on the log output.
+// ============================================================================
+
+#include "esphome/components/network/ip_address.h"
+#include "esphome/core/component.h"
+#include "esphome/core/string_ref.h"
+
+#include <string>
+#include <vector>
+
+namespace esphome::wifi {
+
+class WiFiAP {
+ public:
+  void set_ssid(const char *ssid) { this->ssid_ = ssid; }
+  void set_password(const char *password) { this->password_ = password; }
+  StringRef get_ssid() const { return StringRef(this->ssid_); }
+  StringRef get_password() const { return StringRef(this->password_); }
+
+ protected:
+  std::string ssid_;
+  std::string password_;
+};
+
+class WiFiScanResult {
+ public:
+  WiFiScanResult(const char *ssid, int8_t rssi, bool with_auth, bool hidden)
+      : ssid_(ssid), rssi_(rssi), with_auth_(with_auth), hidden_(hidden) {}
+  StringRef get_ssid() const { return StringRef(this->ssid_); }
+  int8_t get_rssi() const { return this->rssi_; }
+  bool get_with_auth() const { return this->with_auth_; }
+  bool get_is_hidden() const { return this->hidden_; }
+  bool ssid_equals(const WiFiScanResult &other) const { return this->ssid_ == other.ssid_; }
+
+ protected:
+  std::string ssid_;
+  int8_t rssi_;
+  bool with_auth_;
+  bool hidden_;
+};
+
+class WiFiComponent : public Component {
+ public:
+  WiFiComponent();
+  void setup() override;
+  void dump_config() override;
+  float get_setup_priority() const override { return setup_priority::WIFI; }
+
+  bool has_sta() const { return false; }
+  bool is_disabled() const { return false; }
+  // Always connected so network::is_connected() keeps the API server accepting clients
+  bool is_connected() const { return true; }
+  void start_scanning();
+  const std::vector<WiFiScanResult> &get_scan_result() const { return this->scan_result_; }
+  void set_sta(const WiFiAP &ap);
+  void start_connecting(const WiFiAP &ap);
+  void clear_sta();
+  void save_wifi_sta(StringRef ssid, StringRef password);
+  // Called by network::util on any USE_WIFI build
+  const char *get_use_address() const { return "localhost"; }
+  network::IPAddresses get_ip_addresses() { return {}; }
+
+ protected:
+  std::vector<WiFiScanResult> scan_result_;
+};
+
+extern WiFiComponent *global_wifi_component;  // NOLINT(cppcoreguidelines-avoid-non-const-global-variables)
+
+}  // namespace esphome::wifi

tests/integration/fixtures/improv_serial_uart.yaml

@@ -0,0 +1,40 @@
+esphome:
+  # Short name keeps the device info payload under uart_mock's 64 byte log cap
+  name: improv-uart
+
+host:
+api:
+  actions:
+    - action: uart_inject
+      variables:
+        payload: int[]
+      then:
+        - uart_mock.inject_rx:
+            id: mock_uart
+            data: !lambda return std::vector<uint8_t>(payload.begin(), payload.end());
+
+logger:
+  level: DEBUG
+
+external_components:
+  - source:
+      type: local
+      path: EXTERNAL_COMPONENT_PATH
+
+# Host-only stub shadowing the real wifi component (see external_components/wifi)
+wifi:
+  ssid: TestNet
+  password: password1
+
+# Dummy uart entry so the uart component sources are part of the build; the
+# actual bus used by improv_serial is the uart_mock component below
+uart:
+  baud_rate: 115200
+  port: /dev/null
+
+uart_mock:
+  id: mock_uart
+  baud_rate: 115200
+
+improv_serial:
+  uart_id: mock_uart

tests/integration/fixtures/uart_mock_modbus_deprecated_write_buffer.yaml

@@ -0,0 +1,106 @@
+esphome:
+  name: uart-mock-modbus-dep-buffer
+
+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: reg10
+    type: uint16_t
+    initial_value: "0"
+
+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: 0x10
+        value_type: U_WORD
+        read_lambda: return id(reg10);
+        write_lambda: |-
+          id(reg10) = x;
+          return true;
+
+# A number whose write_lambda uses the DEPRECATED buffer parameter (fills `payload` with a legacy raw
+# frame as words: device address + function code + data) instead of the new item->write_* API. The write
+# must still land with its legacy semantics, and the one-time deprecation warning must fire only once per
+# entity no matter how many writes happen.
+number:
+  - platform: modbus_controller
+    modbus_controller_id: modbus_controller_1
+    name: "buf_number"
+    id: buf_number
+    address: 0x10
+    register_type: holding
+    value_type: U_WORD
+    min_value: 0
+    max_value: 1000
+    step: 1
+    write_lambda: |-
+      // Legacy raw frame as words: [addr 0x01 | fc 0x06], register 0x0010, value.
+      payload.push_back(0x0106);
+      payload.push_back(0x0010);
+      payload.push_back((uint16_t) x);
+      return {};
+
+# Reports the server-side register so the test can observe that the deprecated buffer write landed.
+sensor:
+  - platform: template
+    name: "written_value"
+    id: written_value
+    update_interval: 0.5s
+    lambda: "return id(reg10);"
+
+button:
+  - platform: template
+    name: "Start Scenario"
+    id: start_scenario_btn
+    # The test drives the writes via number_command; the mock is autostart.

tests/integration/fixtures/uart_mock_modbus_lambda_write.yaml

@@ -0,0 +1,97 @@
+esphome:
+  name: uart-mock-modbus-lambda-write
+
+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: reg30
+    type: uint16_t
+    initial_value: "0"
+
+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: 0x30
+        value_type: U_WORD
+        read_lambda: return id(reg30);
+        write_lambda: id(reg30) = x; return true;
+
+# A COIL-type switch (assumed_state, write-only) whose write_lambda ignores its own coil type and instead
+# drives a HOLDING-REGISTER write on the mock server through the entity itself: `item` IS the command, so
+# item->write_single_register() sends a register write from a coil entity (cross-type). Returning nothing
+# (an empty optional) tells the write path the lambda already dispatched the frame - no default coil write.
+switch:
+  - platform: modbus_controller
+    modbus_controller_id: modbus_controller_1
+    name: "cross_switch"
+    register_type: coil
+    address: 0x00
+    assumed_state: true
+    write_lambda: |-
+      item->write_single_register(0x30, x ? 1234 : 0);
+      return {};
+
+sensor:
+  - platform: modbus_controller
+    modbus_controller_id: modbus_controller_1
+    name: "reg_30"
+    address: 0x30
+    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)

tests/integration/fixtures/uart_mock_modbus_register_offset.yaml

@@ -0,0 +1,138 @@
+esphome:
+  name: uart-mock-modbus-reg-offset
+
+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: reg10
+    type: uint16_t
+    initial_value: "100"
+  - id: reg11
+    type: uint16_t
+    initial_value: "200"
+  - id: reg12
+    type: uint16_t
+    initial_value: "300"
+  - id: reg13
+    type: uint16_t
+    initial_value: "0xABCD"
+
+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: 0x10
+        value_type: U_WORD
+        read_lambda: return id(reg10);
+        write_lambda: id(reg10) = x; return true;
+      - address: 0x11
+        value_type: U_WORD
+        read_lambda: return id(reg11);
+        write_lambda: id(reg11) = x; return true;
+      - address: 0x12
+        value_type: U_WORD
+        read_lambda: return id(reg12);
+        write_lambda: id(reg12) = x; return true;
+      - address: 0x13
+        value_type: U_WORD
+        read_lambda: return id(reg13);
+        write_lambda: id(reg13) = x; return true;
+
+# A holding-register switch at 0x10 with a 2-BYTE offset. offset is byte-based, so the write must target
+# register 0x10 + 2/2 = 0x11. The old (pre-fix) behavior folded offset into the address as a register
+# count, hitting 0x12 instead. assumed_state keeps the switch write-only so it does not read any register.
+switch:
+  - platform: modbus_controller
+    modbus_controller_id: modbus_controller_1
+    name: "offset_switch"
+    register_type: holding
+    address: 0x10
+    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
+  # 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
+    modbus_controller_id: modbus_controller_1
+    name: "read_offset_switch"
+    register_type: holding
+    address: 0x10
+    offset: 6
+    bitmask: 0x1
+
+sensor:
+  - platform: modbus_controller
+    modbus_controller_id: modbus_controller_1
+    name: "reg_10"
+    address: 0x10
+    register_type: holding
+    value_type: U_WORD
+  - platform: modbus_controller
+    modbus_controller_id: modbus_controller_1
+    name: "reg_11"
+    address: 0x11
+    register_type: holding
+    value_type: U_WORD
+  - platform: modbus_controller
+    modbus_controller_id: modbus_controller_1
+    name: "reg_12"
+    address: 0x12
+    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)

tests/integration/log_utils.py

@@ -0,0 +1,43 @@
+"""Helpers for asserting on log output in integration tests."""
+
+from __future__ import annotations
+
+import asyncio
+
+
+class LineWaiter:
+    """Collects log lines and lets a test await one containing all needles.
+
+    Pass ``callback`` as ``run_compiled``'s ``line_callback``; the callback runs
+    on the test's own event loop, so futures are resolved directly. Only one
+    ``wait_for`` may be outstanding at a time (tests await sequentially).
+    """
+
+    def __init__(self) -> None:
+        self.lines: list[str] = []
+        self._needles: tuple[str, ...] = ()
+        self._future: asyncio.Future | None = None
+
+    def callback(self, line: str) -> None:
+        self.lines.append(line)
+        if (
+            self._future is not None
+            and not self._future.done()
+            and all(n in line for n in self._needles)
+        ):
+            self._future.set_result(line)
+            self._future = None
+
+    async def wait_for(self, *needles: str, timeout: float = 10.0) -> str:
+        """Return the first line, past or future, containing every needle."""
+        for line in self.lines:
+            if all(n in line for n in needles):
+                return line
+        assert self._future is None or self._future.done(), "concurrent wait_for"
+        self._needles = needles
+        self._future = asyncio.get_running_loop().create_future()
+        try:
+            return await asyncio.wait_for(self._future, timeout)
+        finally:
+            self._future = None
+            self._needles = ()

tests/integration/test_improv_serial_uart.py

@@ -0,0 +1,140 @@
+"""Integration test for improv_serial over a mocked UART bus.
+
+Drives the improv serial protocol end to end on the host platform:
+the fixture wires improv_serial to a uart_mock bus and shadows the wifi
+component with a host stub. The test injects improv frames through an API
+action and asserts on the framed responses that uart_mock logs as TX lines.
+
+Covered:
+  1. Get Current State reports AUTHORIZED
+  2. Get Device Info returns the firmware/device info RPC response
+  3. Get Wi-Fi Networks returns deduplicated scan results and a terminator
+  4. Wi-Fi Settings provisions: saves credentials and reports PROVISIONED
+"""
+
+from __future__ import annotations
+
+import pytest
+
+from .log_utils import LineWaiter
+from .types import APIClientConnectedFactory, RunCompiledFunction
+
+# Improv serial framing (improv_serial_component.h)
+IMPROV_HEADER = b"IMPROV"
+IMPROV_VERSION = 1
+TYPE_CURRENT_STATE = 0x01
+TYPE_RPC = 0x03
+TYPE_RPC_RESPONSE = 0x04
+
+# improv::Command values
+CMD_GET_CURRENT_STATE = 0x02
+CMD_GET_DEVICE_INFO = 0x03
+CMD_GET_WIFI_NETWORKS = 0x04
+CMD_WIFI_SETTINGS = 0x01
+
+
+def build_rpc_frame(command: int, data: bytes = b"") -> list[int]:
+    """Build a full improv serial frame carrying one RPC command."""
+    payload = bytes([command, len(data)]) + data
+    frame = IMPROV_HEADER + bytes([IMPROV_VERSION, TYPE_RPC, len(payload)]) + payload
+    checksum = sum(frame) & 0xFF
+    return list(frame + bytes([checksum]) + b"\n")
+
+
+def state_frame_hex(state: int) -> str:
+    """Full 12 byte current-state frame as hex, checksum and newline included."""
+    frame = IMPROV_HEADER + bytes([IMPROV_VERSION, TYPE_CURRENT_STATE, 1, state])
+    checksum = sum(frame) & 0xFF
+    return ":".join(f"{b:02X}" for b in frame + bytes([checksum]) + b"\n")
+
+
+def rpc_footer_hex(payload: bytes) -> str:
+    """Checksum and newline footer written after an RPC response payload."""
+    header = IMPROV_HEADER + bytes([IMPROV_VERSION, TYPE_RPC_RESPONSE, len(payload)])
+    checksum = (sum(header) + sum(payload)) & 0xFF
+    return f"{checksum:02X}:0A"
+
+
+def wifi_settings_data(ssid: str, password: str) -> bytes:
+    ssid_b = ssid.encode()
+    pass_b = password.encode()
+    return bytes([len(ssid_b)]) + ssid_b + bytes([len(pass_b)]) + pass_b
+
+
+def hex_of(text: str) -> str:
+    """Colon separated uppercase hex as logged by format_hex_pretty."""
+    return ":".join(f"{b:02X}" for b in text.encode())
+
+
+@pytest.mark.asyncio
+async def test_improv_serial_uart(
+    yaml_config: str,
+    run_compiled: RunCompiledFunction,
+    api_client_connected: APIClientConnectedFactory,
+) -> None:
+    waiter = LineWaiter()
+
+    async with (
+        run_compiled(yaml_config, line_callback=waiter.callback),
+        api_client_connected() as client,
+    ):
+        _entities, services = await client.list_entities_services()
+        inject = next(s for s in services if s.name == "uart_inject")
+
+        # 1. Get Current State: expect the complete current-state frame reporting
+        # AUTHORIZED (0x02), checksum and newline included
+        await client.execute_service(
+            inject, {"payload": build_rpc_frame(CMD_GET_CURRENT_STATE)}
+        )
+        await waiter.wait_for("uart_mock", f"TX 12 bytes: {state_frame_hex(0x02)}")
+
+        # 2. Get Device Info: the always logged 9 byte response header, then the
+        # payload with the firmware name (must stay under uart_mock's 64 byte
+        # hex dump cap or the payload line reads "too large to log")
+        await client.execute_service(
+            inject, {"payload": build_rpc_frame(CMD_GET_DEVICE_INFO)}
+        )
+        await waiter.wait_for("uart_mock", "TX 9 bytes: 49:4D:50:52:4F:56:01:04")
+        await waiter.wait_for("uart_mock", "TX ", hex_of("ESPHome"))
+
+        # 3. Get Wi-Fi Networks: stub scan has TestNet twice (dedup keeps the
+        # stronger), OpenNet, and a hidden entry (filtered). Expect one response
+        # per visible network plus the empty terminator.
+        await client.execute_service(
+            inject, {"payload": build_rpc_frame(CMD_GET_WIFI_NETWORKS)}
+        )
+        await waiter.wait_for("uart_mock", hex_of("TestNet"))
+        await waiter.wait_for("uart_mock", hex_of("OpenNet"))
+        # Terminator: all three writes of the response frame; 9 byte header,
+        # payload [0x04, 0x00, 0x00], then the checksum and newline footer
+        await waiter.wait_for("uart_mock", "TX 9 bytes: 49:4D:50:52:4F:56:01:04:03")
+        await waiter.wait_for("uart_mock", "TX 3 bytes: 04:00:00")
+        await waiter.wait_for(
+            "uart_mock", f"TX 2 bytes: {rpc_footer_hex(bytes([0x04, 0x00, 0x00]))}"
+        )
+        testnet_count = sum(
+            1
+            for line in waiter.lines
+            if "uart_mock" in line and "TX " in line and hex_of("TestNet") in line
+        )
+        assert testnet_count == 1, (
+            f"Duplicate scan entry not deduplicated: {testnet_count} TestNet responses"
+        )
+
+        # 4. Wi-Fi Settings: stub connects immediately; expect the credentials
+        # saved, the PROVISIONED state frame (0x04), and the settings response
+        await client.execute_service(
+            inject,
+            {
+                "payload": build_rpc_frame(
+                    CMD_WIFI_SETTINGS, wifi_settings_data("NewNet", "secret123")
+                )
+            },
+        )
+        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]))}"
+        )

tests/integration/test_uart_mock_modbus.py

@@ -24,7 +24,7 @@ from dataclasses import dataclass
 from aioesphomeapi import ButtonInfo, NumberInfo, SwitchInfo
 import pytest
 
-from .state_utils import SensorTracker, find_entity
+from .state_utils import SensorTracker, find_entity, wait_for_state
 from .types import APIClientConnectedFactory, RunCompiledFunction
 
 
@@ -965,3 +965,148 @@ async def test_uart_mock_modbus_client_read_write(
         await tracker.setup_and_start_scenario(client)
         await tracker.await_all(futures)
         _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,
+    run_compiled: RunCompiledFunction,
+    api_client_connected: APIClientConnectedFactory,
+) -> None:
+    """Test that a byte offset on a holding-register write is byte-accurate.
+
+    `offset` is a byte offset, so a holding-register write at address 0x10 with offset: 2 must target
+    register 0x10 + 2/2 = 0x11. The pre-fix behavior folded the byte offset into the address as a register
+    count (0x10 + 2 = 0x12). The switch is assumed_state (write-only), so reg_11 turning 0xFFFF pins the
+    fix; had the write landed on 0x12 the wait would time out and reg_12 would change instead.
+    """
+
+    tracker = SensorTracker(["reg_10", "reg_11", "reg_12"])
+    initial = tracker.expect_all({"reg_10": 100, "reg_11": 200, "reg_12": 300})
+    wrote_11 = tracker.expect("reg_11", 65535)
+
+    async with (
+        run_compiled(yaml_config),
+        api_client_connected() as client,
+    ):
+        entities = await tracker.setup_and_start_scenario(client)
+        await tracker.await_all(initial, timeout=4.0)
+
+        switch = find_entity(entities, "offset_switch", SwitchInfo)
+        assert switch is not None, "offset_switch not found"
+        client.switch_command(switch.key, True)
+
+        # reg_11 (0x10 + offset 2/2) must receive the write; if the write went to 0x12 this times out.
+        await tracker.await_change(wrote_11, "reg_11", timeout=4.0)
+        # And 0x12 (the pre-fix register-offset target) must be untouched.
+        assert tracker.sensor_states["reg_12"][-1] == 300, (
+            "reg_12 (0x12) should be untouched - offset is byte-based, so the write targets 0x11; "
+            f"got {tracker.sensor_states['reg_12']}"
+        )
+
+        # Read path: read_offset_switch has byte offset 6. Post-fix the switch folds the whole registers
+        # into its address (0x10 + 6/2 = 0x13, residual byte 0) and joins the 0x10..0x13 range, so the
+        # read lands in-bounds on 0xABCD (bit 0 set) -> ON. Pre-fix the whole byte offset folded into the
+        # address (0x16); the server answers ILLEGAL_DATA_ADDRESS there and the switch never publishes.
+        read_switch = find_entity(entities, "read_offset_switch", SwitchInfo)
+        assert read_switch is not None, "read_offset_switch not found"
+        # The ON transition happened at the first poll and 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) == read_switch.key
+                and getattr(s, "state", None) is True
+            ),
+            timeout=6.0,
+        )
+
+
+@pytest.mark.asyncio
+async def test_uart_mock_modbus_lambda_write(
+    yaml_config: str,
+    run_compiled: RunCompiledFunction,
+    api_client_connected: APIClientConnectedFactory,
+) -> None:
+    """Test a write_lambda that drives the write through the entity itself (item is the command).
+
+    `cross_switch` is a coil-type switch whose write_lambda ignores its own type and calls
+    item->write_single_register(0x30, ...) - a register write issued from a coil entity. The lambda
+    returns an empty optional, so the write path detects the lambda already dispatched a frame and does
+    not fall back to the default coil write. Success is reg_30 reading back the value the lambda wrote,
+    which proves both the new item->write_* path and cross-type flexibility.
+    """
+
+    tracker = SensorTracker(["reg_30"])
+    initial = tracker.expect("reg_30", 0)
+    wrote_30 = tracker.expect("reg_30", 1234)
+
+    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_30", timeout=4.0)
+
+        switch = find_entity(entities, "cross_switch", SwitchInfo)
+        assert switch is not None, "cross_switch not found"
+        client.switch_command(switch.key, True)
+
+        # The coil switch's lambda wrote register 0x30 via item->write_single_register(); reg_30 must
+        # read back 1234. If the entity-as-command dispatch were broken, no register write would go out
+        # and this would time out.
+        await tracker.await_change(wrote_30, "reg_30", timeout=4.0)
+
+
+@pytest.mark.asyncio
+async def test_uart_mock_modbus_deprecated_write_buffer(
+    yaml_config: str,
+    run_compiled: RunCompiledFunction,
+    api_client_connected: APIClientConnectedFactory,
+) -> None:
+    """Test the deprecated write_lambda buffer path still works, and warns once per entity.
+
+    buf_number's write_lambda fills the old `payload` buffer with a legacy raw frame as words (device
+    address + function code + data) and returns {} instead of calling item->write_*. Both writes must
+    land - a filled buffer is sent, as the docs have always described - and the one-time deprecation
+    warning must fire exactly once per entity regardless of how many writes happen.
+    """
+
+    warn_count = 0
+
+    def line_callback(line: str) -> None:
+        nonlocal warn_count
+        if "write_lambda buffer" in line:
+            warn_count += 1
+
+    tracker = SensorTracker(["written_value"])
+
+    async with (
+        run_compiled(yaml_config, line_callback=line_callback),
+        api_client_connected() as client,
+    ):
+        entities = await tracker.setup_and_start_scenario(client)
+        number = find_entity(entities, "buf_number", NumberInfo)
+        assert number is not None, "buf_number not found"
+
+        # First write via the deprecated buffer path.
+        client.number_command(number.key, 111)
+        await tracker.await_change(
+            tracker.expect("written_value", 111), "written_value", timeout=4.0
+        )
+        # Second write: lands too, but must not warn again (warn-once per entity).
+        client.number_command(number.key, 222)
+        await tracker.await_change(
+            tracker.expect("written_value", 222), "written_value", timeout=4.0
+        )
+
+    assert warn_count == 1, (
+        f"deprecation warning should fire exactly once per entity, got {warn_count}"
+    )

tests/script/test_platformio_install_deps.py

@@ -0,0 +1,649 @@
+"""Tests for script/platformio_install_deps.py."""
+
+from argparse import Namespace
+import importlib.util
+import inspect
+from pathlib import Path
+import shutil
+import sys
+from types import SimpleNamespace
+from unittest.mock import MagicMock, patch
+
+from platformio import fs
+from platformio.cache import ContentCache
+from platformio.exception import InvalidJSONFile
+from platformio.package.manager._install import PackageManagerInstallMixin
+from platformio.package.manager.base import BasePackageManager
+from platformio.package.manager.library import LibraryPackageManager
+from platformio.package.manager.tool import ToolPackageManager
+from platformio.package.meta import PackageCompatibility, PackageItem, PackageSpec
+import pytest
+from semantic_version import Version
+
+_SCRIPT = Path(__file__).parents[2] / "script" / "platformio_install_deps.py"
+
+
+def _load_script():
+    spec = importlib.util.spec_from_file_location("platformio_install_deps", _SCRIPT)
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    # The real ContentCache would create dirs under the user's core dir
+    module.ContentCache = lambda *_: None
+    return module
+
+
+def test_spec_key_collapses_destinations() -> None:
+    """Two specs delivering one package share a directory and one key."""
+    mod = _load_script()
+    assert mod.spec_key("esphome/noise-c @ 0.1.21") == "noise-c"
+    assert mod.spec_key("esphome/noise-c@0.1.21") == "noise-c"
+    assert mod.spec_key("ESP32Async/AsyncTCP @ ^3.4.10") == mod.spec_key(
+        "esp32async/asynctcp @ 3.5.0"
+    )
+    url = "https://github.com/pioarduino/platform-espressif32/releases/download/{v}/platform-espressif32.zip"
+    assert mod.spec_key(url.format(v="55.03.311")) == mod.spec_key(
+        url.format(v="54.03.20")
+    )
+
+
+def test_parse_specs_and_cli_args(tmp_path: Path) -> None:
+    """Parsing skips unpinned and interpolated entries; the CLI rebuild
+    keeps the original flag pairing."""
+    ini = tmp_path / "platformio.ini"
+    ini.write_text(
+        "[env:a]\n"
+        "platform = fake/platform@1\n"
+        "lib_deps =\n"
+        "    esphome/noise-c @ 0.1.21\n"
+        "    ${common.lib_deps}\n"
+        "    internal_lib\n"
+        "[env:b]\n"
+        "lib_deps =\n"
+        "    esphome/noise-c @ 0.1.21\n"
+    )
+    mod = _load_script()
+    args = Namespace(libraries=True, platforms=True, tools=False)
+    libs, platforms, tools = mod.parse_specs(str(ini), args)
+    # exact-string duplicates collapse; distinct version pins survive
+    assert libs == ["esphome/noise-c @ 0.1.21"]
+    assert platforms == ["fake/platform@1"]
+    assert tools == []
+    assert mod.build_cli_args(libs, platforms, tools) == [
+        "-l",
+        "esphome/noise-c @ 0.1.21",
+        "-p",
+        "fake/platform@1",
+    ]
+
+
+class _FakeManager:
+    """Scripted manager_cls: records installs, raises on demand."""
+
+    installed: set = set()
+    fail: set = set()
+    calls: list = []
+    lock_events: list = []
+    base_dir: str = ""  # per-test tmp base; set by _reset_fake
+
+    def __init__(self, package_dir) -> None:
+        assert package_dir is None
+
+    @staticmethod
+    def _key(spec) -> str:
+        return spec if isinstance(spec, str) else str(spec)
+
+    def get_package(self, spec):
+        if self._key(spec) in self.installed:
+            return SimpleNamespace(path="/tmp/fake-pkg", spec=self._key(spec))
+        return None
+
+    def memcache_reset(self) -> None:
+        type(self).resets = getattr(type(self), "resets", 0) + 1
+
+    @property
+    def package_dir(self) -> str:
+        return str(Path(type(self).base_dir) / "packages")
+
+    def get_download_dir(self) -> str:
+        return str(Path(type(self).base_dir) / "downloads")
+
+    def get_tmp_dir(self) -> str:
+        return str(Path(type(self).base_dir) / "tmp")
+
+    def lock(self) -> None:
+        type(self).lock_events.append("lock")
+
+    def unlock(self) -> None:
+        type(self).lock_events.append("unlock")
+
+    def _install(self, spec, skip_dependencies, compatibility=None):
+        assert skip_dependencies is True
+        if self._key(spec) in self.fail:
+            raise RuntimeError("boom")
+        type(self).calls.append(spec)
+        type(self).compat_calls.append((self._key(spec), compatibility))
+        type(self).installed.add(self._key(spec))  # atomic under the GIL
+
+    def get_pkg_dependencies(self, pkg):
+        return getattr(type(self), "deps", {}).get(pkg.spec)
+
+    dependency_to_spec = staticmethod(BasePackageManager.dependency_to_spec)
+
+
+def _reset_fake(base_dir: str = "", **kwargs) -> type:
+    # A fresh subclass per test: nothing leaks between tests through the
+    # class-level scripted state
+    return type(
+        "_ScriptedManager",
+        (_FakeManager,),
+        {
+            "base_dir": base_dir,
+            "installed": kwargs.get("installed", set()),
+            "fail": kwargs.get("fail", set()),
+            "calls": [],
+            "compat_calls": [],
+            "lock_events": [],
+        },
+    )
+
+
+def test_parallel_install_empty_specs_is_a_no_op(tmp_path: Path) -> None:
+    mod = _load_script()
+    cls = _reset_fake(str(tmp_path))
+    mod.parallel_install(cls, [])
+    assert cls.calls == [] and cls.lock_events == []
+
+
+def test_parallel_install_behavior(tmp_path: Path) -> None:
+    """Duplicates collapse to one install, installed specs are filtered,
+    URL specs stay out of the wave, and the lock wraps the pool."""
+    mod = _load_script()
+    cls = _reset_fake(str(tmp_path), installed={"esphome/already @ 1.0"})
+    mod.parallel_install(
+        cls,
+        [
+            "esphome/noise-c @ 0.1.21",
+            "esphome/noise-c @ 0.1.21",
+            "esphome/already @ 1.0",
+            "https://x/framework.tar.xz",
+        ],
+    )
+    assert cls.calls == ["esphome/noise-c @ 0.1.21"]
+    assert cls.lock_events == ["lock", "unlock"]
+
+
+def test_parallel_install_failure_cleans_torn_destination(
+    tmp_path: Path, capsys
+) -> None:
+    """A failed install resets the memcache, removes what get_package can
+    see, and reports; the others still install."""
+    mod = _load_script()
+    cls = _reset_fake(str(tmp_path), fail={"esphome/bad @ 1.0"})
+
+    removed = []
+
+    torn = str(tmp_path / "packages" / "torn-pkg")  # never created; only rmtree'd
+
+    def get_package(self, spec):
+        if spec == "esphome/bad @ 1.0" and getattr(cls, "resets", 0):
+            return SimpleNamespace(path=torn, spec=spec)
+        return _FakeManager.get_package(self, spec)
+
+    cls.get_package = get_package  # throwaway subclass; nothing to restore
+    with patch.object(mod.fs, "rmtree", side_effect=removed.append):
+        mod.parallel_install(cls, ["esphome/bad @ 1.0", "esphome/good @ 1.0"])
+    assert "esphome/good @ 1.0" in cls.calls
+    assert removed == [torn]
+    out = capsys.readouterr().out
+    assert "Pre-install of esphome/bad @ 1.0 failed" in out
+    assert "Pre-install failed for 1 of 2 package(s)" in out
+
+
+def test_parallel_install_runs_dependency_waves(tmp_path: Path) -> None:
+    """Dependencies of wave-installed packages install in a second wave,
+    deduped by name; name-only platform libs stay with the serial pass."""
+    mod = _load_script()
+    cls = _reset_fake(str(tmp_path))
+    cls.deps = {
+        "esphome/noise-c @ 0.1.21": [
+            {"owner": "esphome", "name": "libsodium", "version": "^1.0"},
+            {"name": "SPI"},
+        ],
+        "esphome/wg @ 1.0": [
+            {"owner": "esphome", "name": "libsodium", "version": "^1.0"},
+        ],
+    }
+    mod.parallel_install(cls, ["esphome/noise-c @ 0.1.21", "esphome/wg @ 1.0"])
+    assert len(cls.calls) == 3  # the shared dep installs exactly once
+    assert {mod.spec_key(c) for c in cls.calls} == {"noise-c", "wg", "libsodium"}
+    # Wave-1 strings carry no compatibility; the dependency wave does
+    compats = dict(cls.compat_calls)
+    assert compats["esphome/noise-c @ 0.1.21"] is None
+    dep_compat = next(v for k, v in cls.compat_calls if "libsodium" in k)
+    assert dep_compat is not None  # mirrors pio's install_dependency
+
+
+def test_dependency_wave_excludes_url_specs(tmp_path: Path) -> None:
+    """A dependency pinned to a URL surfaces as spec.uri; it must stay out
+    of the wave like string URL specs do."""
+    mod = _load_script()
+    cls = _reset_fake(str(tmp_path))
+    cls.deps = {
+        "esphome/noise-c @ 0.1.21": [
+            {"name": "vendored", "version": "https://github.com/x/y.git"},
+        ],
+    }
+    mod.parallel_install(cls, ["esphome/noise-c @ 0.1.21"])
+    assert {mod.spec_key(c) for c in cls.calls} == {"noise-c"}
+
+
+def test_failed_cleanup_fails_the_build(tmp_path: Path) -> None:
+    """A torn destination still on disk after rmtree must fail the build:
+    fs.rmtree never raises (its onexc handler prints), so only the
+    destination's absence proves the cleanup worked."""
+    mod = _load_script()
+    cls = _reset_fake(str(tmp_path), fail={"esphome/bad @ 1.0"})
+    torn = tmp_path / "packages" / "torn-pkg"
+    torn.mkdir(parents=True)
+
+    def get_package(self, spec):
+        if getattr(cls, "resets", 0):
+            return SimpleNamespace(path=str(torn), spec=spec)
+        return None
+
+    cls.get_package = get_package  # throwaway subclass; nothing to restore
+    with (
+        patch.object(mod.fs, "rmtree", lambda path: None),  # onexc swallowed
+        pytest.raises(mod.CleanupError, match="could not remove"),
+    ):
+        mod.parallel_install(cls, ["esphome/bad @ 1.0"])
+    assert cls.lock_events == ["lock", "unlock"]  # still released
+
+
+def test_unverifiable_torn_destination_fails_the_build(tmp_path: Path) -> None:
+    """When the scan fails, the spec's own .piopm decides: an unremovable
+    leftover fails the build."""
+    mod = _load_script()
+    cls = _reset_fake(str(tmp_path), fail={"esphome/bad @ 1.0"})
+    dest = Path(cls.base_dir) / "packages" / "bad"
+    dest.mkdir(parents=True)
+    (dest / ".piopm").write_text('{"spec": {"owner": "esphome", "name": "bad"}}')
+
+    def bad_reset(self):
+        raise OSError("scan broken")
+
+    cls.memcache_reset = bad_reset
+    with (
+        patch.object(mod.fs, "rmtree", lambda path: None),  # onexc swallowed
+        pytest.raises(mod.CleanupError, match="could not remove"),
+    ):
+        mod.parallel_install(cls, ["esphome/bad @ 1.0"])
+
+
+def test_unverifiable_scan_without_leftover_degrades(tmp_path: Path, capsys) -> None:
+    """A failing scan with no destination on disk is never a build
+    failure blaming this spec."""
+    mod = _load_script()
+    cls = _reset_fake(str(tmp_path), fail={"esphome/bad @ 1.0"})
+    resets = {"n": 0}
+
+    def bad_reset(self):
+        # Fail clean_torn's reset; the coordinator's later reset works
+        resets["n"] += 1
+        if resets["n"] <= 1:
+            raise OSError("scan broken")
+
+    cls.memcache_reset = bad_reset
+    mod.parallel_install(cls, ["esphome/bad @ 1.0"])
+    assert "No resolvable destination to clean" in capsys.readouterr().out
+
+
+def test_unresolvable_torn_destination_is_printed(tmp_path: Path, capsys) -> None:
+    """A failed install with no resolvable package prints, so an invisible
+    torn directory is at least traceable."""
+    mod = _load_script()
+    cls = _reset_fake(str(tmp_path), fail={"esphome/bad @ 1.0"})
+    mod.parallel_install(cls, ["esphome/bad @ 1.0"])
+    assert "No resolvable destination to clean" in capsys.readouterr().out
+
+
+def test_unparsable_torn_destination_is_removed(tmp_path: Path, capsys) -> None:
+    """A torn dir get_package cannot resolve but whose .piopm names the
+    spec is removed instead of surviving into the serial pass."""
+    mod = _load_script()
+    cls = _reset_fake(str(tmp_path), fail={"esphome/bad @ 1.0"})
+    dest = Path(cls.base_dir) / "packages" / "bad"
+    dest.mkdir(parents=True)
+    (dest / ".piopm").write_text('{"spec": {"owner": "esphome", "name": "bad"}}')
+
+    with patch.object(mod.fs, "rmtree", shutil.rmtree):
+        mod.parallel_install(cls, ["esphome/bad @ 1.0"])
+    assert not dest.exists()
+    assert "Removed torn destination" in capsys.readouterr().out
+
+
+def test_parse_specs_tools_branch(tmp_path: Path) -> None:
+    """platform_packages parsing keeps owner'd tools and rewrites github
+    URL pins to bare URLs the wave then skips via parsed.uri."""
+    mod = _load_script()
+    ini = tmp_path / "platformio.ini"
+    ini.write_text(
+        "[env:t]\n"
+        "platform_packages =\n"
+        "    ${common.platform_packages}\n"
+        "    platformio/tool-scons@~4.40801.0\n"
+        "    framework-arduinopico@https://github.com/earlephilhower/arduino-pico/releases/download/6.0.0/rp2040-6.0.0.zip\n"
+    )
+    args = Namespace(libraries=False, platforms=False, tools=True)
+    libs, platforms, tools = mod.parse_specs(str(ini), args)
+    assert libs == [] and platforms == []
+    assert tools == [
+        "platformio/tool-scons@~4.40801.0",
+        "https://github.com/earlephilhower/arduino-pico/releases/download/6.0.0/rp2040-6.0.0.zip",
+    ]
+    assert mod.build_cli_args([], [], tools)[:2] == ["-t", tools[0]]
+
+
+def test_warm_store_still_walks_dependencies(tmp_path: Path) -> None:
+    """Already-installed top-level packages still feed the dependency
+    wave; a warm store can be missing a transitive dep."""
+    mod = _load_script()
+    cls = _reset_fake(str(tmp_path), installed={"esphome/noise-c @ 0.1.21"})
+    cls.deps = {
+        "esphome/noise-c @ 0.1.21": [
+            {"owner": "esphome", "name": "libsodium", "version": "^1.0"},
+        ],
+    }
+    mod.parallel_install(cls, ["esphome/noise-c @ 0.1.21"])
+    assert [mod.spec_key(c) for c in cls.calls] == ["libsodium"]
+
+
+def test_worker_system_exit_still_cleans(tmp_path: Path, capsys) -> None:
+    """A worker SystemExit runs the torn cleanup before propagating; the
+    serial pass must never trust its leftovers."""
+    mod = _load_script()
+    cls = _reset_fake(str(tmp_path))
+    torn = tmp_path / "packages" / "torn-pkg"
+    torn.mkdir(parents=True)
+
+    def exiting_install(self, spec, skip_dependencies, compatibility=None):
+        raise SystemExit(0)
+
+    def get_package(self, spec):
+        if getattr(cls, "resets", 0):
+            return SimpleNamespace(path=str(torn), spec=spec)
+        return None
+
+    cls._install = exiting_install
+    cls.get_package = get_package
+
+    def real_rmtree(path):
+        Path(path).rmdir()
+
+    with (
+        patch.object(mod.fs, "rmtree", real_rmtree),
+        pytest.raises(SystemExit),
+    ):
+        mod.parallel_install(cls, ["esphome/bad @ 1.0"])
+    assert not torn.exists()
+
+
+def test_unlock_failure_is_fatal(tmp_path: Path) -> None:
+    """A failed unlock must fail the build: the serial pass in another
+    process would block on the held flock."""
+    mod = _load_script()
+    cls = _reset_fake(str(tmp_path))
+
+    def bad_unlock(self):
+        raise OSError("flock broke")
+
+    cls.unlock = bad_unlock
+    with pytest.raises(mod.LockReleaseError, match="manager lock"):
+        mod.parallel_install(cls, ["esphome/good @ 1.0"])
+
+
+def test_unlock_failure_keeps_inflight_error_as_context(tmp_path: Path) -> None:
+    """An in-flight CleanupError stays attached when the unlock fault
+    takes over the raise."""
+    mod = _load_script()
+    cls = _reset_fake(str(tmp_path), fail={"esphome/bad @ 1.0"})
+    torn = tmp_path / "packages" / "bad"
+    torn.mkdir(parents=True)
+
+    def get_package(self, spec):
+        if getattr(cls, "resets", 0):
+            return SimpleNamespace(path=str(torn), spec=spec)
+        return None
+
+    def bad_unlock(self):
+        raise OSError("flock broke")
+
+    cls.get_package = get_package
+    cls.unlock = bad_unlock
+    with (
+        patch.object(mod.fs, "rmtree", lambda path: None),  # leaves torn
+        pytest.raises(mod.LockReleaseError) as err,
+    ):
+        mod.parallel_install(cls, ["esphome/bad @ 1.0"])
+    assert isinstance(err.value.__cause__.__context__, mod.CleanupError)
+
+
+def test_chdir_failure_does_not_fail_the_wave(tmp_path: Path, monkeypatch) -> None:
+    """A lost cwd is suppressed: further waves may misbehave and fall to
+    the serial pass, whose cwd is pinned."""
+    mod = _load_script()
+    cls = _reset_fake(str(tmp_path))
+    monkeypatch.setattr(mod.os, "chdir", MagicMock(side_effect=OSError("gone")))
+    mod.parallel_install(cls, ["esphome/good @ 1.0"])
+    assert cls.calls == ["esphome/good @ 1.0"]
+
+
+def test_piopm_match_removes_manifest_named_torn_dir(tmp_path: Path, capsys) -> None:
+    """A torn dir named by its manifest (not the registry spec) is found
+    through its .piopm and removed."""
+    mod = _load_script()
+    cls = _reset_fake(str(tmp_path), fail={"esphome/bad @ 1.0"})
+    torn = tmp_path / "packages" / "ManifestName"
+    torn.mkdir(parents=True)
+    (torn / ".piopm").write_text('{"spec": {"owner": "esphome", "name": "bad"}}')
+    innocent = tmp_path / "packages" / "innocent"
+    innocent.mkdir()
+    (innocent / ".piopm").write_text('{"spec": {"owner": "o", "name": "other"}}')
+    with patch.object(mod.fs, "rmtree", shutil.rmtree):
+        mod.parallel_install(cls, ["esphome/bad @ 1.0"])
+    assert not torn.exists()
+    assert innocent.exists()  # another package's valid metadata survives
+    assert "Removed torn destination" in capsys.readouterr().out
+
+
+def test_unscannable_package_dir_fails_the_build(tmp_path: Path) -> None:
+    """A storage dir the cleanup cannot scan is not proof of cleanliness."""
+    mod = _load_script()
+    cls = _reset_fake(str(tmp_path), fail={"esphome/bad @ 1.0"})
+    real_iterdir = Path.iterdir
+
+    def broken_iterdir(self):
+        if self.name == "packages":
+            raise PermissionError("denied")
+        return real_iterdir(self)
+
+    with (
+        patch.object(Path, "iterdir", broken_iterdir),
+        pytest.raises(mod.CleanupError, match="cleanup failed"),
+    ):
+        mod.parallel_install(cls, ["esphome/bad @ 1.0"])
+
+
+def test_stray_file_in_package_dir_is_ignored(tmp_path: Path) -> None:
+    """A plain file (or a pio-link) beside the packages is skipped by
+    pio's own scan and must never hard-fail the build."""
+    mod = _load_script()
+    cls = _reset_fake(str(tmp_path), fail={"esphome/bad @ 1.0"})
+    (tmp_path / "packages").mkdir(parents=True)
+    (tmp_path / "packages" / "stray.pio-link").write_text("x")
+    (tmp_path / "packages" / "no-metadata").mkdir()  # pio overwrites these
+    mod.parallel_install(cls, ["esphome/bad @ 1.0"])
+    assert (tmp_path / "packages" / "stray.pio-link").exists()
+    assert (tmp_path / "packages" / "no-metadata").exists()
+
+
+def test_unreadable_piopm_dir_is_removed(tmp_path: Path) -> None:
+    """A persistently corrupt .piopm under this spec's own name would
+    crash pio's storage scan; the dir is removed rather than left to
+    break the serial pass."""
+    mod = _load_script()
+    cls = _reset_fake(str(tmp_path), fail={"esphome/bad @ 1.0"})
+    torn = tmp_path / "packages" / "bad"
+    torn.mkdir(parents=True)
+    (torn / ".piopm").write_text("{not json")
+    mod.parallel_install(cls, ["esphome/bad @ 1.0"])
+    assert not torn.exists()
+
+
+def test_unreadable_piopm_under_other_name_survives(tmp_path: Path) -> None:
+    """A corrupt .piopm in another package's dir may be a worker mid-copy;
+    a failing spec must not remove a directory it does not own."""
+    mod = _load_script()
+    cls = _reset_fake(str(tmp_path), fail={"esphome/bad @ 1.0"})
+    other = tmp_path / "packages" / "innocent"
+    other.mkdir(parents=True)
+    (other / ".piopm").write_text("{not json")
+    mod.parallel_install(cls, ["esphome/bad @ 1.0"])
+    assert other.exists()
+
+
+def test_unexpected_cleanup_class_becomes_cleanup_error(tmp_path: Path) -> None:
+    """Cleanup failures of any class fail the build; nothing may be
+    downgraded to the serial fallback over a torn directory."""
+    mod = _load_script()
+    cls = _reset_fake(str(tmp_path), fail={"esphome/bad @ 1.0"})
+
+    with (
+        patch.object(
+            mod, "piopm_matches", MagicMock(side_effect=ValueError("bad spec"))
+        ),
+        pytest.raises(mod.CleanupError, match="cleanup failed"),
+    ):
+        mod.parallel_install(cls, ["esphome/bad @ 1.0"])
+
+
+def test_main_cleanup_error_fails_before_generic_fallback(tmp_path: Path) -> None:
+    """A CleanupError must escape main's serial fallback: the clause order
+    decides whether a stuck torn package fails the image build."""
+    mod = _load_script()
+    ini = tmp_path / "platformio.ini"
+    ini.write_text("[env:t]\nlib_deps =\n    esphome/x @ 1.0\n")
+    with (
+        patch.object(
+            mod, "parallel_install", side_effect=mod.CleanupError("stuck torn pkg")
+        ),
+        patch.object(mod.subprocess, "check_call"),
+        patch.object(sys, "argv", ["platformio_install_deps.py", str(ini), "-l"]),
+        pytest.raises(mod.CleanupError),
+    ):
+        mod.main()
+
+
+def test_main_generic_failure_still_runs_serial_pass(tmp_path: Path) -> None:
+    """A non-CleanupError wave failure prints, dumps the traceback, and
+    still reaches the authoritative serial pass with the pinned cwd."""
+    mod = _load_script()
+    ini = tmp_path / "platformio.ini"
+    ini.write_text("[env:t]\nlib_deps =\n    esphome/x @ 1.0\n")
+    with (
+        patch.object(mod, "parallel_install", side_effect=RuntimeError("boom")),
+        patch.object(mod.subprocess, "check_call") as mock_call,
+        patch.object(sys, "argv", ["platformio_install_deps.py", str(ini), "-l"]),
+    ):
+        mod.main()
+    mock_call.assert_called_once()
+    args, kwargs = mock_call.call_args
+    assert args[0][:4] == ["platformio", "pkg", "install", "-g"]
+    assert "esphome/x @ 1.0" in args[0]
+    assert kwargs["cwd"] == Path.cwd()
+
+
+def test_content_cache_creates_its_dir(tmp_path: Path, monkeypatch) -> None:
+    """The cold-cache hardening relies on ContentCache.__init__ creating
+    the namespace dir; pin the side effect, not mere callability."""
+    monkeypatch.setenv("PLATFORMIO_CACHE_DIR", str(tmp_path / "cache"))
+    ContentCache("http")
+    assert (tmp_path / "cache" / "http").is_dir()
+
+
+def test_piopm_matches_without_name_matches_nothing(tmp_path: Path) -> None:
+    """A spec with no derivable name can never match a directory."""
+    mod = _load_script()
+    assert mod.piopm_matches(str(tmp_path), "") == []
+
+
+def test_unresolvable_spec_stays_out_of_the_wave(tmp_path: Path, capsys) -> None:
+    """A spec with no derivable name is left to the serial pass; a raw
+    string key would break the one-per-destination dedupe."""
+    mod = _load_script()
+    cls = _reset_fake(str(tmp_path))
+    nameless = PackageSpec(requirements="^1.0")
+    mod.parallel_install(cls, [nameless])
+    assert cls.calls == []
+    assert "Skipping unresolvable spec" in capsys.readouterr().out
+
+
+def test_parallel_install_unlocks_when_pool_fails(tmp_path: Path) -> None:
+    mod = _load_script()
+    cls = _reset_fake(str(tmp_path))
+    with (
+        patch.object(mod, "ThreadPoolExecutor", side_effect=RuntimeError("no")),
+        pytest.raises(RuntimeError),
+    ):
+        mod.parallel_install(cls, ["esphome/a @ 1.0"])
+    assert cls.lock_events == ["lock", "unlock"]
+
+
+def test_parse_specs_unreadable_ini_fails_loudly(tmp_path: Path) -> None:
+    """A bad path must not silently build an image with no dependencies."""
+    mod = _load_script()
+    args = Namespace(libraries=True, platforms=False, tools=False)
+    with pytest.raises(SystemExit):
+        mod.parse_specs(str(tmp_path / "missing.ini"), args)
+
+
+def test_platformio_surface_for_install_deps_script() -> None:
+    """A PlatformIO bump that changes these members must fail here, not
+    silently turn the docker image's parallel preinstall into a no-op."""
+    # The script calls these positionally; pin the positions, not just
+    # membership, so a parameter reorder trips the wire too
+    params = inspect.signature(PackageManagerInstallMixin._install).parameters
+    assert list(params)[1] == "spec"
+    assert "skip_dependencies" in params
+    assert "compatibility" in params
+    for cls in (ToolPackageManager, LibraryPackageManager):
+        assert list(inspect.signature(cls.__init__).parameters)[1] == "package_dir"
+    for name in (
+        "lock",
+        "unlock",
+        "get_package",
+        "memcache_reset",
+        "get_pkg_dependencies",
+        "dependency_to_spec",
+        "get_download_dir",
+        "get_tmp_dir",
+    ):
+        assert callable(getattr(BasePackageManager, name))
+    # Losing any of these turns the wave into main()'s silent serial
+    # fallback: ensure_spec runs in the coordinator, the spec attributes
+    # feed the dedupe, cleanup, and dependency filters
+    assert callable(BasePackageManager.ensure_spec)
+    spec = PackageSpec("owner/name @ ^1.0")
+    assert spec.name == "name"
+    assert spec.owner == "owner"
+    assert spec.uri is None
+    assert spec.external is False
+    assert Version("1.5.0") in spec.requirements
+    # The failure-cleanup path degrades to a single line if these vanish
+    assert callable(fs.rmtree)
+    assert callable(fs.load_json)
+    # piopm_matches only tolerates a corrupt .piopm through this base;
+    # losing it would flip a wave failure from degrade to build failure
+    assert issubclass(InvalidJSONFile, ValueError)
+    assert PackageItem("pkg-dir").path == "pkg-dir"
+    assert callable(PackageCompatibility.from_dependency)

tests/unit_tests/core/test_config.py

@@ -3,7 +3,6 @@
 from collections.abc import Callable
 import os
 from pathlib import Path
-import types
 from typing import Any
 from unittest.mock import MagicMock, Mock, patch
 
@@ -705,33 +704,6 @@ def test_include_file_with_c_header(
         assert '#include "c_library.h"' in mock_raw_statement.text
 
 
-def test_get_usable_cpu_count() -> None:
-    """Test get_usable_cpu_count returns CPU count."""
-    count = config.get_usable_cpu_count()
-    assert isinstance(count, int)
-    assert count > 0
-
-
-def test_get_usable_cpu_count_with_process_cpu_count() -> None:
-    """Test get_usable_cpu_count uses process_cpu_count when available."""
-    # Test with process_cpu_count (Python 3.13+)
-    # Create a mock os module with process_cpu_count
-
-    mock_os = types.SimpleNamespace(process_cpu_count=lambda: 8, cpu_count=lambda: 4)
-
-    with patch("esphome.core.config.os", mock_os):
-        # When process_cpu_count exists, it should be used
-        count = config.get_usable_cpu_count()
-        assert count == 8
-
-    # Test fallback to cpu_count when process_cpu_count not available
-    mock_os_no_process = types.SimpleNamespace(cpu_count=lambda: 4)
-
-    with patch("esphome.core.config.os", mock_os_no_process):
-        count = config.get_usable_cpu_count()
-        assert count == 4
-
-
 def test_list_target_platforms(tmp_path: Path) -> None:
     """Test _list_target_platforms returns available platforms."""
     # Create mock components directory structure

tests/unit_tests/test_config_validation.py

@@ -2565,6 +2565,52 @@ def test_returning_lambda_no_return() -> None:
         cv.returning_lambda(Lambda("int x = 5;"))
 
 
+def test_returning_lambda_return_only_in_comment() -> None:
+    with pytest.raises(Invalid, match="return statement"):
+        cv.returning_lambda(Lambda("// return 5;\nint x = 5;"))
+
+
+def test_returning_lambda_missing_semicolon_is_accepted() -> None:
+    """A forgotten semicolon is left for the C++ compiler to report."""
+    assert isinstance(cv.returning_lambda(Lambda("return x")), Lambda)
+
+
+@pytest.mark.parametrize(
+    ("value", "expected"),
+    [
+        ("return 5;", True),
+        ("if (x) { return x; } return 0;", True),
+        ("if (x) return 1; else return 0;", True),
+        ("switch (x) { case 0: return 1; }", True),
+        # a semicolon means code: any return keyword counts
+        ("return not x;", True),
+        ("return a and b;", True),
+        ("please return the sensor; then wait", True),
+        # a forgotten semicolon is still lambda source; the compiler reports it
+        ("return id(x).state", True),
+        ("return x", True),
+        ("return 5", True),
+        ("return not x", True),
+        # accepted: a one-word tail is indistinguishable from 'return x'
+        ("return soon", True),
+        ("Alert: return home", True),
+        ("static value", False),
+        ("no returns here", False),
+        ("the_return_value", False),
+        # without a semicolon, prose is not lambda source
+        ("please return the item", False),
+        ("return to sender", False),
+        ("return a and b", False),
+        # return only inside a comment is not a return statement
+        ("// return 5;\nint x = 5;", False),
+        ("/* return 5; */ int x = 5;", False),
+        ("return 5; // done", True),
+    ],
+)
+def test_looks_like_returning_lambda(value: str, expected: bool) -> None:
+    assert cv.looks_like_returning_lambda(value) is expected
+
+
 # ---------------------------------------------------------------------------
 # dimensions
 # ---------------------------------------------------------------------------

tests/unit_tests/test_espidf_framework.py

@@ -912,7 +912,7 @@ def test_prefetch_leaves_unverifiable_entries_to_the_installer(
             "esphome.espidf.framework.run_command",
             return_value=(True, json.dumps(entries), ""),
         ),
-        patch("esphome.espidf.framework.download_with_resume") as download,
+        patch("esphome.framework_helpers.download_with_resume") as download,
         patch("esphome.espidf.framework.get_system_python_path", return_value="python"),
         patch("esphome.framework_helpers._BatchDownloadProgress") as progress_cls,
     ):
@@ -935,7 +935,7 @@ def test_prefetch_all_entries_unverifiable_is_a_noop(tmp_path: Path) -> None:
             "esphome.espidf.framework.run_command",
             return_value=(True, json.dumps(entries), ""),
         ),
-        patch("esphome.espidf.framework.download_with_resume") as download,
+        patch("esphome.framework_helpers.download_with_resume") as download,
         patch("esphome.espidf.framework.get_system_python_path", return_value="python"),
     ):
         _prefetch_idf_tool_archives(tmp_path, "esp32", ["required"], None)
@@ -953,7 +953,7 @@ def test_prefetch_dedupes_entries_by_dest(tmp_path: Path) -> None:
             "esphome.espidf.framework.run_command",
             return_value=(True, json.dumps(entries), ""),
         ),
-        patch("esphome.espidf.framework.download_with_resume") as download,
+        patch("esphome.framework_helpers.download_with_resume") as download,
         patch("esphome.espidf.framework.get_system_python_path", return_value="python"),
         patch("esphome.framework_helpers._BatchDownloadProgress"),
     ):
@@ -968,7 +968,7 @@ def test_prefetch_downloads_each_archive_with_resume(tmp_path: Path) -> None:
             "esphome.espidf.framework.run_command",
             return_value=(True, _PREFETCH_JSON, ""),
         ),
-        patch("esphome.espidf.framework.download_with_resume") as download,
+        patch("esphome.framework_helpers.download_with_resume") as download,
         patch("esphome.espidf.framework.get_system_python_path", return_value="python"),
         patch("esphome.framework_helpers._BatchDownloadProgress") as progress_cls,
     ):
@@ -1012,7 +1012,7 @@ def test_prefetch_downloads_archives_concurrently(tmp_path: Path) -> None:
             "esphome.espidf.framework.run_command",
             return_value=(True, json.dumps(entries), ""),
         ),
-        patch("esphome.espidf.framework.download_with_resume") as download,
+        patch("esphome.framework_helpers.download_with_resume") as download,
         patch("esphome.espidf.framework.get_system_python_path", return_value="python"),
         patch(
             "esphome.framework_helpers.ThreadPoolExecutor", wraps=ThreadPoolExecutor
@@ -1033,7 +1033,7 @@ def test_prefetch_skips_already_downloaded_archives(tmp_path: Path) -> None:
             "esphome.espidf.framework.run_command",
             return_value=(True, _PREFETCH_JSON, ""),
         ),
-        patch("esphome.espidf.framework.download_with_resume") as download,
+        patch("esphome.framework_helpers.download_with_resume") as download,
         patch("esphome.espidf.framework.get_system_python_path", return_value="python"),
     ):
         _prefetch_idf_tool_archives(tmp_path, "esp32", ["required"], None)
@@ -1066,7 +1066,7 @@ def test_prefetch_failures_never_raise(
     with (
         patch("esphome.espidf.framework.run_command", return_value=run_result),
         patch(
-            "esphome.espidf.framework.download_with_resume",
+            "esphome.framework_helpers.download_with_resume",
             side_effect=download_error,
         ),
         patch("esphome.espidf.framework.get_system_python_path", return_value="python"),
@@ -1088,7 +1088,7 @@ def test_prefetch_total_failure_logs_error(
             return_value=(True, _PREFETCH_JSON, ""),
         ),
         patch(
-            "esphome.espidf.framework.download_with_resume",
+            "esphome.framework_helpers.download_with_resume",
             side_effect=OSError("proxy refuses everything"),
         ),
         patch("esphome.espidf.framework.get_system_python_path", return_value="python"),
@@ -1113,7 +1113,7 @@ def test_prefetch_one_failed_archive_does_not_stop_the_rest(
             return_value=(True, _PREFETCH_JSON, ""),
         ),
         patch(
-            "esphome.espidf.framework.download_with_resume",
+            "esphome.framework_helpers.download_with_resume",
             side_effect=_fail_cmake_download,
         ) as download,
         patch("esphome.espidf.framework.get_system_python_path", return_value="python"),
@@ -1134,7 +1134,7 @@ def test_prefetch_finishes_progress_bar_and_cancels_queue(tmp_path: Path) -> Non
             "esphome.espidf.framework.run_command",
             return_value=(True, _PREFETCH_JSON, ""),
         ),
-        patch("esphome.espidf.framework.download_with_resume"),
+        patch("esphome.framework_helpers.download_with_resume"),
         patch("esphome.espidf.framework.get_system_python_path", return_value="python"),
         patch("esphome.framework_helpers._BatchDownloadProgress") as progress_cls,
         patch("esphome.framework_helpers.ThreadPoolExecutor") as pool_cls,

tests/unit_tests/test_framework_helpers.py

@@ -2280,6 +2280,32 @@ class TestGetProjectCxxCompileFlags:
             assert get_project_cxx_compile_flags() == []
 
 
+def test_resume_fetch_job_threads_tracker(tmp_path: Path) -> None:
+    """The batch runner passes the tracker positionally; the shared adapter
+    must deliver it as download_with_resume's progress keyword."""
+    from esphome.framework_helpers import resume_fetch_job
+
+    with patch("esphome.framework_helpers.download_with_resume") as mock_download:
+        fetch = resume_fetch_job("https://x/a.zip", tmp_path / "a", sha256="ff", size=9)
+        tracker = lambda done: None  # noqa: E731
+        fetch(tracker)
+    mock_download.assert_called_once_with(
+        "https://x/a.zip", tmp_path / "a", progress=tracker, sha256="ff", size=9
+    )
+
+
+def test_warn_prefetch_failures_names_each_failure(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """The shared failure loop warns per job with the failure reason."""
+    from esphome.framework_helpers import warn_prefetch_failures
+
+    warn_prefetch_failures([("toolchain-x@1", OSError("down"))])
+    assert "Could not prefetch toolchain-x@1: down" in caplog.text
+    warn_prefetch_failures([("lib", OSError("gone"))], "Prefetch of %s failed: %s")
+    assert "Prefetch of lib failed: gone" in caplog.text
+
+
 @pytest.mark.parametrize(
     ("platform", "input_path", "expected"),
     [
@@ -2312,3 +2338,18 @@ def test_strip_win_long_path_prefix(
     r"""``\\?\`` and ``\\?\UNC\`` prefixes are stripped only on win32."""
     with patch("esphome.framework_helpers.sys.platform", platform):
         assert framework_helpers.strip_win_long_path_prefix(input_path) == expected
+
+
+def test_discard_partial_download_logs_undeletable(
+    tmp_path: Path, caplog: pytest.LogCaptureFixture
+) -> None:
+    """An unremovable staging file leaves a debug trace; the caller's
+    cache is never pruned, so silence would hide unbounded growth."""
+    dest = tmp_path / "archive"
+    dest.write_bytes(b"stale")
+    with (
+        patch.object(Path, "unlink", side_effect=OSError("busy")),
+        caplog.at_level(logging.DEBUG),
+    ):
+        framework_helpers.discard_partial_download(dest)
+    assert "Could not remove" in caplog.text

tests/unit_tests/test_helpers.py

@@ -4,6 +4,7 @@ import os
 from pathlib import Path
 import socket
 import stat
+import types
 from unittest.mock import MagicMock, patch
 
 from aioesphomeapi.host_resolver import AddrInfo, IPv4Sockaddr, IPv6Sockaddr
@@ -1154,3 +1155,26 @@ def test_progressbar_interrupt_keeps_finished_bar_done(monkeypatch) -> None:
 def test_format_duration(seconds: float, expected: str) -> None:
     """Test that durations are rendered as short human-readable strings."""
     assert helpers.format_duration(seconds) == expected
+
+
+def test_get_usable_cpu_count() -> None:
+    """Returns a positive int on the real host."""
+    count = helpers.get_usable_cpu_count()
+    assert isinstance(count, int)
+    assert count > 0
+
+
+def test_get_usable_cpu_count_sources() -> None:
+    """Prefers process_cpu_count, falls back to cpu_count, degrades to 1."""
+    mock_os = types.SimpleNamespace(process_cpu_count=lambda: 8, cpu_count=lambda: 4)
+    with patch("esphome.helpers.os", mock_os):
+        assert helpers.get_usable_cpu_count() == 8
+
+    mock_os_no_process = types.SimpleNamespace(cpu_count=lambda: 4)
+    with patch("esphome.helpers.os", mock_os_no_process):
+        assert helpers.get_usable_cpu_count() == 4
+
+    # An undeterminable count degrades to one worker, never zero
+    mock_os_unknown = types.SimpleNamespace(cpu_count=lambda: None)
+    with patch("esphome.helpers.os", mock_os_unknown):
+        assert helpers.get_usable_cpu_count() == 1

tests/unit_tests/test_platformio_toolchain.py

@@ -932,8 +932,13 @@ def test_run_compile(setup_core: Path, mock_run_platformio_cli_run: Mock) -> Non
     config = {CONF_ESPHOME: {CONF_COMPILE_PROCESS_LIMIT: 4}}
     mock_run_platformio_cli_run.return_value = 0
 
-    toolchain.run_compile(config, verbose=True)
+    with patch(
+        "esphome.platformio.prefetch.prefetch_platformio_packages"
+    ) as mock_prefetch:
+        toolchain.run_compile(config, verbose=True)
 
+    # The only wiring of the prefetch into a build lives here
+    mock_prefetch.assert_called_once_with()
     mock_run_platformio_cli_run.assert_called_once_with(config, True, "-j4")
 
 
@@ -947,7 +952,8 @@ def test_run_compile_without_process_limit(
     config = {CONF_ESPHOME: {}}
     mock_run_platformio_cli_run.return_value = 0
 
-    toolchain.run_compile(config, verbose=False)
+    with patch("esphome.platformio.prefetch.prefetch_platformio_packages"):
+        toolchain.run_compile(config, verbose=False)
 
     mock_run_platformio_cli_run.assert_called_once_with(config, False)
 
@@ -1677,8 +1683,8 @@ def pio_core_dir(tmp_path: Path) -> Path:
 
 
 def test_current_python_minor_matches_running_interpreter() -> None:
-    """_current_python_minor returns major.minor of the running interpreter."""
-    assert toolchain._current_python_minor() == _CURRENT_MINOR
+    """current_python_minor returns major.minor of the running interpreter."""
+    assert toolchain.current_python_minor() == _CURRENT_MINOR
 
 
 def test_pio_stamp_round_trip(tmp_path: Path) -> None: