Merge remote-tracking branch 'upstream/dev' into integration

tests/component_tests/esp32/test_esp32.py

@@ -16,8 +16,8 @@ from esphome.const import (
     CONF_ESPHOME,
     CONF_IGNORE_PIN_VALIDATION_ERROR,
     CONF_NUMBER,
-    KEY_NATIVE_IDF,
     PlatformFramework,
+    Toolchain,
 )
 from esphome.core import CORE
 from tests.component_tests.types import SetCoreConfigCallable
@@ -266,7 +266,7 @@ def test_native_idf_enables_reproducible_build(
 
     CORE.config_path = component_config_path("reproducible_build.yaml")
     CORE.config = read_config({})
-    CORE.data[KEY_NATIVE_IDF] = True
+    CORE.toolchain = Toolchain.ESP_IDF
     generate_cpp_contents(CORE.config)
 
     sdkconfig = CORE.data[KEY_ESP32][KEY_SDKCONFIG_OPTIONS]

tests/component_tests/lvgl/config/widget_state_test.yaml

@@ -0,0 +1,83 @@
+esphome:
+  name: test
+
+esp32:
+  board: esp32dev
+  framework:
+    type: esp-idf
+
+spi:
+  - id: spi_bus
+    clk_pin: GPIO18
+    mosi_pin: GPIO23
+
+display:
+  - platform: mipi_spi
+    spi_id: spi_bus
+    model: st7789v
+    id: tft_display
+    dimensions:
+      width: 240
+      height: 320
+    cs_pin: GPIO22
+    dc_pin: GPIO21
+    auto_clear_enabled: false
+    invert_colors: false
+    update_interval: never
+
+lvgl:
+  id: lvgl_id
+  displays: tft_display
+  pages:
+    - id: main_page
+      widgets:
+        # Widget with multiple static states; one true, one false.
+        - button:
+            id: btn_static
+            state:
+              checked: true
+              disabled: false
+
+        # Widget with a templated (lambda) state.
+        - button:
+            id: btn_lambda
+            state:
+              pressed: !lambda return true;
+
+        # Button referenced by enable/disable actions; the on_click handler
+        # exercises both branches of the obj_disable/obj_enable code path.
+        - button:
+            id: btn_actions
+            on_click:
+              - lvgl.widget.disable: btn_actions
+              - lvgl.widget.enable: btn_actions
+
+        # Button matrix with two buttons; matrix_btn_a is targeted by
+        # lvgl.widget.disable/enable actions to exercise the
+        # MatrixButton.set_state code path.
+        - buttonmatrix:
+            id: matrix
+            rows:
+              - buttons:
+                  - id: matrix_btn_a
+                    text: A
+                    control:
+                      checkable: true
+                  - id: matrix_btn_b
+                    text: B
+                    control:
+                      checkable: true
+            on_click:
+              - lvgl.widget.disable: matrix_btn_a
+              - lvgl.widget.enable: matrix_btn_a
+
+        # Switch derived from an LVGL switch widget – exercises
+        # set_state(LV_STATE.CHECKED, v) inside the control lambda.
+        - switch:
+            id: switch_widget
+
+switch:
+  - platform: lvgl
+    id: lvgl_switch
+    name: LVGL Switch
+    widget: switch_widget

tests/component_tests/lvgl/test_widget_state.py

@@ -0,0 +1,167 @@
+"""Tests for LVGL widget state code generation.
+
+These tests cover the change from the old ``add_state``/``clear_state`` helpers
+on :class:`Widget` (and on :class:`MatrixButton`) to a single ``set_state``
+method that delegates to the new C++ helpers
+``LvglComponent::lv_obj_set_state_value`` and
+``LvglComponent::lv_buttonmatrix_set_button_ctrl_value``.
+"""
+
+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:
+    """Generate the C++ output for the shared widget-state YAML config once
+    per module.
+
+    Module-scoped so the (relatively expensive) codegen runs a single time;
+    the function-scoped fixtures from ``conftest.py`` (e.g. ``generate_main``)
+    can't be requested from a higher-scoped fixture, so the small amount of
+    setup is inlined here.  The captured string is independent of
+    ``CORE.reset()`` calls that the per-test autouse fixtures perform after
+    this fixture has produced its value.
+    """
+    config_path = Path(request.fspath).parent / "config" / "widget_state_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_global_section + CORE.cpp_main_section
+    finally:
+        CORE.config_path = original_path
+        CORE.reset()
+
+
+def test_static_state_emits_set_state_value(main_cpp: str) -> None:
+    """A widget with ``state: { checked: true, disabled: false }`` should
+    generate one ``lv_obj_set_state_value`` call per entry, with the
+    appropriate boolean argument.
+    """
+    assert (
+        "LvglComponent::lv_obj_set_state_value(btn_static, LV_STATE_CHECKED, true)"
+        in main_cpp
+    )
+    assert (
+        "LvglComponent::lv_obj_set_state_value(btn_static, LV_STATE_DISABLED, false)"
+        in main_cpp
+    )
+
+
+def test_lambda_state_emits_set_state_value_with_lambda(main_cpp: str) -> None:
+    """A widget with ``state: { pressed: !lambda return true; }`` should
+    generate ``lv_obj_set_state_value(..., LV_STATE_PRESSED, <expr>)`` where
+    ``<expr>`` is the lambda's return value (cast or inlined), not a static
+    bool.
+    """
+    # The set_state call is emitted for the templated state.
+    assert (
+        "LvglComponent::lv_obj_set_state_value(btn_lambda, LV_STATE_PRESSED,"
+        in main_cpp
+    )
+    # And it must NOT have collapsed the lambda to a literal true/false.
+    assert (
+        "LvglComponent::lv_obj_set_state_value(btn_lambda, LV_STATE_PRESSED, true)"
+        not in main_cpp
+    )
+    # The legacy if/else over add_state/remove_state is gone.
+    assert "lv_obj_add_state(btn_lambda, LV_STATE_PRESSED)" not in main_cpp
+    assert "lv_obj_remove_state(btn_lambda, LV_STATE_PRESSED)" not in main_cpp
+
+
+def test_widget_disable_action_uses_set_state_value(main_cpp: str) -> None:
+    """``lvgl.widget.disable: btn_actions`` should emit a
+    ``set_state_value(..., LV_STATE_DISABLED, true)`` call rather than the
+    legacy ``lv_obj_add_state``.
+    """
+    assert (
+        "LvglComponent::lv_obj_set_state_value(btn_actions, LV_STATE_DISABLED, true)"
+        in main_cpp
+    )
+    # No leftover legacy add_state for the DISABLED state of this widget.
+    assert "lv_obj_add_state(btn_actions, LV_STATE_DISABLED)" not in main_cpp
+
+
+def test_widget_enable_action_uses_set_state_value(main_cpp: str) -> None:
+    """``lvgl.widget.enable: btn_actions`` should emit a
+    ``set_state_value(..., LV_STATE_DISABLED, false)`` call rather than the
+    legacy ``lv_obj_remove_state``.
+    """
+    assert (
+        "LvglComponent::lv_obj_set_state_value(btn_actions, LV_STATE_DISABLED, false)"
+        in main_cpp
+    )
+    assert "lv_obj_remove_state(btn_actions, LV_STATE_DISABLED)" not in main_cpp
+
+
+def test_buttonmatrix_disable_action_uses_helper(main_cpp: str) -> None:
+    """``lvgl.widget.disable: matrix_btn_a`` should route through the new
+    ``lv_buttonmatrix_set_button_ctrl_value`` helper for button index 0
+    with the ``DISABLED`` control bit set to ``true``, instead of the
+    legacy ``lv_buttonmatrix_set_button_ctrl``.
+
+    The button matrix obj is the compound's ``obj`` member and the index
+    is the position of the button in the row layout.
+    """
+    assert (
+        "LvglComponent::lv_buttonmatrix_set_button_ctrl_value(matrix->obj, 0, "
+        "LV_BUTTONMATRIX_CTRL_DISABLED, true)"
+    ) in main_cpp
+
+
+def test_buttonmatrix_enable_action_uses_helper(main_cpp: str) -> None:
+    """``lvgl.widget.enable: matrix_btn_a`` should route through the new
+    ``lv_buttonmatrix_set_button_ctrl_value`` helper for button index 0
+    with the ``DISABLED`` control bit set to ``false``, instead of the
+    legacy ``lv_buttonmatrix_clear_button_ctrl``.
+    """
+    assert (
+        "LvglComponent::lv_buttonmatrix_set_button_ctrl_value(matrix->obj, 0, "
+        "LV_BUTTONMATRIX_CTRL_DISABLED, false)"
+    ) in main_cpp
+    # The legacy clear_button_ctrl path is gone for the matrix button enable
+    # action.
+    assert (
+        "lv_buttonmatrix_clear_button_ctrl(matrix->obj, 0, LV_BUTTONMATRIX_CTRL_DISABLED)"
+        not in main_cpp
+    )
+
+
+def test_lvgl_switch_control_calls_set_state_value(main_cpp: str) -> None:
+    """The LVGL switch platform installs a control lambda that mirrors the
+    switch's bool value into ``LV_STATE_CHECKED`` via
+    ``lv_obj_set_state_value`` (replacing the previous if/else over
+    ``add_state``/``clear_state`` plus an explicit ``send_event`` of
+    ``lv_api_event``).
+    """
+    # The control lambda calls the new helper with the bool ``v`` parameter.
+    assert (
+        "LvglComponent::lv_obj_set_state_value(switch_widget, LV_STATE_CHECKED, v)"
+        in main_cpp
+    )
+    # The deprecated lv_api_event symbol must no longer appear anywhere.
+    assert "lv_api_event" not in main_cpp
+
+
+def test_default_state_does_not_emit_set_state_value(main_cpp: str) -> None:
+    """A widget without a ``state:`` block must not generate any
+    ``lv_obj_set_state_value`` calls for it. (Sanity-check that the
+    new code path is opt-in driven by the YAML.)
+    """
+    assert (
+        "LvglComponent::lv_obj_set_state_value(switch_widget, LV_STATE_DISABLED"
+        not in main_cpp
+    )
+    assert (
+        "LvglComponent::lv_obj_set_state_value(btn_static, LV_STATE_PRESSED"
+        not in main_cpp
+    )

tests/components/lvgl/common.yaml

@@ -38,7 +38,7 @@ number:
   - platform: lvgl
     widget: slider_id
     name: LVGL Slider Number
-    update_on_release: true
+    trigger: on_release
     restore_value: true
   - platform: lvgl
     widget: lv_arc

tests/components/lvgl/lvgl-package.yaml

@@ -309,6 +309,11 @@ lvgl:
                 - logger.log:
                     format: "Roller changed = %d: %s"
                     args: [x, text.c_str()]
+            on_update:
+              then:
+                - logger.log:
+                    format: "Roller updated = %d: %s"
+                    args: [x, text.c_str()]
         - animimg:
             height: 60
             id: anim_img
@@ -630,6 +635,10 @@ lvgl:
               logger.log:
                 format: "state now %d"
                 args: [x]
+            on_update:
+              logger.log:
+                format: "button updated %d"
+                args: [x]
             on_short_click:
               lvgl.widget.hide: hello_label
             on_long_press:
@@ -758,6 +767,9 @@ lvgl:
                       lambda: return tile == id(tile_1);
                     then:
                       - logger.log: "tile 1 is now showing"
+            on_update:
+              then:
+                - logger.log: "tileview updated programmatically"
             tiles:
               - id: tile_1
                 scroll_snap_y: center
@@ -983,6 +995,11 @@ lvgl:
                 - logger.log:
                     format: "Arc value is %f"
                     args: [x]
+            on_update:
+              then:
+                - logger.log:
+                    format: "Arc updated to %f"
+                    args: [x]
             scroll_on_focus: true
             value: 75
             min_value: 1
@@ -1085,6 +1102,11 @@ lvgl:
                 - logger.log:
                     format: "slider value %f"
                     args: [x]
+            on_update:
+              then:
+                - logger.log:
+                    format: "slider updated to %f"
+                    args: [x]
             on_click:
               then:
                 - lvgl.slider.update:
@@ -1197,6 +1219,10 @@ lvgl:
                         logger.log:
                           format: "Dropdown changed = %d: %s"
                           args: [x, text.c_str()]
+                      on_update:
+                        logger.log:
+                          format: "Dropdown updated = %d: %s"
+                          args: [x, text.c_str()]
                       on_cancel:
                         logger.log:
                           format: "Dropdown closed = %d"
@@ -1437,6 +1463,30 @@ color:
     blue_int: 64
     white_int: 255
 
+sensor:
+  - platform: lvgl
+    widget: lv_arc_1
+    id: lvgl_arc1_sensor_on_change
+    name: LVGL Arc1 Sensor on_change
+    trigger: on_change
+  - platform: lvgl
+    widget: bar_id
+    id: lvgl_bar_sensor_on_release
+    name: LVGL Bar Sensor on_release
+    trigger: on_release
+
+number:
+  - platform: lvgl
+    widget: lv_arc_1
+    id: lvgl_arc1_number_on_update
+    name: LVGL Arc1 Number on_update
+    trigger: on_update
+  - platform: lvgl
+    widget: spinbox_id
+    id: lvgl_spinbox_number_on_change
+    name: LVGL Spinbox Number on_change
+    trigger: on_change
+
 select:
   - platform: lvgl
     id: lv_roller_select

tests/components/ms8607/common.yaml

@@ -0,0 +1,11 @@
+sensor:
+  - platform: ms8607
+    i2c_id: i2c_bus
+    temperature:
+      name: Temperature
+    humidity:
+      name: Humidity
+    pressure:
+      name: Pressure
+    address: 0x76
+    update_interval: 15s

tests/components/ms8607/test.esp32-idf.yaml

@@ -0,0 +1,8 @@
+substitutions:
+  i2c_scl: GPIO16
+  i2c_sda: GPIO17
+
+packages:
+  i2c: !include ../../test_build_components/common/i2c/esp32-idf.yaml
+
+<<: !include common.yaml

tests/components/ms8607/test.esp8266-ard.yaml

@@ -0,0 +1,8 @@
+substitutions:
+  i2c_scl: GPIO5
+  i2c_sda: GPIO4
+
+packages:
+  i2c: !include ../../test_build_components/common/i2c/esp8266-ard.yaml
+
+<<: !include common.yaml

tests/components/ms8607/test.rp2040-ard.yaml

@@ -0,0 +1,8 @@
+substitutions:
+  i2c_scl: GPIO5
+  i2c_sda: GPIO4
+
+packages:
+  i2c: !include ../../test_build_components/common/i2c/rp2040-ard.yaml
+
+<<: !include common.yaml

tests/components/nrf52/test.nrf52-adafruit.yaml

@@ -20,4 +20,4 @@ nrf52:
     voltage: 2.1V
     uicr_erase: true
   framework:
-    version: "2.6.1-a"
+    version: "2.6.1-b"

tests/components/script/common.yaml

@@ -7,6 +7,12 @@ esphome:
           prefix: "Test"
           param2: 0
           param3: true
+      - script.execute:
+          id: my_script_with_array_params
+          ints: [42, 100]
+          floats: [1.5, 2.5]
+          bools: [true, false]
+          strings: ["a", "b"]
       - script.wait: my_script
       - script.stop: my_script
       - if:
@@ -34,6 +40,16 @@ script:
     mode: restart
     then:
       - lambda: 'ESP_LOGD("main", "Hello World!");'
+  - id: my_script_with_array_params
+    parameters:
+      ints: int[]
+      floats: float[]
+      bools: bool[]
+      strings: string[]
+    then:
+      - lambda: |-
+          ESP_LOGD("main", "ints=%d floats=%f bools=%d strings=%s",
+                   ints[0], floats[0], bools[0], strings[0].c_str());
   - id: my_script_with_params
     parameters:
       prefix: string

tests/components/speaker/spdif_mode.esp32-idf.yaml

@@ -0,0 +1,25 @@
+substitutions:
+  i2s_bclk_pin: GPIO27
+  i2s_lrclk_pin: GPIO26
+  i2s_mclk_pin: GPIO25
+  i2s_dout_pin: GPIO12
+  spdif_data_pin: GPIO4
+
+packages:
+  i2c: !include ../../test_build_components/common/i2c/esp32-idf.yaml
+
+i2s_audio:
+  - id: i2s_output
+
+speaker:
+  - platform: i2s_audio
+    id: speaker_id
+    dac_type: external
+    i2s_dout_pin: ${spdif_data_pin}
+    spdif_mode: true
+    use_apll: true
+    timeout: 2s
+    sample_rate: 48000
+    bits_per_sample: 16bit
+    channel: stereo
+    i2s_mode: primary

tests/components/zigbee/common_esp32.yaml

@@ -1,10 +1,10 @@
+packages:
+  - !include common.yaml
+
 binary_sensor:
   - platform: template
     name: "Garage Door Open 10"
     report: "enable"
-  - platform: template
-    name: "Garage Door Open 11"
-    report: "coordinator"
   - platform: template
     name: "Garage Door Open 12"
     report: "force"
@@ -12,3 +12,7 @@ binary_sensor:
 zigbee:
   model: zigbee_test
   router: true
+  power_source: MAINS_SINGLE_PHASE
+  on_join:
+    then:
+      - logger.log: "Joined network"

tests/dashboard/test_web_server.py

@@ -128,8 +128,8 @@ def mock_storage_json() -> Generator[MagicMock]:
 
 @pytest.fixture
 def mock_idedata() -> Generator[MagicMock]:
-    """Fixture to mock platformio_api.IDEData."""
-    with patch("esphome.dashboard.web_server.platformio_api.IDEData") as mock:
+    """Fixture to mock platformio toolchain.IDEData."""
+    with patch("esphome.dashboard.web_server.toolchain.IDEData") as mock:
         yield mock
 
 

tests/integration/conftest.py

@@ -23,7 +23,7 @@ import pytest_asyncio
 
 import esphome.config
 from esphome.core import CORE
-from esphome.platformio_api import get_idedata
+from esphome.platformio.toolchain import get_idedata
 
 from .const import (
     API_CONNECTION_TIMEOUT,

tests/integration/fixtures/external_components/crc8_test_component/crc8_test_component.cpp

@@ -2,8 +2,7 @@
 #include "esphome/core/helpers.h"
 #include "esphome/core/log.h"
 
-namespace esphome {
-namespace crc8_test_component {
+namespace esphome::crc8_test_component {
 
 static const char *const TAG = "crc8_test";
 
@@ -166,5 +165,4 @@ void CRC8TestComponent::log_test_result(const char *test_name, bool passed) {
   }
 }
 
-}  // namespace crc8_test_component
-}  // namespace esphome
+}  // namespace esphome::crc8_test_component

tests/integration/fixtures/external_components/crc8_test_component/crc8_test_component.h

@@ -4,8 +4,7 @@
 #include "esphome/core/helpers.h"
 #include "esphome/core/log.h"
 
-namespace esphome {
-namespace crc8_test_component {
+namespace esphome::crc8_test_component {
 
 class CRC8TestComponent : public Component {
  public:
@@ -25,5 +24,4 @@ class CRC8TestComponent : public Component {
                    uint8_t poly = 0x8C, bool msb_first = false);
 };
 
-}  // namespace crc8_test_component
-}  // namespace esphome
+}  // namespace esphome::crc8_test_component

tests/integration/fixtures/external_components/custom_api_device_component/custom_api_device_component.cpp

@@ -2,8 +2,7 @@
 #include "esphome/core/log.h"
 
 #ifdef USE_API
-namespace esphome {
-namespace custom_api_device_component {
+namespace esphome::custom_api_device_component {
 
 static const char *const TAG = "custom_api";
 
@@ -58,6 +57,5 @@ void CustomAPIDeviceComponent::on_ha_state_changed(std::string entity_id, std::s
   ESP_LOGI(TAG, "This subscription uses std::string API for backward compatibility");
 }
 
-}  // namespace custom_api_device_component
-}  // namespace esphome
+}  // namespace esphome::custom_api_device_component
 #endif  // USE_API

tests/integration/fixtures/external_components/custom_api_device_component/custom_api_device_component.h

@@ -6,8 +6,7 @@
 #include "esphome/components/api/custom_api_device.h"
 
 #ifdef USE_API
-namespace esphome {
-namespace custom_api_device_component {
+namespace esphome::custom_api_device_component {
 
 using namespace api;
 
@@ -28,6 +27,5 @@ class CustomAPIDeviceComponent : public Component, public CustomAPIDevice {
   void on_ha_state_changed(std::string entity_id, std::string state);
 };
 
-}  // namespace custom_api_device_component
-}  // namespace esphome
+}  // namespace esphome::custom_api_device_component
 #endif  // USE_API

tests/integration/fixtures/external_components/defer_stress_component/defer_stress_component.cpp

@@ -5,8 +5,7 @@
 #include <vector>
 #include <chrono>
 
-namespace esphome {
-namespace defer_stress_component {
+namespace esphome::defer_stress_component {
 
 static const char *const TAG = "defer_stress";
 
@@ -71,5 +70,4 @@ void DeferStressComponent::run_multi_thread_test() {
   ESP_LOGI(TAG, "All threads finished in %lldms. Created %d defer requests", thread_time, this->total_defers_.load());
 }
 
-}  // namespace defer_stress_component
-}  // namespace esphome
+}  // namespace esphome::defer_stress_component

tests/integration/fixtures/external_components/defer_stress_component/defer_stress_component.h

@@ -3,8 +3,7 @@
 #include "esphome/core/component.h"
 #include <atomic>
 
-namespace esphome {
-namespace defer_stress_component {
+namespace esphome::defer_stress_component {
 
 class DeferStressComponent : public Component {
  public:
@@ -16,5 +15,4 @@ class DeferStressComponent : public Component {
   std::atomic<int> executed_defers_{0};
 };
 
-}  // namespace defer_stress_component
-}  // namespace esphome
+}  // namespace esphome::defer_stress_component

tests/integration/fixtures/external_components/loop_test_component/loop_test_component.cpp

@@ -1,7 +1,6 @@
 #include "loop_test_component.h"
 
-namespace esphome {
-namespace loop_test_component {
+namespace esphome::loop_test_component {
 
 void LoopTestComponent::setup() { ESP_LOGI(TAG, "[%s] Setup called", this->name_.c_str()); }
 
@@ -63,5 +62,4 @@ void LoopTestUpdateComponent::update() {
            this->update_count_, loop_disabled ? "YES" : "NO");
 }
 
-}  // namespace loop_test_component
-}  // namespace esphome
+}  // namespace esphome::loop_test_component

tests/integration/fixtures/external_components/loop_test_component/loop_test_component.h

@@ -6,8 +6,7 @@
 #include "esphome/core/automation.h"
 #include "esphome/core/helpers.h"
 
-namespace esphome {
-namespace loop_test_component {
+namespace esphome::loop_test_component {
 
 static const char *const TAG = "loop_test_component";
 
@@ -79,5 +78,4 @@ class LoopTestUpdateComponent : public PollingComponent {
   int disable_loop_after_{0};
 };
 
-}  // namespace loop_test_component
-}  // namespace esphome
+}  // namespace esphome::loop_test_component

tests/integration/fixtures/external_components/loop_test_component/loop_test_isr_component.cpp

@@ -2,8 +2,7 @@
 #include "esphome/core/hal.h"
 #include "esphome/core/application.h"
 
-namespace esphome {
-namespace loop_test_component {
+namespace esphome::loop_test_component {
 
 static const char *const ISR_TAG = "loop_test_isr_component";
 
@@ -76,5 +75,4 @@ void IRAM_ATTR LoopTestISRComponent::simulate_isr_enable() {
   // For testing, we'll track the call count and log it from the main loop
 }
 
-}  // namespace loop_test_component
-}  // namespace esphome
+}  // namespace esphome::loop_test_component

tests/integration/fixtures/external_components/loop_test_component/loop_test_isr_component.h

@@ -4,8 +4,7 @@
 #include "esphome/core/log.h"
 #include "esphome/core/hal.h"
 
-namespace esphome {
-namespace loop_test_component {
+namespace esphome::loop_test_component {
 
 class LoopTestISRComponent : public Component {
  public:
@@ -28,5 +27,4 @@ class LoopTestISRComponent : public Component {
   int isr_call_count_{0};
 };
 
-}  // namespace loop_test_component
-}  // namespace esphome
+}  // namespace esphome::loop_test_component

tests/integration/fixtures/external_components/scheduler_bulk_cleanup_component/scheduler_bulk_cleanup_component.cpp

@@ -2,8 +2,7 @@
 #include "esphome/core/log.h"
 #include "esphome/core/helpers.h"
 
-namespace esphome {
-namespace scheduler_bulk_cleanup_component {
+namespace esphome::scheduler_bulk_cleanup_component {
 
 static const char *const TAG = "bulk_cleanup";
 
@@ -68,5 +67,4 @@ void SchedulerBulkCleanupComponent::trigger_bulk_cleanup() {
   }
 }
 
-}  // namespace scheduler_bulk_cleanup_component
-}  // namespace esphome
+}  // namespace esphome::scheduler_bulk_cleanup_component

tests/integration/fixtures/external_components/scheduler_bulk_cleanup_component/scheduler_bulk_cleanup_component.h

@@ -3,8 +3,7 @@
 #include "esphome/core/component.h"
 #include "esphome/core/application.h"
 
-namespace esphome {
-namespace scheduler_bulk_cleanup_component {
+namespace esphome::scheduler_bulk_cleanup_component {
 
 class SchedulerBulkCleanupComponent : public Component {
  public:
@@ -14,5 +13,4 @@ class SchedulerBulkCleanupComponent : public Component {
   void trigger_bulk_cleanup();
 };
 
-}  // namespace scheduler_bulk_cleanup_component
-}  // namespace esphome
+}  // namespace esphome::scheduler_bulk_cleanup_component

tests/integration/fixtures/external_components/scheduler_heap_stress_component/heap_scheduler_stress_component.cpp

@@ -6,8 +6,7 @@
 #include <chrono>
 #include <random>
 
-namespace esphome {
-namespace scheduler_heap_stress_component {
+namespace esphome::scheduler_heap_stress_component {
 
 static const char *const TAG = "scheduler_heap_stress";
 
@@ -100,5 +99,4 @@ void SchedulerHeapStressComponent::run_multi_thread_test() {
   ESP_LOGI(TAG, "All threads finished in %lldms. Created %d callbacks", thread_time, this->total_callbacks_.load());
 }
 
-}  // namespace scheduler_heap_stress_component
-}  // namespace esphome
+}  // namespace esphome::scheduler_heap_stress_component

tests/integration/fixtures/external_components/scheduler_heap_stress_component/heap_scheduler_stress_component.h

@@ -3,8 +3,7 @@
 #include "esphome/core/component.h"
 #include <atomic>
 
-namespace esphome {
-namespace scheduler_heap_stress_component {
+namespace esphome::scheduler_heap_stress_component {
 
 class SchedulerHeapStressComponent : public Component {
  public:
@@ -18,5 +17,4 @@ class SchedulerHeapStressComponent : public Component {
   std::atomic<int> executed_callbacks_{0};
 };
 
-}  // namespace scheduler_heap_stress_component
-}  // namespace esphome
+}  // namespace esphome::scheduler_heap_stress_component

tests/integration/fixtures/external_components/scheduler_rapid_cancellation_component/rapid_cancellation_component.cpp

@@ -6,8 +6,7 @@
 #include <random>
 #include <sstream>
 
-namespace esphome {
-namespace scheduler_rapid_cancellation_component {
+namespace esphome::scheduler_rapid_cancellation_component {
 
 static const char *const TAG = "scheduler_rapid_cancellation";
 
@@ -76,5 +75,4 @@ void SchedulerRapidCancellationComponent::run_rapid_cancellation_test() {
   });
 }
 
-}  // namespace scheduler_rapid_cancellation_component
-}  // namespace esphome
+}  // namespace esphome::scheduler_rapid_cancellation_component

tests/integration/fixtures/external_components/scheduler_rapid_cancellation_component/rapid_cancellation_component.h

@@ -3,8 +3,7 @@
 #include "esphome/core/component.h"
 #include <atomic>
 
-namespace esphome {
-namespace scheduler_rapid_cancellation_component {
+namespace esphome::scheduler_rapid_cancellation_component {
 
 class SchedulerRapidCancellationComponent : public Component {
  public:
@@ -18,5 +17,4 @@ class SchedulerRapidCancellationComponent : public Component {
   std::atomic<int> total_executed_{0};
 };
 
-}  // namespace scheduler_rapid_cancellation_component
-}  // namespace esphome
+}  // namespace esphome::scheduler_rapid_cancellation_component

tests/integration/fixtures/external_components/scheduler_recursive_timeout_component/recursive_timeout_component.cpp

@@ -1,8 +1,7 @@
 #include "recursive_timeout_component.h"
 #include "esphome/core/log.h"
 
-namespace esphome {
-namespace scheduler_recursive_timeout_component {
+namespace esphome::scheduler_recursive_timeout_component {
 
 static const char *const TAG = "scheduler_recursive_timeout";
 
@@ -36,5 +35,4 @@ void SchedulerRecursiveTimeoutComponent::run_recursive_timeout_test() {
   });
 }
 
-}  // namespace scheduler_recursive_timeout_component
-}  // namespace esphome
+}  // namespace esphome::scheduler_recursive_timeout_component

tests/integration/fixtures/external_components/scheduler_recursive_timeout_component/recursive_timeout_component.h

@@ -2,8 +2,7 @@
 
 #include "esphome/core/component.h"
 
-namespace esphome {
-namespace scheduler_recursive_timeout_component {
+namespace esphome::scheduler_recursive_timeout_component {
 
 class SchedulerRecursiveTimeoutComponent : public Component {
  public:
@@ -16,5 +15,4 @@ class SchedulerRecursiveTimeoutComponent : public Component {
   int nested_level_{0};
 };
 
-}  // namespace scheduler_recursive_timeout_component
-}  // namespace esphome
+}  // namespace esphome::scheduler_recursive_timeout_component

tests/integration/fixtures/external_components/scheduler_simultaneous_callbacks_component/simultaneous_callbacks_component.cpp

@@ -5,8 +5,7 @@
 #include <chrono>
 #include <sstream>
 
-namespace esphome {
-namespace scheduler_simultaneous_callbacks_component {
+namespace esphome::scheduler_simultaneous_callbacks_component {
 
 static const char *const TAG = "scheduler_simultaneous_callbacks";
 
@@ -105,5 +104,4 @@ void SchedulerSimultaneousCallbacksComponent::run_simultaneous_callbacks_test()
   });
 }
 
-}  // namespace scheduler_simultaneous_callbacks_component
-}  // namespace esphome
+}  // namespace esphome::scheduler_simultaneous_callbacks_component

tests/integration/fixtures/external_components/scheduler_simultaneous_callbacks_component/simultaneous_callbacks_component.h

@@ -3,8 +3,7 @@
 #include "esphome/core/component.h"
 #include <atomic>
 
-namespace esphome {
-namespace scheduler_simultaneous_callbacks_component {
+namespace esphome::scheduler_simultaneous_callbacks_component {
 
 class SchedulerSimultaneousCallbacksComponent : public Component {
  public:
@@ -20,5 +19,4 @@ class SchedulerSimultaneousCallbacksComponent : public Component {
   std::atomic<int> max_concurrent_{0};
 };
 
-}  // namespace scheduler_simultaneous_callbacks_component
-}  // namespace esphome
+}  // namespace esphome::scheduler_simultaneous_callbacks_component

tests/integration/fixtures/external_components/scheduler_string_lifetime_component/string_lifetime_component.cpp

@@ -4,8 +4,7 @@
 #include <thread>
 #include <chrono>
 
-namespace esphome {
-namespace scheduler_string_lifetime_component {
+namespace esphome::scheduler_string_lifetime_component {
 
 static const char *const TAG = "scheduler_string_lifetime";
 
@@ -258,5 +257,4 @@ void SchedulerStringLifetimeComponent::test_lambda_capture_lifetime() {
   });
 }
 
-}  // namespace scheduler_string_lifetime_component
-}  // namespace esphome
+}  // namespace esphome::scheduler_string_lifetime_component

tests/integration/fixtures/external_components/scheduler_string_lifetime_component/string_lifetime_component.h

@@ -4,8 +4,7 @@
 #include <vector>
 #include <string>
 
-namespace esphome {
-namespace scheduler_string_lifetime_component {
+namespace esphome::scheduler_string_lifetime_component {
 
 class SchedulerStringLifetimeComponent : public Component {
  public:
@@ -33,5 +32,4 @@ class SchedulerStringLifetimeComponent : public Component {
   int tests_failed_{0};
 };
 
-}  // namespace scheduler_string_lifetime_component
-}  // namespace esphome
+}  // namespace esphome::scheduler_string_lifetime_component

tests/integration/fixtures/external_components/scheduler_string_name_stress_component/string_name_stress_component.cpp

@@ -7,8 +7,7 @@
 #include <string>
 #include <sstream>
 
-namespace esphome {
-namespace scheduler_string_name_stress_component {
+namespace esphome::scheduler_string_name_stress_component {
 
 static const char *const TAG = "scheduler_string_name_stress";
 
@@ -106,5 +105,4 @@ void SchedulerStringNameStressComponent::run_string_name_stress_test() {
   });
 }
 
-}  // namespace scheduler_string_name_stress_component
-}  // namespace esphome
+}  // namespace esphome::scheduler_string_name_stress_component

tests/integration/fixtures/external_components/scheduler_string_name_stress_component/string_name_stress_component.h

@@ -3,8 +3,7 @@
 #include "esphome/core/component.h"
 #include <atomic>
 
-namespace esphome {
-namespace scheduler_string_name_stress_component {
+namespace esphome::scheduler_string_name_stress_component {
 
 class SchedulerStringNameStressComponent : public Component {
  public:
@@ -18,5 +17,4 @@ class SchedulerStringNameStressComponent : public Component {
   std::atomic<int> executed_callbacks_{0};
 };
 
-}  // namespace scheduler_string_name_stress_component
-}  // namespace esphome
+}  // namespace esphome::scheduler_string_name_stress_component

tests/integration/fixtures/host_ota_rejects_garbage.yaml

@@ -0,0 +1,9 @@
+esphome:
+  name: host-ota-test
+host:
+api:
+ota:
+  - platform: esphome
+    port: __OTA_PORT__
+logger:
+  level: DEBUG

tests/integration/fixtures/host_ota_self_update.yaml

@@ -0,0 +1,9 @@
+esphome:
+  name: host-ota-test
+host:
+api:
+ota:
+  - platform: esphome
+    port: __OTA_PORT__
+logger:
+  level: DEBUG

tests/integration/fixtures/script_array_params.yaml

@@ -0,0 +1,36 @@
+esphome:
+  name: test-script-array-params
+
+host:
+
+api:
+  actions:
+    - action: run_array_script
+      then:
+        - script.execute:
+            id: array_script
+            ints: [42, 100]
+            floats: [1.5, 2.5]
+            bools: [true, false]
+            strings: ["hello", "world"]
+
+logger:
+  level: DEBUG
+
+script:
+  - id: array_script
+    parameters:
+      ints: int[]
+      floats: float[]
+      bools: bool[]
+      strings: string[]
+    then:
+      - lambda: |-
+          ESP_LOGI("test", "ints size=%u [0]=%d [1]=%d",
+                   (unsigned) ints.size(), ints[0], ints[1]);
+          ESP_LOGI("test", "floats size=%u [0]=%.2f [1]=%.2f",
+                   (unsigned) floats.size(), floats[0], floats[1]);
+          ESP_LOGI("test", "bools size=%u [0]=%d [1]=%d",
+                   (unsigned) bools.size(), (int) bools[0], (int) bools[1]);
+          ESP_LOGI("test", "strings size=%u [0]=%s [1]=%s",
+                   (unsigned) strings.size(), strings[0].c_str(), strings[1].c_str());

tests/integration/test_host_ota.py

@@ -0,0 +1,152 @@
+"""End-to-end OTA tests on the host platform.
+
+Exercises the native OTA protocol against a real host binary, then asserts
+pid is preserved across the post-OTA execv. A second OTA on the post-exec
+instance covers the FD_CLOEXEC path.
+"""
+
+from __future__ import annotations
+
+import asyncio
+from collections.abc import Generator
+from contextlib import contextmanager
+import socket
+
+import pytest
+
+from esphome import espota2
+
+from .conftest import run_binary, wait_and_connect_api_client
+from .const import LOCALHOST, PORT_POLL_INTERVAL, PORT_WAIT_TIMEOUT
+from .types import CompileFunction, ConfigWriter
+
+DEVICE_NAME = "host-ota-test"
+
+
+@contextmanager
+def _reserve_port() -> Generator[tuple[int, socket.socket]]:
+    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+    s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+    s.bind(("", 0))
+    try:
+        yield s.getsockname()[1], s
+    finally:
+        s.close()
+
+
+async def _wait_for_port(host: str, port: int, timeout: float) -> None:
+    """Poll until a TCP port accepts connections, or raise TimeoutError."""
+    loop = asyncio.get_running_loop()
+    deadline = loop.time() + timeout
+    while loop.time() < deadline:
+        try:
+            _, writer = await asyncio.open_connection(host, port)
+        except (ConnectionRefusedError, OSError):
+            await asyncio.sleep(PORT_POLL_INTERVAL)
+            continue
+        writer.close()
+        await writer.wait_closed()
+        return
+    raise TimeoutError(f"Port {port} on {host} did not open within {timeout}s")
+
+
+@pytest.mark.asyncio
+async def test_host_ota_self_update(
+    yaml_config: str,
+    write_yaml_config: ConfigWriter,
+    compile_esphome: CompileFunction,
+    reserved_tcp_port: tuple[int, socket.socket],
+) -> None:
+    """Self-OTA: upload the running binary back to itself, expect re-exec."""
+    api_port, api_socket = reserved_tcp_port
+    with _reserve_port() as (ota_port, ota_socket):
+        yaml_config = yaml_config.replace("__OTA_PORT__", str(ota_port))
+        config_path = await write_yaml_config(yaml_config)
+        binary_path = await compile_esphome(config_path)
+        api_socket.close()
+        ota_socket.close()
+
+        loop = asyncio.get_running_loop()
+        ota_staged = loop.create_future()
+        rebooted = loop.create_future()
+
+        def on_log(line: str) -> None:
+            if not ota_staged.done() and "OTA staged at" in line:
+                ota_staged.set_result(True)
+            if not rebooted.done() and "Rebooting safely" in line:
+                rebooted.set_result(True)
+
+        async with run_binary(binary_path, line_callback=on_log) as (proc, _lines):
+            await _wait_for_port(LOCALHOST, api_port, PORT_WAIT_TIMEOUT)
+            pid_before = proc.pid
+            async with wait_and_connect_api_client(port=api_port) as client:
+                info_before = await client.device_info()
+                assert info_before.name == DEVICE_NAME
+
+            # espota2 is blocking; run in executor.
+            rc, _ = await loop.run_in_executor(
+                None, espota2.run_ota, LOCALHOST, ota_port, None, binary_path
+            )
+            assert rc == 0, "espota2 reported failure"
+
+            await asyncio.wait_for(ota_staged, timeout=10.0)
+            await asyncio.wait_for(rebooted, timeout=10.0)
+            await _wait_for_port(LOCALHOST, api_port, PORT_WAIT_TIMEOUT)
+
+            # execv preserves pid; mismatch means external respawn.
+            assert proc.returncode is None, "process exited instead of execing"
+            assert proc.pid == pid_before
+
+            async with wait_and_connect_api_client(port=api_port) as client:
+                info_after = await client.device_info()
+                assert info_after.name == DEVICE_NAME
+                assert info_after.name == info_before.name
+
+            # Second OTA: catches FD_CLOEXEC regressions (EADDRINUSE on rebind).
+            rc, _ = await loop.run_in_executor(
+                None, espota2.run_ota, LOCALHOST, ota_port, None, binary_path
+            )
+            assert rc == 0, "second OTA failed -- listener leaked across execv"
+            await _wait_for_port(LOCALHOST, api_port, PORT_WAIT_TIMEOUT)
+            assert proc.pid == pid_before
+
+
+@pytest.mark.asyncio
+async def test_host_ota_rejects_garbage(
+    yaml_config: str,
+    write_yaml_config: ConfigWriter,
+    compile_esphome: CompileFunction,
+    reserved_tcp_port: tuple[int, socket.socket],
+    integration_test_dir,
+) -> None:
+    """Bogus payload is rejected and the device keeps running."""
+    api_port, api_socket = reserved_tcp_port
+    with _reserve_port() as (ota_port, ota_socket):
+        yaml_config = yaml_config.replace("__OTA_PORT__", str(ota_port))
+        config_path = await write_yaml_config(yaml_config)
+        binary_path = await compile_esphome(config_path)
+
+        # 192 bytes that are neither ELF nor Mach-O.
+        bogus_path = integration_test_dir / "bogus.bin"
+        bogus_path.write_bytes(b"NOT-AN-EXECUTABLE-AT-ALL" * 8)
+
+        api_socket.close()
+        ota_socket.close()
+
+        async with run_binary(binary_path) as (proc, _lines):
+            await _wait_for_port(LOCALHOST, api_port, PORT_WAIT_TIMEOUT)
+            pid_before = proc.pid
+
+            loop = asyncio.get_running_loop()
+            rc, _ = await loop.run_in_executor(
+                None, espota2.run_ota, LOCALHOST, ota_port, None, bogus_path
+            )
+            assert rc == 1
+
+            await asyncio.sleep(0.5)
+            assert proc.returncode is None, "process died on rejected OTA"
+            assert proc.pid == pid_before
+
+            async with wait_and_connect_api_client(port=api_port) as client:
+                info = await client.device_info()
+                assert info.name == DEVICE_NAME

tests/integration/test_script_array_params.py

@@ -0,0 +1,71 @@
+"""Integration test for script array parameters (issue #16367).
+
+Verifies that script parameters of array types (`int[]`, `float[]`, `bool[]`,
+`string[]`) compile and execute correctly. Prior to the fix in
+`esphome/components/script/__init__.py`, the `script.execute` codegen emitted
+the Python `repr` of the list (e.g. `return [42, 100];`) instead of a C++
+braced initializer, causing compile failures.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import re
+
+import pytest
+
+from .types import APIClientConnectedFactory, RunCompiledFunction
+
+
+@pytest.mark.asyncio
+async def test_script_array_params(
+    yaml_config: str,
+    run_compiled: RunCompiledFunction,
+    api_client_connected: APIClientConnectedFactory,
+) -> None:
+    """Execute a script with int[], float[], bool[], string[] parameters."""
+    loop = asyncio.get_running_loop()
+    seen: dict[str, str] = {}
+    done = loop.create_future()
+
+    patterns = {
+        "ints": re.compile(r"ints size=(\d+) \[0\]=(-?\d+) \[1\]=(-?\d+)"),
+        "floats": re.compile(
+            r"floats size=(\d+) \[0\]=(-?\d+\.\d+) \[1\]=(-?\d+\.\d+)"
+        ),
+        "bools": re.compile(r"bools size=(\d+) \[0\]=(\d+) \[1\]=(\d+)"),
+        "strings": re.compile(r"strings size=(\d+) \[0\]=(\w+) \[1\]=(\w+)"),
+    }
+
+    def check_output(line: str) -> None:
+        for key, pat in patterns.items():
+            if (m := pat.search(line)) and key not in seen:
+                seen[key] = m.group(0)
+        if len(seen) == len(patterns) and not done.done():
+            done.set_result(True)
+
+    async with (
+        run_compiled(yaml_config, line_callback=check_output),
+        api_client_connected() as client,
+    ):
+        _, services = await client.list_entities_services()
+        service = next((s for s in services if s.name == "run_array_script"), None)
+        assert service is not None, "run_array_script service not found"
+        await client.execute_service(service, {})
+
+        try:
+            await asyncio.wait_for(done, timeout=5.0)
+        except TimeoutError:
+            pytest.fail(f"Did not receive all expected log lines. Saw: {seen}")
+
+        assert (m := patterns["ints"].search(seen["ints"]))
+        assert m.group(1) == "2" and m.group(2) == "42" and m.group(3) == "100"
+
+        assert (m := patterns["floats"].search(seen["floats"]))
+        assert m.group(1) == "2" and m.group(2) == "1.50" and m.group(3) == "2.50"
+
+        assert (m := patterns["bools"].search(seen["bools"]))
+        assert m.group(1) == "2" and m.group(2) == "1" and m.group(3) == "0"
+
+        assert (m := patterns["strings"].search(seen["strings"]))
+        assert m.group(1) == "2" and m.group(2) == "hello" and m.group(3) == "world"

tests/unit_tests/components/test_rp2040.py

@@ -0,0 +1,29 @@
+"""Tests for RP2040 component public helpers."""
+
+from esphome.components.rp2040 import board_id_has_wifi
+
+
+def test_board_id_has_wifi_for_known_wifi_board() -> None:
+    """``rpipicow`` is the canonical Pico W → True."""
+    assert board_id_has_wifi("rpipicow") is True
+
+
+def test_board_id_has_wifi_for_known_non_wifi_board() -> None:
+    """Plain ``rpipico`` has no CYW43 → False."""
+    assert board_id_has_wifi("rpipico") is False
+
+
+def test_board_id_has_wifi_for_rp2350_w_variant() -> None:
+    """``rpipico2w`` is the RP2350 Pico 2 W → True."""
+    assert board_id_has_wifi("rpipico2w") is True
+
+
+def test_board_id_has_wifi_for_unknown_board_returns_true() -> None:
+    """Unknown ids fail open so a custom board is not rejected.
+
+    The validator falls back to ESPHome's compile-time check; the
+    helper returning True here means the wizard emits a ``wifi:``
+    block and any genuinely-unsupported config trips the existing
+    "no CYW43" guard at compile time.
+    """
+    assert board_id_has_wifi("not-a-real-board-id") is True

tests/unit_tests/components/test_wifi.py

@@ -0,0 +1,125 @@
+"""Tests for WiFi component public helpers."""
+
+import pytest
+
+from esphome.components.esp32 import const
+from esphome.components.wifi import has_native_wifi, variant_has_wifi
+from esphome.const import Platform
+
+
+@pytest.mark.parametrize(
+    "variant",
+    [
+        # Upstream's canonical uppercase form.
+        const.VARIANT_ESP32,
+        const.VARIANT_ESP32S2,
+        const.VARIANT_ESP32S3,
+        const.VARIANT_ESP32C3,
+        const.VARIANT_ESP32C6,
+        # Lowercase form external callers (e.g. device-builder's
+        # ``Esp32Variant`` StrEnum) surface.
+        "esp32",
+        "esp32s3",
+        "esp32c3",
+        # Mixed-case — defence in depth against future callers that
+        # pull the value off some other serialisation.
+        "Esp32",
+    ],
+)
+def test_variant_has_wifi_for_native_phy_variants(variant: str) -> None:
+    """Variants with a native WiFi PHY → True, case-insensitive."""
+    assert variant_has_wifi(variant) is True
+
+
+@pytest.mark.parametrize(
+    "variant",
+    [
+        # Upstream's canonical uppercase form.
+        const.VARIANT_ESP32H2,
+        const.VARIANT_ESP32P4,
+        # Lowercase form external callers (e.g. device-builder's
+        # ``Esp32Variant`` StrEnum) surface.
+        "esp32h2",
+        "esp32p4",
+        # Mixed-case — defence in depth against future callers that
+        # pull the value off some other serialisation.
+        "Esp32H2",
+    ],
+)
+def test_variant_has_wifi_for_no_phy_variants(variant: str) -> None:
+    """Variants that need ``esp32_hosted`` → False, case-insensitive."""
+    assert variant_has_wifi(variant) is False
+
+
+def test_has_native_wifi_dispatches_esp32_to_variant_check() -> None:
+    """ESP32 platform routes through ``variant_has_wifi``."""
+    assert (
+        has_native_wifi(platform=Platform.ESP32, variant=const.VARIANT_ESP32C3) is True
+    )
+    assert (
+        has_native_wifi(platform=Platform.ESP32, variant=const.VARIANT_ESP32H2) is False
+    )
+
+
+def test_has_native_wifi_esp32_variant_case_insensitive() -> None:
+    """has_native_wifi accepts lowercase variant input.
+
+    External callers (device-builder's wizard, etc.) may surface
+    variant strings from their own enums that don't match upstream's
+    uppercase convention. The dispatcher should classify them
+    identically.
+    """
+    assert has_native_wifi(platform=Platform.ESP32, variant="esp32h2") is False
+    assert has_native_wifi(platform=Platform.ESP32, variant="esp32c3") is True
+
+
+def test_has_native_wifi_dispatches_rp2040_to_board_check() -> None:
+    """RP2040 platform routes through ``rp2040.board_id_has_wifi``."""
+    assert has_native_wifi(platform=Platform.RP2040, board="rpipicow") is True
+    assert has_native_wifi(platform=Platform.RP2040, board="rpipico") is False
+
+
+def test_has_native_wifi_returns_false_for_nrf52() -> None:
+    """nRF52 family is BLE-only — no Wi-Fi PHY in the platform."""
+    assert has_native_wifi(platform=Platform.NRF52) is False
+
+
+def test_has_native_wifi_returns_false_for_host() -> None:
+    """``host`` platform compiles ESPHome to a host binary — no radio at all."""
+    assert has_native_wifi(platform=Platform.HOST) is False
+
+
+def test_has_native_wifi_returns_false_for_unknown_platform() -> None:
+    """Unknown platform string fails closed.
+
+    A future platform added to ESPHome that's missed here returns
+    False rather than silently emitting a ``wifi:`` block external
+    tooling would have to compile and reject — fail-closed surfaces
+    the gap as an obvious "needs wifi support added" signal.
+    """
+    assert has_native_wifi(platform="not-a-real-platform") is False
+
+
+@pytest.mark.parametrize(
+    "platform",
+    [
+        Platform.ESP8266,
+        Platform.BK72XX,
+        Platform.RTL87XX,
+        Platform.LN882X,
+        Platform.LIBRETINY_OLDSTYLE,
+    ],
+)
+def test_has_native_wifi_returns_true_for_wifi_first_platforms(platform: str) -> None:
+    """Catch-all Wi-Fi-first platforms → True regardless of board / variant."""
+    assert has_native_wifi(platform=platform) is True
+
+
+def test_has_native_wifi_esp32_without_variant_assumes_wifi() -> None:
+    """ESP32 without a variant id falls open to True (the chip family default)."""
+    assert has_native_wifi(platform=Platform.ESP32) is True
+
+
+def test_has_native_wifi_rp2040_without_board_assumes_wifi() -> None:
+    """RP2040 without a board id falls open to True (custom-board default)."""
+    assert has_native_wifi(platform=Platform.RP2040) is True

tests/unit_tests/conftest.py

@@ -64,15 +64,15 @@ def mock_copy_file_if_changed() -> Generator[Mock, None, None]:
 
 @pytest.fixture
 def mock_run_platformio_cli() -> Generator[Mock, None, None]:
-    """Mock run_platformio_cli for platformio_api."""
-    with patch("esphome.platformio_api.run_platformio_cli") as mock:
+    """Mock run_platformio_cli for platformio toolchain."""
+    with patch("esphome.platformio.toolchain.run_platformio_cli") as mock:
         yield mock
 
 
 @pytest.fixture
 def mock_run_platformio_cli_run() -> Generator[Mock, None, None]:
-    """Mock run_platformio_cli_run for platformio_api."""
-    with patch("esphome.platformio_api.run_platformio_cli_run") as mock:
+    """Mock run_platformio_cli_run for platformio toolchain."""
+    with patch("esphome.platformio.toolchain.run_platformio_cli_run") as mock:
         yield mock
 
 
@@ -92,8 +92,8 @@ def mock_esp8266_decode_pc() -> Generator[Mock, None, None]:
 
 @pytest.fixture
 def mock_run_external_process() -> Generator[Mock, None, None]:
-    """Mock run_external_process for platformio_api."""
-    with patch("esphome.platformio_api.run_external_process") as mock:
+    """Mock run_external_process for platformio toolchain."""
+    with patch("esphome.platformio.toolchain.run_external_process") as mock:
         yield mock
 
 
@@ -113,8 +113,8 @@ def mock_subprocess_run() -> Generator[Mock, None, None]:
 
 @pytest.fixture
 def mock_get_idedata() -> Generator[Mock, None, None]:
-    """Mock get_idedata for platformio_api."""
-    with patch("esphome.platformio_api.get_idedata") as mock:
+    """Mock get_idedata for platformio toolchain."""
+    with patch("esphome.platformio.toolchain.get_idedata") as mock:
         yield mock
 
 

tests/unit_tests/test_config_validation.py

@@ -793,3 +793,187 @@ def test_update_interval__never_passes_through() -> None:
     """update_interval: never must still map to SCHEDULER_DONT_RUN."""
     result = config_validation.update_interval("never")
     assert result.total_milliseconds == SCHEDULER_DONT_RUN
+
+
+# ---------------------------------------------------------------------------
+# Visibility UI-hint kwarg
+# ---------------------------------------------------------------------------
+
+
+def test_optional_default_visibility_is_none() -> None:
+    """An ``Optional`` with no ``visibility`` kwarg reports ``None``.
+
+    Consumers can read the attribute directly with plain attribute
+    access; absence (``None``) means "render on the editor's main
+    form."
+    """
+    o = config_validation.Optional("foo")
+    assert o.visibility is None
+
+
+def test_optional_visibility_advanced() -> None:
+    """``visibility=Visibility.ADVANCED`` is recorded on the marker."""
+    o = config_validation.Optional(
+        "foo", visibility=config_validation.Visibility.ADVANCED
+    )
+    assert o.visibility is config_validation.Visibility.ADVANCED
+
+
+def test_optional_visibility_yaml_only() -> None:
+    """``visibility=Visibility.YAML_ONLY`` is recorded on the marker."""
+    o = config_validation.Optional(
+        "foo", visibility=config_validation.Visibility.YAML_ONLY
+    )
+    assert o.visibility is config_validation.Visibility.YAML_ONLY
+
+
+def test_visibility_str_values_match_dump_emission() -> None:
+    """``Visibility`` is a ``StrEnum`` whose values are the literal
+    strings the schema dumper emits.
+
+    The schema bundle consumers (catalog generators, third-party
+    schema-aware tooling) shouldn't need an enum import to read the
+    field — pinning the on-the-wire spelling here keeps the dump
+    contract stable.
+    """
+    assert str(config_validation.Visibility.ADVANCED) == "advanced"
+    assert str(config_validation.Visibility.YAML_ONLY) == "yaml_only"
+
+
+def test_optional_visibility_does_not_affect_validation() -> None:
+    """The kwarg is an advisory UI hint — it must not change how the
+    validator behaves. A schema with ``visibility`` applied must
+    accept and reject the same values it would without it.
+    """
+    plain = config_validation.Schema(
+        {config_validation.Optional("foo", default=42): config_validation.int_}
+    )
+    flagged = config_validation.Schema(
+        {
+            config_validation.Optional(
+                "foo",
+                default=42,
+                visibility=config_validation.Visibility.YAML_ONLY,
+            ): config_validation.int_
+        }
+    )
+    # Same accept / default-fill behavior.
+    assert plain({"foo": 7}) == flagged({"foo": 7}) == {"foo": 7}
+    assert plain({}) == flagged({}) == {"foo": 42}
+    # Same rejection on bad input.
+    with pytest.raises(Invalid):
+        plain({"foo": "not-an-int"})
+    with pytest.raises(Invalid):
+        flagged({"foo": "not-an-int"})
+
+
+def test_required_default_visibility_is_none() -> None:
+    """``Required`` mirrors ``Optional`` for the ``visibility`` kwarg."""
+    r = config_validation.Required("foo")
+    assert r.visibility is None
+
+
+def test_required_visibility_kwarg() -> None:
+    """``Required`` accepts ``visibility`` for symmetry with ``Optional``.
+
+    Required fields rarely need the kwarg, but exposing it lets
+    consumers apply uniform logic across key markers.
+    """
+    r = config_validation.Required(
+        "foo", visibility=config_validation.Visibility.ADVANCED
+    )
+    assert r.visibility is config_validation.Visibility.ADVANCED
+
+
+def test_polling_component_schema_visibility_opt_in() -> None:
+    """``visibility=`` propagates to the inherited ``update_interval``.
+
+    Time platforms pass ``Visibility.ADVANCED``; sensors and other
+    polling components leave it ``None`` and keep the un-flagged shape.
+    """
+    default = config_validation.polling_component_schema("15min")
+    advanced = config_validation.polling_component_schema(
+        "15min", visibility=config_validation.Visibility.ADVANCED
+    )
+    default_keys = {str(k): k for k in default.schema}
+    advanced_keys = {str(k): k for k in advanced.schema}
+    assert default_keys["update_interval"].visibility is None
+    assert (
+        advanced_keys["update_interval"].visibility
+        is config_validation.Visibility.ADVANCED
+    )
+    # The opt-in only touches update_interval — setup_priority
+    # still inherits its YAML_ONLY visibility from COMPONENT_SCHEMA
+    # in both shapes.
+    assert (
+        default_keys["setup_priority"].visibility
+        is config_validation.Visibility.YAML_ONLY
+    )
+    assert (
+        advanced_keys["setup_priority"].visibility
+        is config_validation.Visibility.YAML_ONLY
+    )
+
+
+def test_polling_component_schema_no_default_ignores_visibility() -> None:
+    """``visibility`` is silently ignored when the field is Required.
+
+    When ``default_update_interval=None`` the field becomes
+    ``Required``. Hiding a Required field behind an advanced
+    disclosure is a UX hazard — a collapsed-by-default editor could
+    let the user submit without noticing the form has an unfilled
+    required field. The helper accepts the kwarg unconditionally
+    for caller ergonomics but doesn't honour it on this branch.
+    """
+    schema = config_validation.polling_component_schema(
+        None, visibility=config_validation.Visibility.ADVANCED
+    )
+    keys = {str(k): k for k in schema.schema}
+    assert isinstance(keys["update_interval"], config_validation.Required)
+    assert keys["update_interval"].visibility is None
+
+
+def test_visibility_marker_is_per_field_no_mutation() -> None:
+    """Each field's ``visibility`` is recorded as the author wrote it.
+
+    Cascading semantics — "a stricter parent forces its descendants
+    at-least as strict" — live on the consumer side, not in the
+    marker itself. The schema marker stays as-written so consumers
+    can walk the parent chain and compute the effective visibility
+    themselves; mutating the marker would lose the per-field author
+    intent.
+
+    Pin both directions of the no-mutation contract: an inner
+    ``YAML_ONLY`` under an ``ADVANCED`` parent stays ``YAML_ONLY``
+    on the marker (the consumer's effective-visibility cascade
+    would also report ``YAML_ONLY`` since it's stricter), and an
+    un-marked inner field stays ``None`` on the marker (the
+    cascade's job is to compute ``ADVANCED`` from the parent — a
+    detail this test deliberately doesn't pin, since it's a
+    consumer concern).
+    """
+    inner_unset = config_validation.Optional("baz")
+    inner_yaml_only = config_validation.Optional(
+        "qux", visibility=config_validation.Visibility.YAML_ONLY
+    )
+    parent = config_validation.Optional(
+        "foo", visibility=config_validation.Visibility.ADVANCED
+    )
+
+    # Wire them into a nested schema — none of the markers' own
+    # ``visibility`` should change as a result.
+    schema = config_validation.Schema(
+        {
+            parent: config_validation.Schema(
+                {
+                    inner_unset: config_validation.int_,
+                    inner_yaml_only: config_validation.string,
+                }
+            )
+        }
+    )
+    assert schema  # touch the schema so any deferred mutation runs
+
+    assert parent.visibility is config_validation.Visibility.ADVANCED
+    assert inner_unset.visibility is None
+    assert inner_yaml_only.visibility is config_validation.Visibility.YAML_ONLY

tests/unit_tests/test_core.py

@@ -592,6 +592,30 @@ class TestEsphomeCore:
         assert target.is_esp32 is False
         assert target.is_esp8266 is True
 
+    def test_firmware_bin__default(self, target):
+        """Default platforms produce <pioenvs>/<name>/firmware.bin."""
+        target.name = "test-device"
+        target.data[const.KEY_CORE] = {const.KEY_TARGET_PLATFORM: "esp32"}
+        assert target.firmware_bin == Path(
+            "foo/build/.pioenvs/test-device/firmware.bin"
+        )
+
+    def test_firmware_bin__libretiny(self, target):
+        """The libretiny platform produces firmware.uf2."""
+        target.name = "test-device"
+        target.data[const.KEY_CORE] = {const.KEY_TARGET_PLATFORM: "bk72xx"}
+        assert target.firmware_bin == Path(
+            "foo/build/.pioenvs/test-device/firmware.uf2"
+        )
+
+    def test_firmware_bin__host(self, target):
+        """Host platform produces a native ELF/Mach-O named `program`,
+        not firmware.bin -- needed for `esphome upload` to find the
+        right artifact for the host OTA backend."""
+        target.name = "test-device"
+        target.data[const.KEY_CORE] = {const.KEY_TARGET_PLATFORM: "host"}
+        assert target.firmware_bin == Path("foo/build/.pioenvs/test-device/program")
+
     @pytest.mark.skipif(os.name == "nt", reason="Unix-specific test")
     def test_data_dir_default_unix(self, target):
         """Test data_dir returns .esphome in config directory by default on Unix."""
@@ -854,6 +878,23 @@ class TestEsphomeCore:
         target.testing_ensure_platform_registered("sensor")
         assert target.platform_counts["sensor"] == 3
 
+    def test_bootloader_bin__native_idf(self, target):
+        """Native ESP-IDF builds emit the bootloader under build/bootloader/bootloader.bin."""
+        target.toolchain = const.Toolchain.ESP_IDF
+
+        assert target.bootloader_bin == Path(
+            "foo/build/build/bootloader/bootloader.bin"
+        )
+
+    def test_bootloader_bin__platformio(self, target):
+        """For PlatformIO builds bootloader.bin lives in the env-specific .pioenvs directory."""
+        target.name = "test-device"
+        target.toolchain = const.Toolchain.PLATFORMIO
+
+        assert target.bootloader_bin == Path(
+            "foo/build/.pioenvs/test-device/bootloader.bin"
+        )
+
     def test_add_library__extracts_short_name_from_path(self, target):
         """Test add_library extracts short name from library paths like owner/lib."""
         target.data[const.KEY_CORE] = {

tests/unit_tests/test_espidf_component.py

@@ -0,0 +1,357 @@
+import json
+import os
+from unittest.mock import MagicMock
+
+import pytest
+
+from esphome.const import (
+    KEY_CORE,
+    KEY_TARGET_FRAMEWORK,
+    KEY_TARGET_PLATFORM,
+    Framework,
+    Platform,
+)
+from esphome.core import CORE, Library
+import esphome.espidf.component
+from esphome.espidf.component import (
+    GitSource,
+    IDFComponent,
+    InvalidIDFComponent,
+    URLSource,
+    _check_library_data,
+    _collect_filtered_files,
+    _convert_library_to_component,
+    _detect_requires,
+    _parse_library_json,
+    _parse_library_properties,
+    _process_dependencies,
+    _split_list_by_condition,
+    generate_cmakelists_txt,
+    generate_idf_component_yml,
+)
+
+
+@pytest.fixture(name="tmp_component")
+def fixture_tmp_component(tmp_path):
+    c = IDFComponent("owner/name", "1.0.0", source=MagicMock())
+    c.path = tmp_path
+    return c
+
+
+@pytest.fixture(name="esp32_idf_core")
+def fixture_esp32_idf_core():
+    CORE.data[KEY_CORE] = {}
+    CORE.data[KEY_CORE][KEY_TARGET_PLATFORM] = str(Platform.ESP32)
+    CORE.data[KEY_CORE][KEY_TARGET_FRAMEWORK] = str(Framework.ESP_IDF)
+
+
+def test_idf_component_str():
+    c = IDFComponent("foo/bar", "1.0", source=URLSource("http://dummy.com"))
+    assert str(c) == "foo/bar@1.0=http://dummy.com"
+
+
+def test_idf_component_sanitized_name():
+    c = IDFComponent("foo/bar bar-bar", "1.0", source=URLSource("http://dummy.com"))
+    assert c.get_sanitized_name() == "foo/bar_bar-bar"
+
+
+def test_idf_component_require_name():
+    c = IDFComponent("foo/bar", "1.0", source=URLSource("http://dummy.com"))
+    assert c.get_require_name() == "foo__bar"
+
+
+def test_collect_filtered_files_basic(tmp_path):
+    f1 = tmp_path / "a.c"
+    f2 = tmp_path / "b" / "b.cpp"
+    f1.write_text("int a;")
+    f2.parent.mkdir(parents=True)
+    f2.write_text("int b;")
+
+    result = _collect_filtered_files(tmp_path, ["+<*>"])
+    assert str(f1) in result
+    assert str(f2) in result
+
+
+def test_collect_filtered_files_exclude(tmp_path):
+    f1 = tmp_path / "a.c"
+    f2 = tmp_path / "b.cpp"
+    f1.write_text("int a;")
+    f2.write_text("int b;")
+
+    result = _collect_filtered_files(tmp_path, ["+<*> -<*.cpp>"])
+    assert str(f1) in result
+    assert str(f2) not in result
+
+
+def test_detect_requires(tmp_path):
+    f = tmp_path / "main.c"
+    f.write_text('#include "mbedtls/foo.h"')
+
+    result = _detect_requires([str(f)])
+    assert "mbedtls" in result
+
+
+def test_detect_requires_ignores_invalid_file(tmp_path):
+    result = _detect_requires([str(tmp_path / "missing.c")])
+    assert result == set()
+
+
+def test_split_list_by_condition():
+    items = ["-Iinclude", "-Llib", "-Wall"]
+
+    matched, rest = _split_list_by_condition(
+        items, lambda x: x[2:] if x.startswith("-I") else None
+    )
+
+    assert matched == ["include"]
+    assert "-Llib" in rest
+    assert "-Wall" in rest
+
+
+def test_generate_cmakelists_txt_basic(tmp_component):
+    src_dir = tmp_component.path / "src"
+    src_dir.mkdir()
+    f = src_dir / "main.c"
+    f.write_text("int main() {}")
+
+    tmp_component.data = {}
+
+    content = generate_cmakelists_txt(tmp_component)
+
+    assert "idf_component_register" in content
+    assert "main.c" in content
+
+
+def test_generate_cmakelists_txt_with_flags(tmp_component, tmp_path):
+    src_dir = tmp_component.path / "src"
+    src_dir.mkdir()
+    (src_dir / "main.c").write_text("int main() {}")
+
+    dep = IDFComponent("dep", "1.0", source=URLSource("http://dummy.com"))
+    dep.path = tmp_path / "dep"
+    tmp_component.dependencies = [dep]
+
+    tmp_component.data = {
+        "build": {"flags": ["-Iinclude", "-Llib", "-lmylib", "-Wall", "-DTEST"]}
+    }
+
+    content = generate_cmakelists_txt(tmp_component)
+    sep = "\\\\" if os.name == "nt" else "/"
+    assert (
+        content
+        == f"""idf_component_register(
+  SRCS "src{sep}main.c"
+  INCLUDE_DIRS "src"
+  REQUIRES dep
+)
+target_compile_options(${{COMPONENT_LIB}} PUBLIC
+  "-DTEST"
+)
+target_compile_options(${{COMPONENT_LIB}} PRIVATE
+  "-Wall"
+)
+target_link_directories(${{COMPONENT_LIB}} INTERFACE
+  "lib"
+)
+target_link_libraries(${{COMPONENT_LIB}} INTERFACE
+  "mylib"
+)
+"""
+    )
+
+
+def test_generate_idf_component_yml_basic(tmp_component):
+    tmp_component.data = {"description": "test", "repository": {"url": "http://aaa"}}
+    result = generate_idf_component_yml(tmp_component)
+
+    assert result == "description: test\nversion: 1.0.0\nrepository: http://aaa\n"
+
+
+def test_generate_idf_component_yml_with_dependencies(tmp_component, tmp_path):
+    dep = IDFComponent("dep", "1.0", source=URLSource("http://dummy.com"))
+    dep.path = tmp_path / "dep"
+
+    tmp_component.dependencies = [dep]
+    tmp_component.data = {}
+
+    result = generate_idf_component_yml(tmp_component)
+
+    assert (
+        result
+        == f"""version: 1.0.0
+dependencies:
+  dep:
+    version: '1.0'
+    override_path: {dep.path}
+"""
+    )
+
+
+def test_generate_idf_component_yml_arduino_registry_dep(tmp_component):
+    # Synthetic arduino-esp32 dep with no source / no path: should emit a
+    # version-only entry so the IDF component manager resolves it from the
+    # registry instead of via git.
+    dep = IDFComponent("espressif/arduino-esp32", "3.3.8", source=None)
+
+    tmp_component.dependencies = [dep]
+    tmp_component.data = {}
+
+    result = generate_idf_component_yml(tmp_component)
+
+    assert (
+        result
+        == """version: 1.0.0
+dependencies:
+  espressif/arduino-esp32:
+    version: 3.3.8
+"""
+    )
+
+
+def test_generate_idf_component_yml_missing_path_reraises(tmp_component):
+    # A dep without a path and without a recognised source should re-raise
+    # the underlying RuntimeError instead of silently producing a bad manifest.
+    dep = IDFComponent("foo/bar", "1.0", source=None)
+
+    tmp_component.dependencies = [dep]
+    tmp_component.data = {}
+
+    with pytest.raises(RuntimeError):
+        generate_idf_component_yml(tmp_component)
+
+
+def test_check_library_data_valid(esp32_idf_core):
+    _check_library_data({"platforms": "*", "frameworks": "*"})
+
+
+def test_check_library_data_valid2(esp32_idf_core):
+    _check_library_data({"platforms": "*"})
+
+
+def test_check_library_data_valid3(esp32_idf_core):
+    _check_library_data({})
+
+
+def test_check_library_data_valid4(esp32_idf_core):
+    _check_library_data({"platforms": "espressif32", "frameworks": "*"})
+
+
+def test_check_library_data_valid5(esp32_idf_core):
+    _check_library_data({"platforms": "*", "frameworks": "espidf"})
+
+
+def test_check_library_data_invalid_platform(esp32_idf_core):
+    with pytest.raises(InvalidIDFComponent):
+        _check_library_data({"platforms": ["other"], "frameworks": "*"})
+
+
+def test_check_library_data_invalid_framework(esp32_idf_core):
+    with pytest.raises(InvalidIDFComponent):
+        _check_library_data({"platforms": "*", "frameworks": ["other"]})
+
+
+def test_extra_script_logs_warning(caplog, esp32_idf_core):
+    extra_script = "myscript.sh"
+
+    with caplog.at_level("WARNING"):
+        _check_library_data({"build": {"extraScript": extra_script}})
+
+    assert "not supported" in caplog.text
+    assert "myscript.sh" in caplog.text
+
+
+def test_parse_library_json(tmp_path):
+    f = tmp_path / "library.json"
+    f.write_text(json.dumps({"name": "test"}))
+
+    result = _parse_library_json(f)
+    assert result["name"] == "test"
+
+
+def test_parse_library_properties(tmp_path):
+    f = tmp_path / "library.properties"
+    f.write_text(
+        """
+name=Test
+version=1.0
+# description=ABCD
+empty=
+"""
+    )
+
+    result = _parse_library_properties(f)
+
+    assert result["name"] == "Test"
+    assert result["version"] == "1.0"
+    assert "empty" not in result
+
+
+def test_convert_library_with_repository():
+    lib = Library("name", None, "https://github.com/foo/bar.git#v1.2.3")
+
+    result = _convert_library_to_component(lib)
+
+    assert result.name == "foo/bar"
+    assert result.version == "1.2.3"
+    assert isinstance(result.source, GitSource)
+
+
+def test_convert_library_missing_ref():
+    lib = Library("name", None, "https://github.com/foo/bar.git")
+
+    with pytest.raises(ValueError):
+        _convert_library_to_component(lib)
+
+
+def test_convert_library_registry(monkeypatch):
+    lib = Library("foo/bar", "^1.0.0", None)
+
+    monkeypatch.setattr(
+        esphome.espidf.component,
+        "_get_package_from_pio_registry",
+        lambda o, n, r: ("foo", "bar", "1.2.3", "http://example.com/pkg.zip"),
+    )
+
+    result = _convert_library_to_component(lib)
+
+    assert result.name == "foo/bar"
+    assert result.version == "1.2.3"
+    assert isinstance(result.source, URLSource)
+
+
+def test_process_dependencies_adds_valid_dependency(tmp_component, monkeypatch):
+    tmp_component.data = {
+        "dependencies": [
+            {
+                "name": "foo",
+                "version": "1.0",
+            }
+        ]
+    }
+
+    monkeypatch.setattr(
+        esphome.espidf.component,
+        "_generate_idf_component",
+        lambda lib: esphome.espidf.component.IDFComponent(
+            lib.name, lib.version, source=URLSource("http://dummy.com")
+        ),
+    )
+
+    monkeypatch.setattr(esphome.espidf.component, "_check_library_data", lambda x: None)
+
+    _process_dependencies(tmp_component)
+
+    assert len(tmp_component.dependencies) == 1
+
+
+def test_process_dependencies_skips_invalid(tmp_component):
+    tmp_component.data = {
+        "dependencies": [
+            {"name": "foo", "version": "1.0", "platforms": ["arduino"]},
+            {"invalid": "entry"},
+        ]
+    }
+
+    _process_dependencies(tmp_component)
+
+    assert tmp_component.dependencies == []

tests/unit_tests/test_espota2.py

@@ -135,7 +135,9 @@ def test_receive_exactly_with_error_response(mock_socket: Mock) -> None:
     """Test receive_exactly raises OTAError on error response."""
     mock_socket.recv.return_value = bytes([espota2.RESPONSE_ERROR_AUTH_INVALID])
 
-    with pytest.raises(espota2.OTAError, match="Error auth:.*Authentication invalid"):
+    with pytest.raises(
+        espota2.OTAError, match="receiving auth:.*Authentication invalid"
+    ):
         espota2.receive_exactly(mock_socket, 1, "auth", [espota2.RESPONSE_OK])
 
     mock_socket.close.assert_called_once()
@@ -145,61 +147,69 @@ def test_receive_exactly_socket_error(mock_socket: Mock) -> None:
     """Test receive_exactly handles socket errors."""
     mock_socket.recv.side_effect = OSError("Connection reset")
 
-    with pytest.raises(espota2.OTAError, match="Error receiving acknowledge test"):
+    with pytest.raises(espota2.OTAError, match="receiving test response"):
         espota2.receive_exactly(mock_socket, 1, "test", espota2.RESPONSE_OK)
 
 
 @pytest.mark.parametrize(
     ("error_code", "expected_msg"),
     [
-        (espota2.RESPONSE_ERROR_MAGIC, "Error: Invalid magic byte"),
-        (espota2.RESPONSE_ERROR_UPDATE_PREPARE, "Error: Couldn't prepare flash memory"),
-        (espota2.RESPONSE_ERROR_AUTH_INVALID, "Error: Authentication invalid"),
+        (espota2.RESPONSE_ERROR_MAGIC, "Invalid magic byte"),
+        (espota2.RESPONSE_ERROR_UPDATE_PREPARE, "Couldn't prepare flash memory"),
+        (espota2.RESPONSE_ERROR_AUTH_INVALID, "Authentication invalid"),
         (
             espota2.RESPONSE_ERROR_WRITING_FLASH,
-            "Error: Writing OTA data to flash memory failed",
+            "Writing OTA data to flash memory failed",
         ),
-        (espota2.RESPONSE_ERROR_UPDATE_END, "Error: Finishing update failed"),
+        (espota2.RESPONSE_ERROR_UPDATE_END, "Finishing update failed"),
         (
             espota2.RESPONSE_ERROR_INVALID_BOOTSTRAPPING,
-            "Error: Please press the reset button",
+            "Please press the reset button",
         ),
         (
             espota2.RESPONSE_ERROR_WRONG_CURRENT_FLASH_CONFIG,
-            "Error: ESP has been flashed with wrong flash size",
+            "ESP has been flashed with wrong flash size",
         ),
         (
             espota2.RESPONSE_ERROR_WRONG_NEW_FLASH_CONFIG,
-            "Error: ESP does not have the requested flash size",
+            "ESP does not have the requested flash size",
         ),
         (
             espota2.RESPONSE_ERROR_ESP8266_NOT_ENOUGH_SPACE,
-            "Error: ESP does not have enough space",
+            "ESP does not have enough space",
         ),
         (
             espota2.RESPONSE_ERROR_ESP32_NOT_ENOUGH_SPACE,
-            "Error: The OTA partition on the ESP is too small",
+            "The OTA partition on the ESP is too small",
         ),
         (
             espota2.RESPONSE_ERROR_NO_UPDATE_PARTITION,
-            "Error: The OTA partition on the ESP couldn't be found",
+            "The OTA partition on the ESP couldn't be found",
         ),
-        (espota2.RESPONSE_ERROR_MD5_MISMATCH, "Error: Application MD5 code mismatch"),
+        (espota2.RESPONSE_ERROR_MD5_MISMATCH, "Application MD5 code mismatch"),
         (
             espota2.RESPONSE_ERROR_SIGNATURE_INVALID,
-            "Error: Firmware signature verification failed",
+            "Firmware signature verification failed",
         ),
         (
             espota2.RESPONSE_ERROR_UNSUPPORTED_OTA_TYPE,
-            "Error: The requested OTA type is not supported by the device",
+            "The requested OTA type is not supported by the device",
         ),
         (
             espota2.RESPONSE_ERROR_PARTITION_TABLE_VERIFY,
-            "Error: The partition table update could not be verified",
+            "The partition table update could not be verified",
         ),
         (
             espota2.RESPONSE_ERROR_PARTITION_TABLE_UPDATE,
-            "Error: An error occurred while updating the partition table",
+            "An error occurred while updating the partition table",
+        ),
+        (
+            espota2.RESPONSE_ERROR_BOOTLOADER_VERIFY,
+            "The bootloader update could not be verified",
+        ),
+        (
+            espota2.RESPONSE_ERROR_BOOTLOADER_UPDATE,
+            "An error occurred while updating the bootloader",
         ),
         (espota2.RESPONSE_ERROR_UNKNOWN, "Unknown error from ESP"),
     ],
@@ -254,7 +264,7 @@ def test_send_check_socket_error(mock_socket: Mock) -> None:
     """Test send_check handles socket errors."""
     mock_socket.sendall.side_effect = OSError("Broken pipe")
 
-    with pytest.raises(espota2.OTAError, match="Error sending test"):
+    with pytest.raises(espota2.OTAError, match="sending test"):
         espota2.send_check(mock_socket, b"data", "test")
 
 
@@ -409,7 +419,9 @@ def test_perform_ota_md5_auth_wrong_password(
 
     mock_socket.recv.side_effect = recv_responses
 
-    with pytest.raises(espota2.OTAError, match="Error auth.*Authentication invalid"):
+    with pytest.raises(
+        espota2.OTAError, match="receiving auth.*Authentication invalid"
+    ):
         espota2.perform_ota(mock_socket, "wrongpassword", mock_file, "test.bin")
 
     # Verify the socket was closed after auth failure
@@ -433,7 +445,9 @@ def test_perform_ota_sha256_auth_wrong_password(
 
     mock_socket.recv.side_effect = recv_responses
 
-    with pytest.raises(espota2.OTAError, match="Error auth.*Authentication invalid"):
+    with pytest.raises(
+        espota2.OTAError, match="receiving auth.*Authentication invalid"
+    ):
         espota2.perform_ota(mock_socket, "wrongpassword", mock_file, "test.bin")
 
     # Verify the socket was closed after auth failure
@@ -476,7 +490,7 @@ def test_perform_ota_unexpected_auth_response(mock_socket: Mock) -> None:
 
     # This will actually raise "Unexpected response from ESP" from check_error
     with pytest.raises(
-        espota2.OTAError, match=r"Error auth: Unexpected response from ESP: 0x03"
+        espota2.OTAError, match=r"receiving auth: Unexpected response from ESP: 0x03"
     ):
         espota2.perform_ota(mock_socket, "password", mock_file, "test.bin")
 
@@ -512,7 +526,7 @@ def test_perform_ota_upload_error(mock_socket: Mock, mock_file: io.BytesIO) -> N
 
     mock_socket.recv.side_effect = recv_responses
 
-    with pytest.raises(espota2.OTAError, match="Error receiving acknowledge chunk OK"):
+    with pytest.raises(espota2.OTAError, match="receiving chunk result response"):
         espota2.perform_ota(mock_socket, None, mock_file, "test.bin")
 
 
@@ -596,7 +610,8 @@ def test_run_ota_wrapper(mock_run_ota_impl: Mock) -> None:
 
 def test_progress_bar(capsys: CaptureFixture[str]) -> None:
     """Test ProgressBar functionality."""
-    progress = espota2.ProgressBar()
+    progress = espota2.ProgressBar("Uploading")
+    progress.enabled = True  # Fake TTY
 
     # Test initial update
     progress.update(0.0)
@@ -992,7 +1007,8 @@ def test_perform_ota_non_app_type_requires_extended_protocol(
     mock_socket.recv.side_effect = recv_responses
 
     with pytest.raises(
-        espota2.OTAError, match="Device does not support extended OTA protocol"
+        espota2.OTAError,
+        match="Device does not support the extended OTA protocol",
     ):
         espota2.perform_ota(
             mock_socket,
@@ -1026,7 +1042,8 @@ def test_perform_ota_non_app_type_requires_partition_access(
     mock_socket.recv.side_effect = recv_responses
 
     with pytest.raises(
-        espota2.OTAError, match="Device does not support partition access"
+        espota2.OTAError,
+        match=(r"running firmware was built without 'allow_partition_access: true'"),
     ):
         espota2.perform_ota(
             mock_socket,
@@ -1037,6 +1054,60 @@ def test_perform_ota_non_app_type_requires_partition_access(
         )
 
 
+@pytest.mark.usefixtures("mock_time")
+def test_perform_ota_partition_access_error_names_bootloader_flag(
+    mock_socket: Mock, mock_file: io.BytesIO
+) -> None:
+    """Bootloader OTA against a stale device must point at the --bootloader flag."""
+    recv_responses = [
+        bytes([espota2.RESPONSE_OK]),
+        bytes([espota2.OTA_VERSION_2_0]),
+        bytes([espota2.RESPONSE_FEATURE_FLAGS]),
+        bytes([0]),  # No partition access
+    ]
+
+    mock_socket.recv.side_effect = recv_responses
+
+    with pytest.raises(
+        espota2.OTAError,
+        match=r"--bootloader.*recompile and upload.*--bootloader.*retry --bootloader",
+    ):
+        espota2.perform_ota(
+            mock_socket,
+            "testpass",
+            mock_file,
+            "test.bin",
+            espota2.OTA_TYPE_UPDATE_BOOTLOADER,
+        )
+
+
+@pytest.mark.usefixtures("mock_time")
+def test_perform_ota_partition_access_error_names_partition_table_flag(
+    mock_socket: Mock, mock_file: io.BytesIO
+) -> None:
+    """Partition-table OTA against a stale device must point at the --partition-table flag."""
+    recv_responses = [
+        bytes([espota2.RESPONSE_OK]),
+        bytes([espota2.OTA_VERSION_2_0]),
+        bytes([espota2.RESPONSE_FEATURE_FLAGS]),
+        bytes([0]),  # No partition access
+    ]
+
+    mock_socket.recv.side_effect = recv_responses
+
+    with pytest.raises(
+        espota2.OTAError,
+        match=r"--partition-table.*retry --partition-table",
+    ):
+        espota2.perform_ota(
+            mock_socket,
+            "testpass",
+            mock_file,
+            "test.bin",
+            espota2.OTA_TYPE_UPDATE_PARTITION_TABLE,
+        )
+
+
 def test_check_error_detects_errors_when_expect_is_none() -> None:
     """check_error must surface device error bytes even when expect is None.
 
@@ -1044,7 +1115,7 @@ def test_check_error_detects_errors_when_expect_is_none() -> None:
     during feature negotiation and nonce reads) silently passed error bytes
     through, turning clean device errors into confusing later failures.
     """
-    with pytest.raises(espota2.OTAError, match="Error: Authentication invalid"):
+    with pytest.raises(espota2.OTAError, match="Authentication invalid"):
         espota2.check_error([espota2.RESPONSE_ERROR_AUTH_INVALID], None)
 
 

tests/unit_tests/test_helpers.py

@@ -1,9 +1,10 @@
+import io
 import logging
 import os
 from pathlib import Path
 import socket
 import stat
-from unittest.mock import patch
+from unittest.mock import MagicMock, patch
 
 from aioesphomeapi.host_resolver import AddrInfo, IPv4Sockaddr, IPv6Sockaddr
 from hypothesis import given
@@ -12,7 +13,8 @@ import pytest
 
 from esphome import helpers
 from esphome.address_cache import AddressCache
-from esphome.core import EsphomeError
+from esphome.core import CORE, EsphomeError
+from esphome.helpers import ProgressBar
 
 
 @pytest.mark.parametrize(
@@ -1024,3 +1026,39 @@ def test_resolve_ip_address_mixed_cached_uncached() -> None:
         assert "192.168.1.10" in addresses  # Direct IP
         assert "192.168.1.50" in addresses  # From cache
         assert "192.168.1.100" in addresses  # From resolver
+
+
+def test_progressbar_enabled_on_tty(monkeypatch) -> None:
+    """Interactive TTY: progress writes through (pre-existing behaviour)."""
+    stream = MagicMock(spec=io.TextIOWrapper)
+    stream.isatty.return_value = True
+    monkeypatch.setattr(CORE, "dashboard", False)
+
+    bar = ProgressBar("Uploading", stream=stream)
+    assert bar.enabled is True
+
+
+def test_progressbar_disabled_on_pipe_without_dashboard(monkeypatch) -> None:
+    """Piped output without --dashboard: progress suppressed."""
+    stream = MagicMock(spec=io.TextIOWrapper)
+    stream.isatty.return_value = False
+    monkeypatch.setattr(CORE, "dashboard", False)
+
+    bar = ProgressBar("Uploading", stream=stream)
+    assert bar.enabled is False
+
+
+def test_progressbar_enabled_on_pipe_with_dashboard(monkeypatch) -> None:
+    r"""Piped output under --dashboard: progress writes through.
+
+    The dashboard captures stderr through a pipe (so ``isatty()`` is False)
+    and parses ``\rUploading: NN%`` frames to drive its progress UI.
+    Gating purely on ``isatty()`` silently disables every dashboard-side
+    flash-progress indicator.
+    """
+    stream = MagicMock(spec=io.TextIOWrapper)
+    stream.isatty.return_value = False
+    monkeypatch.setattr(CORE, "dashboard", True)
+
+    bar = ProgressBar("Uploading", stream=stream)
+    assert bar.enabled is True

tests/unit_tests/test_main.py

@@ -18,18 +18,19 @@ import pytest
 from pytest import CaptureFixture
 from zeroconf import ServiceStateChange
 
-from esphome import platformio_api
 from esphome.__main__ import (
     Purpose,
     _get_configured_xtal_freq,
     _make_crystal_freq_callback,
     _resolve_network_devices,
+    _validate_bootloader_binary,
     _validate_partition_table_binary,
     choose_upload_log_host,
     command_analyze_memory,
     command_bundle,
     command_clean_all,
     command_rename,
+    command_run,
     command_update_all,
     command_wizard,
     compile_program,
@@ -45,6 +46,7 @@ from esphome.__main__ import (
     has_resolvable_address,
     has_web_server_ota,
     mqtt_get_ip,
+    parse_args,
     run_esphome,
     run_miniterm,
     show_logs,
@@ -87,9 +89,15 @@ from esphome.const import (
     PLATFORM_ESP32,
     PLATFORM_ESP8266,
     PLATFORM_RP2040,
+    Toolchain,
 )
 from esphome.core import CORE, EsphomeError
-from esphome.espota2 import OTA_TYPE_UPDATE_APP, OTA_TYPE_UPDATE_PARTITION_TABLE
+from esphome.espota2 import (
+    OTA_TYPE_UPDATE_APP,
+    OTA_TYPE_UPDATE_BOOTLOADER,
+    OTA_TYPE_UPDATE_PARTITION_TABLE,
+)
+from esphome.platformio import toolchain
 from esphome.util import BootselResult, FlashImage
 from esphome.zeroconf import _await_discovery, discover_mdns_devices
 
@@ -143,6 +151,7 @@ def setup_core(
         config[CONF_WIFI] = {CONF_USE_ADDRESS: address}
 
     CORE.config = config
+    CORE.toolchain = Toolchain.PLATFORMIO
 
     if platform is not None:
         CORE.data[KEY_CORE] = {}
@@ -280,7 +289,7 @@ def mock_run_external_process() -> Generator[Mock]:
 
 @pytest.fixture
 def mock_run_external_command_main() -> Generator[Mock]:
-    """Mock run_external_command in __main__ module (different from platformio_api)."""
+    """Mock run_external_command in __main__ module (different from platformio toolchain)."""
     with patch("esphome.__main__.run_external_command") as mock:
         mock.return_value = 0  # Default to success
         yield mock
@@ -1127,6 +1136,7 @@ class MockArgs:
     output: str | None = None
     ota_platform: str | None = None
     partition_table: bool = False
+    bootloader: bool = False
 
 
 def test_upload_program_serial_esp32(
@@ -1191,7 +1201,7 @@ def test_upload_using_esptool_path_conversion(
     CORE.data[KEY_ESP32] = {KEY_VARIANT: VARIANT_ESP32}
 
     # Create mock IDEData with Path objects
-    mock_idedata = MagicMock(spec=platformio_api.IDEData)
+    mock_idedata = MagicMock(spec=toolchain.IDEData)
     mock_idedata.firmware_bin_path = tmp_path / "firmware.bin"
     mock_idedata.extra_flash_images = [
         FlashImage(path=tmp_path / "bootloader.bin", offset="0x1000"),
@@ -1269,7 +1279,7 @@ def test_upload_using_esptool_skips_missing_extra_flash_images(
 
     missing_path = tmp_path / "variants" / "tasmota" / "tinyuf2.bin"
 
-    mock_idedata = MagicMock(spec=platformio_api.IDEData)
+    mock_idedata = MagicMock(spec=toolchain.IDEData)
     mock_idedata.firmware_bin_path = tmp_path / "firmware.bin"
     mock_idedata.extra_flash_images = [
         FlashImage(path=tmp_path / "bootloader.bin", offset="0x1000"),
@@ -1381,8 +1391,8 @@ def test_upload_using_platformio_creates_signed_bin_for_rp2040(
     mock_idedata.firmware_elf_path = str(firmware_elf)
 
     with (
-        patch("esphome.platformio_api.get_idedata", return_value=mock_idedata),
-        patch("esphome.platformio_api.run_platformio_cli_run", return_value=0),
+        patch("esphome.platformio.toolchain.get_idedata", return_value=mock_idedata),
+        patch("esphome.platformio.toolchain.run_platformio_cli_run", return_value=0),
     ):
         result = upload_using_platformio({}, "/dev/ttyACM0")
 
@@ -1398,7 +1408,7 @@ def test_upload_using_platformio_skips_signed_bin_for_non_rp2040(
     """Test that upload_using_platformio doesn't create signed bin for non-RP2040."""
     setup_core(platform=PLATFORM_ESP32)
 
-    with patch("esphome.platformio_api.run_platformio_cli_run", return_value=0):
+    with patch("esphome.platformio.toolchain.run_platformio_cli_run", return_value=0):
         result = upload_using_platformio({}, "/dev/ttyUSB0")
 
     assert result == 0
@@ -1496,7 +1506,7 @@ def test_upload_using_picotool_success(tmp_path: Path) -> None:
 
     config = {}
     with (
-        patch("esphome.platformio_api.get_idedata", return_value=mock_idedata),
+        patch("esphome.platformio.toolchain.get_idedata", return_value=mock_idedata),
         patch("subprocess.run", return_value=mock_result),
     ):
         exit_code = upload_using_picotool(config)
@@ -1516,7 +1526,7 @@ def test_upload_using_picotool_no_elf(tmp_path: Path) -> None:
     mock_idedata.cc_path = "/fake/path/gcc"
 
     config = {}
-    with patch("esphome.platformio_api.get_idedata", return_value=mock_idedata):
+    with patch("esphome.platformio.toolchain.get_idedata", return_value=mock_idedata):
         exit_code = upload_using_picotool(config)
 
     assert exit_code == 1
@@ -1536,7 +1546,7 @@ def test_upload_using_picotool_not_found(tmp_path: Path) -> None:
     mock_idedata.cc_path = "/fake/path/gcc"
 
     config = {}
-    with patch("esphome.platformio_api.get_idedata", return_value=mock_idedata):
+    with patch("esphome.platformio.toolchain.get_idedata", return_value=mock_idedata):
         exit_code = upload_using_picotool(config)
 
     assert exit_code == 1
@@ -1570,7 +1580,7 @@ def test_upload_using_picotool_permission_error(tmp_path: Path) -> None:
 
     config = {}
     with (
-        patch("esphome.platformio_api.get_idedata", return_value=mock_idedata),
+        patch("esphome.platformio.toolchain.get_idedata", return_value=mock_idedata),
         patch("subprocess.run", return_value=mock_result),
     ):
         exit_code = upload_using_picotool(config)
@@ -1816,6 +1826,27 @@ def test_validate_partition_table_binary_missing_file(tmp_path: Path) -> None:
         _validate_partition_table_binary(tmp_path / "does-not-exist.bin")
 
 
+def test_validate_bootloader_binary_rejects_wrong_magic(tmp_path: Path) -> None:
+    data = bytearray(_make_bootloader_bytes())
+    data[0] = 0x00
+    f = tmp_path / "bootloader.bin"
+    f.write_bytes(bytes(data))
+    with pytest.raises(EsphomeError, match="magic"):
+        _validate_bootloader_binary(f)
+
+
+def test_validate_bootloader_binary_missing_file(tmp_path: Path) -> None:
+    with pytest.raises(EsphomeError, match="Cannot read bootloader file"):
+        _validate_bootloader_binary(tmp_path / "does-not-exist.bin")
+
+
+def test_validate_bootloader_binary_rejects_empty_file(tmp_path: Path) -> None:
+    f = tmp_path / "bootloader.bin"
+    f.write_bytes(b"")
+    with pytest.raises(EsphomeError, match="is empty"):
+        _validate_bootloader_binary(f)
+
+
 def test_upload_program_ota_partition_table_invalid_file(
     mock_run_ota: Mock,
     mock_get_port_type: Mock,
@@ -1869,7 +1900,155 @@ def test_upload_program_ota_partition_table_without_allow_flag(
 
     with pytest.raises(
         EsphomeError,
-        match="requires 'allow_partition_access: true'",
+        match=(
+            r"The option --partition-table requires 'allow_partition_access: true'.*"
+            r"retry --partition-table"
+        ),
+    ):
+        upload_program(config, args, devices)
+    mock_run_ota.assert_not_called()
+
+
+def _make_bootloader_bytes() -> bytes:
+    """Build a minimal bootloader image accepted by _validate_bootloader_binary."""
+    table = bytearray(b"\xff")
+    # Starts with: ESP_IMAGE_HEADER_MAGIC (0xE9)
+    table[0] = 0xE9
+    return bytes(table)
+
+
+def test_upload_program_ota_bootloader_with_file_arg(
+    mock_run_ota: Mock,
+    mock_get_port_type: Mock,
+    tmp_path: Path,
+) -> None:
+    """Test upload_program with OTA and bootloader."""
+    setup_core(platform=PLATFORM_ESP32, tmp_path=tmp_path)
+
+    mock_get_port_type.return_value = "NETWORK"
+    mock_run_ota.return_value = (0, "192.168.1.100")
+
+    bootloader_file = tmp_path / "bootloader.bin"
+    bootloader_file.write_bytes(_make_bootloader_bytes())
+
+    config = {
+        CONF_OTA: [
+            {
+                CONF_PLATFORM: CONF_ESPHOME,
+                CONF_PORT: 3232,
+                "allow_partition_access": True,
+            }
+        ]
+    }
+    args = MockArgs(file=str(bootloader_file), bootloader=True)
+    devices = ["192.168.1.100"]
+
+    exit_code, host = upload_program(config, args, devices)
+
+    assert exit_code == 0
+    assert host == "192.168.1.100"
+    mock_run_ota.assert_called_once_with(
+        ["192.168.1.100"],
+        3232,
+        None,
+        bootloader_file,
+        OTA_TYPE_UPDATE_BOOTLOADER,
+    )
+
+
+def test_upload_program_ota_partition_table_and_bootloader_options(
+    mock_run_ota: Mock,
+    mock_get_port_type: Mock,
+    tmp_path: Path,
+) -> None:
+    """--partition-table and --bootloader can't be used together."""
+    setup_core(platform=PLATFORM_ESP32, tmp_path=tmp_path)
+
+    mock_get_port_type.return_value = "NETWORK"
+
+    config = {
+        CONF_OTA: [
+            {
+                CONF_PLATFORM: CONF_ESPHOME,
+                CONF_PORT: 3232,
+                "allow_partition_access": True,
+            }
+        ]
+    }
+    args = MockArgs(file="partitions.bin", partition_table=True, bootloader=True)
+    devices = ["192.168.1.100"]
+
+    with pytest.raises(
+        EsphomeError,
+        match="--partition-table and --bootloader",
+    ):
+        upload_program(config, args, devices)
+    mock_run_ota.assert_not_called()
+
+
+def test_upload_program_ota_bootloader_without_allow_flag(
+    mock_run_ota: Mock,
+    mock_get_port_type: Mock,
+    tmp_path: Path,
+) -> None:
+    """--bootloader must fail fast when allow_partition_access is not enabled in YAML."""
+    setup_core(platform=PLATFORM_ESP32, tmp_path=tmp_path)
+
+    mock_get_port_type.return_value = "NETWORK"
+
+    config = {
+        CONF_OTA: [
+            {
+                CONF_PLATFORM: CONF_ESPHOME,
+                CONF_PORT: 3232,
+            }
+        ]
+    }
+    args = MockArgs(file="bootloader.bin", bootloader=True)
+    devices = ["192.168.1.100"]
+
+    with pytest.raises(
+        EsphomeError,
+        match=(
+            r"The option --bootloader requires 'allow_partition_access: true'.*"
+            r"retry --bootloader"
+        ),
+    ):
+        upload_program(config, args, devices)
+    mock_run_ota.assert_not_called()
+
+
+def test_upload_program_ota_bootloader_platform_web_server(
+    mock_run_ota: Mock,
+    mock_get_port_type: Mock,
+    tmp_path: Path,
+) -> None:
+    """Test bootloader upload with web_server OTA."""
+    setup_core(platform=PLATFORM_ESP32, tmp_path=tmp_path)
+
+    mock_get_port_type.return_value = "NETWORK"
+
+    bootloader_file = tmp_path / "bootloader.bin"
+    bootloader_file.write_bytes(_make_bootloader_bytes())
+
+    config = {
+        CONF_OTA: [
+            {
+                CONF_PLATFORM: CONF_WEB_SERVER,
+                CONF_WEB_SERVER: {
+                    CONF_PORT: 80,
+                    CONF_AUTH: {CONF_USERNAME: "admin", CONF_PASSWORD: "pw"},
+                },
+                "allow_partition_access": True,
+            }
+        ]
+    }
+    args = MockArgs(file=str(bootloader_file), bootloader=True)
+    devices = ["192.168.1.100"]
+
+    with pytest.raises(
+        EsphomeError,
+        match="the web_server OTA path can only update the firmware image",
     ):
         upload_program(config, args, devices)
     mock_run_ota.assert_not_called()
@@ -3420,6 +3599,467 @@ esp32:
     assert "Rename failed" in captured.out
 
 
+def test_command_rename_install_failure_reverts(
+    tmp_path: Path,
+    capfd: CaptureFixture[str],
+    mock_run_external_process: Mock,
+) -> None:
+    """Test rename when the install (esphome run) step fails."""
+    config_file = tmp_path / "oldname.yaml"
+    config_file.write_text("""
+esphome:
+  name: oldname
+
+esp32:
+  board: nodemcu-32s
+""")
+    setup_core(tmp_path=tmp_path)
+    CORE.config_path = config_file
+    CORE.config = {CONF_ESPHOME: {CONF_NAME: "oldname"}}
+
+    args = MockArgs(name="newname", dashboard=False)
+
+    # First call (config validation) succeeds; second (esphome run) fails.
+    mock_run_external_process.side_effect = [0, 1]
+
+    result = command_rename(args, {})
+
+    assert result == 1
+
+    # New file was unlinked when install failed.
+    new_file = tmp_path / "newname.yaml"
+    assert not new_file.exists()
+
+    # Old file is preserved so the device stays reachable under the
+    # original hostname.
+    assert config_file.exists()
+
+
+def test_command_rename_target_exists_refuses(
+    tmp_path: Path,
+    capfd: CaptureFixture[str],
+    mock_run_external_process: Mock,
+) -> None:
+    """Test rename refuses when the target filename already exists.
+
+    Without this guard, the rename would overwrite the unrelated
+    device's YAML and OTA-install our firmware to the wrong device.
+    """
+    config_file = tmp_path / "oldname.yaml"
+    config_file.write_text("""
+esphome:
+  name: oldname
+
+esp32:
+  board: nodemcu-32s
+""")
+    target_file = tmp_path / "newname.yaml"
+    target_file.write_text("""
+esphome:
+  name: someoneelse
+
+esp32:
+  board: nodemcu-32s
+""")
+    target_original = target_file.read_text()
+    setup_core(tmp_path=tmp_path)
+    CORE.config_path = config_file
+    CORE.config = {CONF_ESPHOME: {CONF_NAME: "oldname"}}
+
+    args = MockArgs(name="newname", dashboard=False)
+
+    result = command_rename(args, {})
+
+    assert result == 1
+    # No subprocess work happened — refusal is up-front.
+    mock_run_external_process.assert_not_called()
+    # Target file untouched: same content, still on disk.
+    assert target_file.exists()
+    assert target_file.read_text() == target_original
+    # Source file untouched.
+    assert config_file.exists()
+
+    captured = capfd.readouterr()
+    assert "already exists" in captured.out
+
+
+def test_command_rename_same_name_refuses(
+    tmp_path: Path,
+    capfd: CaptureFixture[str],
+    mock_run_external_process: Mock,
+) -> None:
+    """Test rename refuses when the new name matches the current name.
+
+    A same-name rename would otherwise re-write the YAML and queue
+    a redundant compile + install — wasted work the user almost
+    certainly didn't intend.
+    """
+    config_file = tmp_path / "samename.yaml"
+    config_file.write_text("""
+esphome:
+  name: samename
+
+esp32:
+  board: nodemcu-32s
+""")
+    setup_core(tmp_path=tmp_path)
+    CORE.config_path = config_file
+    CORE.config = {CONF_ESPHOME: {CONF_NAME: "samename"}}
+
+    args = MockArgs(name="samename", dashboard=False)
+
+    result = command_rename(args, {})
+
+    assert result == 1
+    mock_run_external_process.assert_not_called()
+    # File preserved verbatim — no rewrite happened.
+    assert config_file.exists()
+
+    captured = capfd.readouterr()
+    assert "already" in captured.out.lower()
+
+
+def test_command_rename_does_not_touch_friendly_name_substring(
+    tmp_path: Path,
+    mock_run_external_process: Mock,
+) -> None:
+    r"""Test rename does not match the ``name:`` substring of ``friendly_name:``.
+
+    Without anchoring the regex at line start, the pattern
+    ``\s*name:\s+<old>`` could match the trailing ``name:``
+    substring inside ``friendly_name: <old>``. The rewrite would
+    flip both lines to the new name, leaving the user with a
+    silently corrupted ``friendly_name``.
+    """
+    config_file = tmp_path / "oldname.yaml"
+    config_file.write_text("""
+esphome:
+  name: oldname
+  friendly_name: oldname
+
+esp32:
+  board: nodemcu-32s
+""")
+    setup_core(tmp_path=tmp_path)
+    CORE.config_path = config_file
+    CORE.config = {CONF_ESPHOME: {CONF_NAME: "oldname"}}
+
+    args = MockArgs(name="newname", dashboard=False)
+    mock_run_external_process.return_value = 0
+
+    result = command_rename(args, {})
+
+    assert result == 0
+    new_file = tmp_path / "newname.yaml"
+    content = new_file.read_text()
+    # esphome.name swapped.
+    assert 'name: "newname"' in content
+    # friendly_name kept verbatim.
+    assert "friendly_name: oldname" in content
+
+
+def test_command_rename_does_not_match_old_name_as_value_prefix(
+    tmp_path: Path,
+    mock_run_external_process: Mock,
+) -> None:
+    r"""Test rename does not match ``old_name`` as a prefix of a longer value.
+
+    With ``old_name = kitchen`` the value ``kitchen2`` (a sensor
+    or wifi entry) would otherwise match the unanchored
+    ``["']?kitchen["']?`` pattern at the prefix and get
+    rewritten to the new name. The end-of-value lookahead keeps
+    the match restricted to whole tokens.
+    """
+    config_file = tmp_path / "kitchen.yaml"
+    config_file.write_text("""
+esphome:
+  name: kitchen
+
+esp32:
+  board: nodemcu-32s
+
+wifi:
+  ap:
+    ssid: kitchen2
+""")
+    setup_core(tmp_path=tmp_path)
+    CORE.config_path = config_file
+    CORE.config = {CONF_ESPHOME: {CONF_NAME: "kitchen"}}
+
+    args = MockArgs(name="garage", dashboard=False)
+    mock_run_external_process.return_value = 0
+
+    result = command_rename(args, {})
+
+    assert result == 0
+    new_file = tmp_path / "garage.yaml"
+    content = new_file.read_text()
+    assert 'name: "garage"' in content
+    # The wifi ssid value is unrelated and stays intact.
+    assert "ssid: kitchen2" in content
+
+
+def test_command_rename_same_resolved_name_refuses(
+    tmp_path: Path,
+    capfd: CaptureFixture[str],
+    mock_run_external_process: Mock,
+) -> None:
+    """Test rename refuses when ``new_name`` matches the resolved device name.
+
+    The path-equality check only catches the case where the
+    config filename matches the device name. For a config whose
+    filename and ``esphome.name`` differ (here ``weird-file.yaml``
+    holds ``esphome.name: kitchen``), running
+    ``esphome rename weird-file.yaml kitchen`` would otherwise
+    fall through to the rewrite + install: the YAML's name stays
+    ``kitchen``, the file is renamed to ``kitchen.yaml``, and the
+    device gets a redundant flash. Refuse up-front so the
+    "already the device's name" message matches reality.
+    """
+    config_file = tmp_path / "weird-file.yaml"
+    config_file.write_text("""
+esphome:
+  name: kitchen
+
+esp32:
+  board: nodemcu-32s
+""")
+    setup_core(tmp_path=tmp_path)
+    CORE.config_path = config_file
+    CORE.config = {CONF_ESPHOME: {CONF_NAME: "kitchen"}}
+
+    args = MockArgs(name="kitchen", dashboard=False)
+
+    result = command_rename(args, {})
+
+    assert result == 1
+    mock_run_external_process.assert_not_called()
+    # Source file untouched, no derived target written.
+    assert config_file.exists()
+    assert not (tmp_path / "kitchen.yaml").exists()
+
+    captured = capfd.readouterr()
+    assert "already" in captured.out.lower()
+
+
+def test_command_rename_target_path_equals_source_refuses(
+    tmp_path: Path,
+    capfd: CaptureFixture[str],
+    mock_run_external_process: Mock,
+) -> None:
+    """Test rename refuses when the new path resolves to the source file.
+
+    Reachable only when the YAML's filename and ``esphome.name``
+    disagree — here ``kitchen.yaml`` holds ``esphome.name: garage``
+    and the user runs ``esphome rename kitchen.yaml kitchen``. The
+    name-equality check above passes (``garage != kitchen``), but
+    ``<config_dir>/kitchen.yaml`` resolves to the source file
+    itself, so the rewrite would clobber the source mid-rename.
+    Refuse rather than silently overwriting.
+    """
+    config_file = tmp_path / "kitchen.yaml"
+    config_file.write_text("""
+esphome:
+  name: garage
+
+esp32:
+  board: nodemcu-32s
+""")
+    setup_core(tmp_path=tmp_path)
+    CORE.config_path = config_file
+    CORE.config = {CONF_ESPHOME: {CONF_NAME: "garage"}}
+
+    args = MockArgs(name="kitchen", dashboard=False)
+
+    result = command_rename(args, {})
+
+    assert result == 1
+    mock_run_external_process.assert_not_called()
+    # Source file still present and unmodified.
+    assert config_file.exists()
+    assert "name: garage" in config_file.read_text()
+
+    captured = capfd.readouterr()
+    assert "already" in captured.out.lower()
+
+
+def test_command_rename_does_not_touch_lookalike_name_in_other_blocks(
+    tmp_path: Path,
+    mock_run_external_process: Mock,
+) -> None:
+    """Test rename only swaps the esphome.name line.
+
+    A device whose name happens to match a sensor's / output's
+    ``name:`` value must not have those other names rewritten —
+    they're independent. Without an anchor for the esphome block
+    a naive regex would clobber every line whose value matches.
+    """
+    config_file = tmp_path / "kitchen.yaml"
+    config_file.write_text("""
+esphome:
+  name: kitchen
+
+esp32:
+  board: nodemcu-32s
+
+sensor:
+  - platform: template
+    name: kitchen
+    lambda: 'return 0;'
+""")
+    setup_core(tmp_path=tmp_path)
+    CORE.config_path = config_file
+    CORE.config = {CONF_ESPHOME: {CONF_NAME: "kitchen"}}
+
+    args = MockArgs(name="garage", dashboard=False)
+    mock_run_external_process.return_value = 0
+
+    result = command_rename(args, {})
+
+    assert result == 0
+
+    new_file = tmp_path / "garage.yaml"
+    content = new_file.read_text()
+    # esphome.name renamed.
+    assert 'name: "garage"' in content
+    # Sensor's name is the user's entity name — must not be touched.
+    assert "    name: kitchen\n" in content
+
+
+def test_command_rename_preserves_trailing_comment(
+    tmp_path: Path,
+    mock_run_external_process: Mock,
+) -> None:
+    """Test rename preserves a trailing ``# comment`` on the name line."""
+    config_file = tmp_path / "kitchen.yaml"
+    config_file.write_text("""
+esphome:
+  name: kitchen  # primary device
+
+esp32:
+  board: nodemcu-32s
+""")
+    setup_core(tmp_path=tmp_path)
+    CORE.config_path = config_file
+    CORE.config = {CONF_ESPHOME: {CONF_NAME: "kitchen"}}
+
+    args = MockArgs(name="garage", dashboard=False)
+    mock_run_external_process.return_value = 0
+
+    result = command_rename(args, {})
+
+    assert result == 0
+
+    new_file = tmp_path / "garage.yaml"
+    content = new_file.read_text()
+    assert "# primary device" in content
+
+
+def test_command_rename_handles_double_quoted_value(
+    tmp_path: Path,
+    mock_run_external_process: Mock,
+) -> None:
+    """Test rename matches when the existing value is double-quoted."""
+    config_file = tmp_path / "kitchen.yaml"
+    config_file.write_text("""
+esphome:
+  name: "kitchen"
+
+esp32:
+  board: nodemcu-32s
+""")
+    setup_core(tmp_path=tmp_path)
+    CORE.config_path = config_file
+    CORE.config = {CONF_ESPHOME: {CONF_NAME: "kitchen"}}
+
+    args = MockArgs(name="garage", dashboard=False)
+    mock_run_external_process.return_value = 0
+
+    result = command_rename(args, {})
+
+    assert result == 0
+    new_file = tmp_path / "garage.yaml"
+    assert 'name: "garage"' in new_file.read_text()
+
+
+def test_command_rename_handles_single_quoted_value(
+    tmp_path: Path,
+    mock_run_external_process: Mock,
+) -> None:
+    """Test rename matches when the existing value is single-quoted."""
+    config_file = tmp_path / "kitchen.yaml"
+    config_file.write_text("""
+esphome:
+  name: 'kitchen'
+
+esp32:
+  board: nodemcu-32s
+""")
+    setup_core(tmp_path=tmp_path)
+    CORE.config_path = config_file
+    CORE.config = {CONF_ESPHOME: {CONF_NAME: "kitchen"}}
+
+    args = MockArgs(name="garage", dashboard=False)
+    mock_run_external_process.return_value = 0
+
+    result = command_rename(args, {})
+
+    assert result == 0
+    new_file = tmp_path / "garage.yaml"
+    assert 'name: "garage"' in new_file.read_text()
+
+
+def test_command_rename_too_many_substitution_matches_refuses(
+    tmp_path: Path,
+    capfd: CaptureFixture[str],
+    mock_run_external_process: Mock,
+) -> None:
+    """Test rename refuses when ``${var}`` resolves to multiple matches.
+
+    When ``esphome.name: ${device_name}`` and the substitution
+    definition ``device_name: foo`` appears more than once in the
+    YAML (e.g. inside multiple included blocks), the regex rewrite
+    can't tell which one to flip. Rather than silently picking one
+    or rewriting both, the command refuses.
+    """
+    config_file = tmp_path / "oldname.yaml"
+    config_file.write_text("""
+substitutions:
+  device_name: oldname
+
+esphome:
+  name: ${device_name}
+
+# A copy-pasted block that re-declares the substitution at the
+# same indent level - happens when users splice in a packaged
+# fragment without renaming the variable.
+example:
+  device_name: oldname
+
+esp32:
+  board: nodemcu-32s
+""")
+    setup_core(tmp_path=tmp_path)
+    CORE.config_path = config_file
+    CORE.config = {
+        CONF_ESPHOME: {CONF_NAME: "oldname"},
+        CONF_SUBSTITUTIONS: {"device_name": "oldname"},
+    }
+
+    args = MockArgs(name="newname", dashboard=False)
+
+    result = command_rename(args, {})
+
+    assert result == 1
+    mock_run_external_process.assert_not_called()
+    # File untouched.
+    assert config_file.exists()
+    assert "device_name: oldname" in config_file.read_text()
+
+    captured = capfd.readouterr()
+    assert "Too many matches" in captured.out
+
+
 def test_command_update_all_path_string_conversion(
     tmp_path: Path,
     mock_run_external_process: Mock,
@@ -4058,7 +4698,7 @@ def test_command_analyze_memory_success(
     firmware_elf.write_text("mock elf file")
 
     # Mock idedata
-    mock_idedata_obj = MagicMock(spec=platformio_api.IDEData)
+    mock_idedata_obj = MagicMock(spec=toolchain.IDEData)
     mock_idedata_obj.firmware_elf_path = str(firmware_elf)
     mock_idedata_obj.objdump_path = "/path/to/objdump"
     mock_idedata_obj.readelf_path = "/path/to/readelf"
@@ -4130,7 +4770,7 @@ def test_command_analyze_memory_with_external_components(
     firmware_elf.write_text("mock elf file")
 
     # Mock idedata
-    mock_idedata_obj = MagicMock(spec=platformio_api.IDEData)
+    mock_idedata_obj = MagicMock(spec=toolchain.IDEData)
     mock_idedata_obj.firmware_elf_path = str(firmware_elf)
     mock_idedata_obj.objdump_path = "/path/to/objdump"
     mock_idedata_obj.readelf_path = "/path/to/readelf"
@@ -4221,16 +4861,18 @@ def test_command_analyze_memory_no_idedata(
 
 @pytest.fixture
 def mock_compile_build_info_run_compile() -> Generator[Mock]:
-    """Mock platformio_api.run_compile for build_info tests."""
-    with patch("esphome.platformio_api.run_compile", return_value=0) as mock:
+    """Mock toolchain.run_compile for build_info tests."""
+    with patch("esphome.platformio.toolchain.run_compile", return_value=0) as mock:
         yield mock
 
 
 @pytest.fixture
 def mock_compile_build_info_get_idedata() -> Generator[Mock]:
-    """Mock platformio_api.get_idedata for build_info tests."""
+    """Mock toolchain.get_idedata for build_info tests."""
     mock_idedata = MagicMock()
-    with patch("esphome.platformio_api.get_idedata", return_value=mock_idedata) as mock:
+    with patch(
+        "esphome.platformio.toolchain.get_idedata", return_value=mock_idedata
+    ) as mock:
         yield mock
 
 
@@ -5140,7 +5782,7 @@ def test_upload_using_esptool_passes_crystal_callback(
     sdkconfig = build_dir / "sdkconfig.test"
     sdkconfig.write_text("CONFIG_XTAL_FREQ=40\n")
 
-    mock_idedata = MagicMock(spec=platformio_api.IDEData)
+    mock_idedata = MagicMock(spec=toolchain.IDEData)
     mock_idedata.firmware_bin_path = tmp_path / "firmware.bin"
     mock_idedata.extra_flash_images = []
     mock_get_idedata.return_value = mock_idedata
@@ -5170,7 +5812,7 @@ def test_upload_using_esptool_subprocess_passes_crystal_callback(
     sdkconfig = build_dir / "sdkconfig.test"
     sdkconfig.write_text("CONFIG_XTAL_FREQ=40\n")
 
-    mock_idedata = MagicMock(spec=platformio_api.IDEData)
+    mock_idedata = MagicMock(spec=toolchain.IDEData)
     mock_idedata.firmware_bin_path = tmp_path / "firmware.bin"
     mock_idedata.extra_flash_images = []
     mock_get_idedata.return_value = mock_idedata
@@ -5183,3 +5825,96 @@ def test_upload_using_esptool_subprocess_passes_crystal_callback(
     call_kwargs = mock_run_external_process.call_args[1]
     assert "line_callbacks" in call_kwargs
     assert len(call_kwargs["line_callbacks"]) == 1
+
+
+def test_parse_args_run_no_states() -> None:
+    """Test that --no-states is parsed for the run command."""
+    args = parse_args(["esphome", "run", "--no-states", "device.yaml"])
+    assert args.no_states is True
+
+
+def test_parse_args_run_no_states_default() -> None:
+    """Test that no_states defaults to False for the run command."""
+    args = parse_args(["esphome", "run", "device.yaml"])
+    assert args.no_states is False
+
+
+def test_parse_args_logs_no_states() -> None:
+    """Test that --no-states is parsed for the logs command."""
+    args = parse_args(["esphome", "logs", "--no-states", "device.yaml"])
+    assert args.no_states is True
+
+
+@patch("esphome.components.api.client.run_logs")
+def test_command_run_passes_no_states_to_show_logs(
+    mock_run_logs: Mock,
+) -> None:
+    """Test that command_run propagates --no-states through to run_logs."""
+    setup_core(
+        config={
+            "logger": {},
+            CONF_API: {},
+            CONF_MDNS: {CONF_DISABLED: False},
+        },
+        platform=PLATFORM_ESP32,
+    )
+    mock_run_logs.return_value = 0
+
+    args = MockArgs()
+    args.no_states = True
+    args.no_logs = False
+    args.device = None
+
+    with (
+        patch("esphome.__main__.write_cpp", return_value=0),
+        patch("esphome.__main__.compile_program", return_value=0),
+        patch(
+            "esphome.__main__.choose_upload_log_host",
+            return_value=["192.168.1.100"],
+        ),
+        patch("esphome.__main__.upload_program", return_value=(0, "192.168.1.100")),
+        patch("esphome.__main__.get_serial_ports", return_value=[]),
+    ):
+        result = command_run(args, CORE.config)
+
+    assert result == 0
+    mock_run_logs.assert_called_once_with(
+        CORE.config, ["192.168.1.100"], subscribe_states=False
+    )
+
+
+@patch("esphome.components.api.client.run_logs")
+def test_command_run_defaults_subscribe_states_true(
+    mock_run_logs: Mock,
+) -> None:
+    """Test that command_run subscribes states by default (no --no-states)."""
+    setup_core(
+        config={
+            "logger": {},
+            CONF_API: {},
+            CONF_MDNS: {CONF_DISABLED: False},
+        },
+        platform=PLATFORM_ESP32,
+    )
+    mock_run_logs.return_value = 0
+
+    args = MockArgs()
+    args.no_logs = False
+    args.device = None
+
+    with (
+        patch("esphome.__main__.write_cpp", return_value=0),
+        patch("esphome.__main__.compile_program", return_value=0),
+        patch(
+            "esphome.__main__.choose_upload_log_host",
+            return_value=["192.168.1.100"],
+        ),
+        patch("esphome.__main__.upload_program", return_value=(0, "192.168.1.100")),
+        patch("esphome.__main__.get_serial_ports", return_value=[]),
+    ):
+        result = command_run(args, CORE.config)
+
+    assert result == 0
+    mock_run_logs.assert_called_once_with(
+        CORE.config, ["192.168.1.100"], subscribe_states=True
+    )

tests/unit_tests/test_platformio_toolchain.py

@@ -1,4 +1,4 @@
-"""Tests for platformio_api.py path functions."""
+"""Tests for esphome.platformio.toolchain path functions."""
 
 # pylint: disable=protected-access
 
@@ -11,8 +11,8 @@ from unittest.mock import MagicMock, Mock, call, patch
 
 import pytest
 
-from esphome import platformio_api, platformio_runner
 from esphome.core import CORE, EsphomeError
+from esphome.platformio import runner, toolchain
 from esphome.util import FlashImage
 
 
@@ -21,7 +21,7 @@ def test_idedata_firmware_elf_path(setup_core: Path) -> None:
     CORE.build_path = setup_core / "build" / "test"
     CORE.name = "test"
     raw_data = {"prog_path": "/path/to/firmware.elf"}
-    idedata = platformio_api.IDEData(raw_data)
+    idedata = toolchain.IDEData(raw_data)
 
     assert idedata.firmware_elf_path == Path("/path/to/firmware.elf")
 
@@ -32,7 +32,7 @@ def test_idedata_firmware_bin_path(setup_core: Path) -> None:
     CORE.name = "test"
     prog_path = str(Path("/path/to/firmware.elf"))
     raw_data = {"prog_path": prog_path}
-    idedata = platformio_api.IDEData(raw_data)
+    idedata = toolchain.IDEData(raw_data)
 
     result = idedata.firmware_bin_path
     assert isinstance(result, Path)
@@ -47,7 +47,7 @@ def test_idedata_firmware_bin_path_preserves_directory(setup_core: Path) -> None
     CORE.name = "test"
     prog_path = str(Path("/complex/path/to/build/firmware.elf"))
     raw_data = {"prog_path": prog_path}
-    idedata = platformio_api.IDEData(raw_data)
+    idedata = toolchain.IDEData(raw_data)
 
     result = idedata.firmware_bin_path
     expected = Path("/complex/path/to/build/firmware.bin")
@@ -67,7 +67,7 @@ def test_idedata_extra_flash_images(setup_core: Path) -> None:
             ]
         },
     }
-    idedata = platformio_api.IDEData(raw_data)
+    idedata = toolchain.IDEData(raw_data)
 
     images = idedata.extra_flash_images
     assert len(images) == 2
@@ -83,7 +83,7 @@ def test_idedata_extra_flash_images_empty(setup_core: Path) -> None:
     CORE.build_path = setup_core / "build" / "test"
     CORE.name = "test"
     raw_data = {"prog_path": "/path/to/firmware.elf", "extra": {"flash_images": []}}
-    idedata = platformio_api.IDEData(raw_data)
+    idedata = toolchain.IDEData(raw_data)
 
     images = idedata.extra_flash_images
     assert images == []
@@ -97,7 +97,7 @@ def test_idedata_cc_path(setup_core: Path) -> None:
         "prog_path": "/path/to/firmware.elf",
         "cc_path": "/Users/test/.platformio/packages/toolchain-xtensa32/bin/xtensa-esp32-elf-gcc",
     }
-    idedata = platformio_api.IDEData(raw_data)
+    idedata = toolchain.IDEData(raw_data)
 
     assert (
         idedata.cc_path
@@ -132,7 +132,7 @@ def test_load_idedata_returns_dict(
     mock_run_platformio_cli_run.return_value = '{"prog_path": "/test/firmware.elf"}'
 
     config = {"name": "test"}
-    result = platformio_api._load_idedata(config)
+    result = toolchain._load_idedata(config)
 
     assert result is not None
     assert isinstance(result, dict)
@@ -161,7 +161,7 @@ def test_load_idedata_uses_cache_when_valid(
     os.utime(idedata_path, (platformio_ini_mtime + 1, platformio_ini_mtime + 1))
 
     config = {"name": "test"}
-    result = platformio_api._load_idedata(config)
+    result = toolchain._load_idedata(config)
 
     # Should not call _run_idedata since cache is valid
     mock_run_platformio_cli_run.assert_not_called()
@@ -194,7 +194,7 @@ def test_load_idedata_regenerates_when_platformio_ini_newer(
     mock_run_platformio_cli_run.return_value = json.dumps(new_data)
 
     config = {"name": "test"}
-    result = platformio_api._load_idedata(config)
+    result = toolchain._load_idedata(config)
 
     # Should call _run_idedata since platformio.ini is newer
     mock_run_platformio_cli_run.assert_called_once()
@@ -228,7 +228,7 @@ def test_load_idedata_regenerates_on_corrupted_cache(
     mock_run_platformio_cli_run.return_value = json.dumps(new_data)
 
     config = {"name": "test"}
-    result = platformio_api._load_idedata(config)
+    result = toolchain._load_idedata(config)
 
     # Should call _run_idedata since cache is corrupted
     mock_run_platformio_cli_run.assert_called_once()
@@ -253,7 +253,7 @@ def test_run_idedata_parses_json_from_output(
         f"Some preamble\n{json.dumps(expected_data)}\nSome postamble"
     )
 
-    result = platformio_api._run_idedata(config)
+    result = toolchain._run_idedata(config)
 
     assert result == expected_data
 
@@ -267,7 +267,7 @@ def test_run_idedata_raises_on_no_json(
     mock_run_platformio_cli_run.return_value = "No JSON in this output"
 
     with pytest.raises(EsphomeError):
-        platformio_api._run_idedata(config)
+        toolchain._run_idedata(config)
 
 
 def test_run_idedata_raises_on_invalid_json(
@@ -279,7 +279,7 @@ def test_run_idedata_raises_on_invalid_json(
 
     # The ValueError from json.loads is re-raised
     with pytest.raises(ValueError):
-        platformio_api._run_idedata(config)
+        toolchain._run_idedata(config)
 
 
 def test_run_platformio_cli_sets_environment_variables(
@@ -290,7 +290,7 @@ def test_run_platformio_cli_sets_environment_variables(
 
     with patch.dict(os.environ, {}, clear=False):
         mock_run_external_process.return_value = 0
-        platformio_api.run_platformio_cli("test", "arg")
+        toolchain.run_platformio_cli("test", "arg")
 
         # Check environment variables were set
         assert os.environ["PLATFORMIO_FORCE_COLOR"] == "true"
@@ -303,11 +303,11 @@ def test_run_platformio_cli_sets_environment_variables(
         assert "PYTHONWARNINGS" in os.environ
 
         # Check command was called correctly — runs PlatformIO as a subprocess
-        # via the esphome.platformio_runner entry point.
+        # via the esphome.platformio.runner entry point.
         mock_run_external_process.assert_called_once()
         args = mock_run_external_process.call_args[0]
         assert "-m" in args
-        assert "esphome.platformio_runner" in args
+        assert "esphome.platformio.runner" in args
         assert "test" in args
         assert "arg" in args
 
@@ -342,8 +342,8 @@ def test_strip_win_long_path_prefix(
     platform: str, input_path: str, expected: str
 ) -> None:
     r"""``\\?\`` and ``\\?\UNC\`` prefixes are stripped only on win32."""
-    with patch("esphome.platformio_api.sys.platform", platform):
-        assert platformio_api._strip_win_long_path_prefix(input_path) == expected
+    with patch("esphome.platformio.toolchain.sys.platform", platform):
+        assert toolchain._strip_win_long_path_prefix(input_path) == expected
 
 
 def test_run_platformio_cli_strips_win_long_path_prefix(
@@ -366,15 +366,15 @@ def test_run_platformio_cli_strips_win_long_path_prefix(
 
     with (
         patch.dict(os.environ, {}, clear=False),
-        patch("esphome.platformio_api.sys.platform", "win32"),
-        patch("esphome.platformio_api.sys.executable", prefixed_exe),
+        patch("esphome.platformio.toolchain.sys.platform", "win32"),
+        patch("esphome.platformio.toolchain.sys.executable", prefixed_exe),
     ):
         # Pop any pre-existing PYTHONEXEPATH so the assertion below reflects
         # what run_platformio_cli set, not whatever the test runner's
         # environment happened to contain.
         os.environ.pop("PYTHONEXEPATH", None)
         mock_run_external_process.return_value = 0
-        platformio_api.run_platformio_cli("test", "arg")
+        toolchain.run_platformio_cli("test", "arg")
 
         # The subprocess is invoked with the stripped executable path.
         mock_run_external_process.assert_called_once()
@@ -398,12 +398,12 @@ def test_run_platformio_cli_does_not_set_pythonexepath_without_strip(
 
     with (
         patch.dict(os.environ, {}, clear=False),
-        patch("esphome.platformio_api.sys.platform", "linux"),
-        patch("esphome.platformio_api.sys.executable", plain_exe),
+        patch("esphome.platformio.toolchain.sys.platform", "linux"),
+        patch("esphome.platformio.toolchain.sys.executable", plain_exe),
     ):
         os.environ.pop("PYTHONEXEPATH", None)
         mock_run_external_process.return_value = 0
-        platformio_api.run_platformio_cli("test", "arg")
+        toolchain.run_platformio_cli("test", "arg")
 
         mock_run_external_process.assert_called_once()
         args = mock_run_external_process.call_args[0]
@@ -419,7 +419,7 @@ def test_run_platformio_cli_run_builds_command(
     mock_run_platformio_cli.return_value = 0
 
     config = {"name": "test"}
-    platformio_api.run_platformio_cli_run(config, True, "extra", "args")
+    toolchain.run_platformio_cli_run(config, True, "extra", "args")
 
     mock_run_platformio_cli.assert_called_once_with(
         "run", "-d", CORE.build_path, "-v", "extra", "args"
@@ -434,7 +434,7 @@ 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
 
-    platformio_api.run_compile(config, verbose=True)
+    toolchain.run_compile(config, verbose=True)
 
     mock_run_platformio_cli_run.assert_called_once_with(config, True, "-j4")
 
@@ -461,22 +461,22 @@ def test_get_idedata_caches_result(
     config = {"name": "test"}
 
     # First call should load and cache
-    result1 = platformio_api.get_idedata(config)
+    result1 = toolchain.get_idedata(config)
     mock_run_platformio_cli_run.assert_called_once()
 
     # Second call should use cache from CORE.data
-    result2 = platformio_api.get_idedata(config)
+    result2 = toolchain.get_idedata(config)
     mock_run_platformio_cli_run.assert_called_once()  # Still only called once
 
     assert result1 is result2
-    assert isinstance(result1, platformio_api.IDEData)
+    assert isinstance(result1, toolchain.IDEData)
     assert result1.firmware_elf_path == Path("/test/firmware.elf")
 
 
 def test_idedata_addr2line_path_windows(setup_core: Path) -> None:
     """Test IDEData.addr2line_path on Windows."""
     raw_data = {"prog_path": "/path/to/firmware.elf", "cc_path": "C:\\tools\\gcc.exe"}
-    idedata = platformio_api.IDEData(raw_data)
+    idedata = toolchain.IDEData(raw_data)
 
     result = idedata.addr2line_path
     assert result == "C:\\tools\\addr2line.exe"
@@ -485,7 +485,7 @@ def test_idedata_addr2line_path_windows(setup_core: Path) -> None:
 def test_idedata_addr2line_path_unix(setup_core: Path) -> None:
     """Test IDEData.addr2line_path on Unix."""
     raw_data = {"prog_path": "/path/to/firmware.elf", "cc_path": "/usr/bin/gcc"}
-    idedata = platformio_api.IDEData(raw_data)
+    idedata = toolchain.IDEData(raw_data)
 
     result = idedata.addr2line_path
     assert result == "/usr/bin/addr2line"
@@ -494,7 +494,7 @@ def test_idedata_addr2line_path_unix(setup_core: Path) -> None:
 def test_idedata_objdump_path_windows(setup_core: Path) -> None:
     """Test IDEData.objdump_path on Windows."""
     raw_data = {"prog_path": "/path/to/firmware.elf", "cc_path": "C:\\tools\\gcc.exe"}
-    idedata = platformio_api.IDEData(raw_data)
+    idedata = toolchain.IDEData(raw_data)
 
     result = idedata.objdump_path
     assert result == "C:\\tools\\objdump.exe"
@@ -503,7 +503,7 @@ def test_idedata_objdump_path_windows(setup_core: Path) -> None:
 def test_idedata_objdump_path_unix(setup_core: Path) -> None:
     """Test IDEData.objdump_path on Unix."""
     raw_data = {"prog_path": "/path/to/firmware.elf", "cc_path": "/usr/bin/gcc"}
-    idedata = platformio_api.IDEData(raw_data)
+    idedata = toolchain.IDEData(raw_data)
 
     result = idedata.objdump_path
     assert result == "/usr/bin/objdump"
@@ -512,7 +512,7 @@ def test_idedata_objdump_path_unix(setup_core: Path) -> None:
 def test_idedata_readelf_path_windows(setup_core: Path) -> None:
     """Test IDEData.readelf_path on Windows."""
     raw_data = {"prog_path": "/path/to/firmware.elf", "cc_path": "C:\\tools\\gcc.exe"}
-    idedata = platformio_api.IDEData(raw_data)
+    idedata = toolchain.IDEData(raw_data)
 
     result = idedata.readelf_path
     assert result == "C:\\tools\\readelf.exe"
@@ -521,7 +521,7 @@ def test_idedata_readelf_path_windows(setup_core: Path) -> None:
 def test_idedata_readelf_path_unix(setup_core: Path) -> None:
     """Test IDEData.readelf_path on Unix."""
     raw_data = {"prog_path": "/path/to/firmware.elf", "cc_path": "/usr/bin/gcc"}
-    idedata = platformio_api.IDEData(raw_data)
+    idedata = toolchain.IDEData(raw_data)
 
     result = idedata.readelf_path
     assert result == "/usr/bin/readelf"
@@ -547,7 +547,7 @@ def test_patch_structhash(setup_core: Path) -> None:
         },
     ):
         # Call patch_structhash
-        platformio_runner.patch_structhash()
+        runner.patch_structhash()
 
         # Verify both modules had clean_build_dir patched
         # Check that clean_build_dir was set on both modules
@@ -599,7 +599,7 @@ def test_patched_clean_build_dir_removes_outdated(setup_core: Path) -> None:
         },
     ):
         # Call patch_structhash to install the patched function
-        platformio_runner.patch_structhash()
+        runner.patch_structhash()
 
         # Call the patched function
         mock_helpers.clean_build_dir(str(build_dir), [])
@@ -649,7 +649,7 @@ def test_patched_clean_build_dir_keeps_updated(setup_core: Path) -> None:
         },
     ):
         # Call patch_structhash to install the patched function
-        platformio_runner.patch_structhash()
+        runner.patch_structhash()
 
         # Call the patched function
         mock_helpers.clean_build_dir(str(build_dir), [])
@@ -697,7 +697,7 @@ def test_patched_clean_build_dir_creates_missing(setup_core: Path) -> None:
         },
     ):
         # Call patch_structhash to install the patched function
-        platformio_runner.patch_structhash()
+        runner.patch_structhash()
 
         # Call the patched function
         mock_helpers.clean_build_dir(str(build_dir), [])
@@ -727,7 +727,7 @@ def test_patch_file_downloader_succeeds_first_try() -> None:
             ),
         },
     ):
-        platformio_runner.patch_file_downloader()
+        runner.patch_file_downloader()
 
         from platformio.package.download import FileDownloader
 
@@ -766,7 +766,7 @@ def test_patch_file_downloader_retries_on_failure() -> None:
         ),
         patch("time.sleep") as mock_sleep,
     ):
-        platformio_runner.patch_file_downloader()
+        runner.patch_file_downloader()
 
         from platformio.package.download import FileDownloader
 
@@ -807,7 +807,7 @@ def test_patch_file_downloader_raises_after_max_retries() -> None:
         ),
         patch("time.sleep") as mock_sleep,
     ):
-        platformio_runner.patch_file_downloader()
+        runner.patch_file_downloader()
 
         from platformio.package.download import FileDownloader
 
@@ -855,7 +855,7 @@ def test_patch_file_downloader_closes_session_and_response_between_retries() ->
         ),
         patch("time.sleep"),
     ):
-        platformio_runner.patch_file_downloader()
+        runner.patch_file_downloader()
 
         from platformio.package.download import FileDownloader
 
@@ -890,9 +890,9 @@ def test_patch_file_downloader_idempotent() -> None:
         },
     ):
         # Patch multiple times
-        platformio_runner.patch_file_downloader()
-        platformio_runner.patch_file_downloader()
-        platformio_runner.patch_file_downloader()
+        runner.patch_file_downloader()
+        runner.patch_file_downloader()
+        runner.patch_file_downloader()
 
         from platformio.package.download import FileDownloader
 
@@ -910,9 +910,7 @@ def _filter_through_redirect(line: str) -> str:
     from esphome.util import RedirectText
 
     captured = io.StringIO()
-    redirect = RedirectText(
-        captured, filter_lines=platformio_runner.FILTER_PLATFORMIO_LINES
-    )
+    redirect = RedirectText(captured, filter_lines=runner.FILTER_PLATFORMIO_LINES)
     redirect.write(line + "\n")
     return captured.getvalue()
 

tests/unit_tests/test_upload_targets.py

@@ -0,0 +1,81 @@
+"""Tests for the stable upload-targets classification helpers."""
+
+import pytest
+
+from esphome.upload_targets import PortType, get_port_type
+
+
+@pytest.mark.parametrize(
+    "port",
+    [
+        "/dev/ttyUSB0",
+        "/dev/ttyACM0",
+        "/dev/cu.usbserial-1410",
+        "/dev/tty.usbmodem1101",
+        "COM1",
+        "COM23",
+    ],
+)
+def test_get_port_type_serial(port: str) -> None:
+    """Local serial devices classify as SERIAL."""
+    assert get_port_type(port) is PortType.SERIAL
+
+
+def test_get_port_type_bootsel() -> None:
+    """``BOOTSEL`` magic string classifies as BOOTSEL."""
+    assert get_port_type("BOOTSEL") is PortType.BOOTSEL
+
+
+def test_get_port_type_mqtt() -> None:
+    """``MQTT`` magic string classifies as MQTT."""
+    assert get_port_type("MQTT") is PortType.MQTT
+
+
+def test_get_port_type_mqttip() -> None:
+    """``MQTTIP`` magic string classifies as MQTTIP."""
+    assert get_port_type("MQTTIP") is PortType.MQTTIP
+
+
+@pytest.mark.parametrize(
+    "port",
+    [
+        "192.168.1.10",
+        "fe80::1",
+        "device.local",
+        "my-esp.example.com",
+    ],
+)
+def test_get_port_type_network(port: str) -> None:
+    """IP addresses, mDNS, and hostnames classify as NETWORK."""
+    assert get_port_type(port) is PortType.NETWORK
+
+
+def test_port_type_values_are_stable() -> None:
+    """Member values are part of the stable surface.
+
+    External tooling (device-builder, etc.) may compare against the
+    string values directly. Renaming or changing these breaks
+    downstream consumers — guard against accidental edits.
+    """
+    assert PortType.SERIAL.value == "SERIAL"
+    assert PortType.NETWORK.value == "NETWORK"
+    assert PortType.MQTT.value == "MQTT"
+    assert PortType.MQTTIP.value == "MQTTIP"
+    assert PortType.BOOTSEL.value == "BOOTSEL"
+
+
+def test_main_re_exports_for_backwards_compat() -> None:
+    """``esphome.__main__`` re-exports the stable surface.
+
+    The CLI entry point pre-dated the stable module and existing
+    internal callers (and any third-party code that snuck in via
+    ``__main__``) still import from there. The re-export must
+    resolve to the same objects.
+    """
+    from esphome.__main__ import (
+        PortType as MainPortType,
+        get_port_type as main_get_port_type,
+    )
+
+    assert MainPortType is PortType
+    assert main_get_port_type is get_port_type