Merge branch 'esp8266-native-build-spec' into esp8266-native-ninja-emission

tests/component_tests/esp32_ble_tracker/config/scan_window_explicit.yaml

@@ -0,0 +1,19 @@
+esphome:
+  name: scan-window-explicit
+
+esp32:
+  board: esp32dev
+  framework:
+    type: esp-idf
+
+wifi:
+  ssid: MySSID
+
+esp32_ble_tracker:
+  scan_parameters:
+    window: 30ms
+
+bluetooth_proxy:
+  active: true
+
+api:

tests/component_tests/esp32_ble_tracker/config/scan_window_raised.yaml

@@ -0,0 +1,17 @@
+esphome:
+  name: scan-window-raised
+
+esp32:
+  board: esp32dev
+  framework:
+    type: esp-idf
+
+wifi:
+  ssid: MySSID
+
+esp32_ble_tracker:
+
+bluetooth_proxy:
+  active: true
+
+api:

tests/component_tests/esp32_ble_tracker/config/scan_window_scan_only.yaml

@@ -0,0 +1,12 @@
+esphome:
+  name: scan-window-scan-only
+
+esp32:
+  board: esp32dev
+  framework:
+    type: esp-idf
+
+wifi:
+  ssid: MySSID
+
+esp32_ble_tracker:

tests/component_tests/esp32_ble_tracker/config/scan_window_user_set_scan_only.yaml

@@ -0,0 +1,14 @@
+esphome:
+  name: scan-window-user-scan-only
+
+esp32:
+  board: esp32dev
+  framework:
+    type: esp-idf
+
+wifi:
+  ssid: MySSID
+
+esp32_ble_tracker:
+  scan_parameters:
+    connection_scan_window: 20ms

tests/component_tests/esp32_ble_tracker/test_scan_window_default.py

@@ -12,11 +12,12 @@ arbiter a full-duty scan would starve wifi, so the 30 ms default is kept.
 from __future__ import annotations
 
 from collections.abc import Callable
+from pathlib import Path
 
 import pytest
 
 from esphome import config_validation as cv
-from esphome.components.ble_device_base import to_ble_units
+from esphome.components.ble_device_base import CONF_CONNECTION_SCAN_WINDOW, to_ble_units
 from esphome.components.const import CONF_SCAN_PARAMETERS, CONF_WINDOW
 from esphome.components.esp32 import KEY_IDF_VERSION
 from esphome.components.esp32_ble_tracker import (
@@ -120,3 +121,103 @@ def test_short_interval_without_window_still_rejected(
     stage_esp32("5.5.5", wifi=True)
     with pytest.raises(cv.Invalid, match="needs to be smaller than scan interval"):
         _scan_params({"scan_parameters": {"interval": "20ms"}})
+
+
+# The connection-time fallback window: while a GATT connection is active the
+# scanner drops from a raised full-duty window back to this value so the
+# connection gets guaranteed airtime.
+
+
+def test_raise_arms_connection_scan_window_default(
+    stage_esp32: Callable[..., None],
+) -> None:
+    stage_esp32("5.5.5", wifi=True)
+    params = _scan_params({})
+    assert params[CONF_WINDOW] == params[CONF_INTERVAL]
+    assert to_ble_units(params[CONF_CONNECTION_SCAN_WINDOW]) == 48
+
+
+def test_user_connection_scan_window_survives_raise(
+    stage_esp32: Callable[..., None],
+) -> None:
+    stage_esp32("5.5.5", wifi=True)
+    params = _scan_params({"scan_parameters": {"connection_scan_window": "60ms"}})
+    assert params[CONF_WINDOW] == params[CONF_INTERVAL]
+    assert to_ble_units(params[CONF_CONNECTION_SCAN_WINDOW]) == 96
+
+
+def test_unraised_window_gets_no_connection_scan_window_default(
+    stage_esp32: Callable[..., None],
+) -> None:
+    stage_esp32("5.5.4", wifi=True)
+    assert CONF_CONNECTION_SCAN_WINDOW not in _scan_params({})
+
+
+def test_connection_scan_window_above_interval_rejected(
+    stage_esp32: Callable[..., None],
+) -> None:
+    stage_esp32("5.5.5", wifi=True)
+    with pytest.raises(
+        cv.Invalid, match="connection_scan_window .* needs to be smaller"
+    ):
+        _scan_params({"scan_parameters": {"connection_scan_window": "400ms"}})
+
+
+def test_connection_scan_window_above_window_rejected(
+    stage_esp32: Callable[..., None],
+) -> None:
+    """A connection window above the (post-raise) window would widen the scan
+    during connections; the reject runs after the raise so a fallback below a
+    raised window still validates (covered by the survives-raise test)."""
+    stage_esp32("5.5.5", wifi=True)
+    with pytest.raises(
+        cv.Invalid, match="connection_scan_window .* needs to be smaller"
+    ):
+        _scan_params(
+            {"scan_parameters": {"window": "30ms", "connection_scan_window": "300ms"}}
+        )
+
+
+def test_connection_scan_window_truncation_collapse_rejected(
+    stage_esp32: Callable[..., None],
+) -> None:
+    """A connection window that truncates into the interval's 0.625 ms unit
+    would silently program a full-duty scan during connections."""
+    stage_esp32("5.5.5", wifi=True)
+    with pytest.raises(cv.Invalid, match="connection_scan_window .* both truncate"):
+        _scan_params(
+            {
+                "scan_parameters": {
+                    "interval": "320.5ms",
+                    "connection_scan_window": "320.2ms",
+                }
+            }
+        )
+
+
+@pytest.mark.parametrize(
+    ("config_file", "window_call", "connection_call", "warns"),
+    [
+        # Raised window with GATT clients: the injected fallback is emitted.
+        ("scan_window_raised.yaml", "set_scan_window(512)", True, False),
+        # Explicit window: nothing injected.
+        ("scan_window_explicit.yaml", "set_scan_window(48)", False, False),
+        # Scan-only build compiles the path out: the injected default is
+        # dropped silently, a user-set value warns.
+        ("scan_window_scan_only.yaml", "set_scan_window(512)", False, False),
+        ("scan_window_user_set_scan_only.yaml", "set_scan_window(512)", False, True),
+    ],
+)
+def test_connection_scan_window_codegen(
+    generate_main: Callable[[str | Path], str],
+    component_config_path: Callable[[str], Path],
+    caplog: pytest.LogCaptureFixture,
+    config_file: str,
+    window_call: str,
+    connection_call: bool,
+    warns: bool,
+) -> None:
+    main_cpp = generate_main(component_config_path(config_file))
+    assert window_call in main_cpp
+    assert ("set_connection_scan_window(48)" in main_cpp) == connection_call
+    assert ("'connection_scan_window' has no effect" in caplog.text) == warns

tests/components/camera/__init__.py

@@ -0,0 +1,11 @@
+import esphome.codegen as cg
+from tests.testing_helpers import ComponentManifestOverride
+
+
+def override_manifest(manifest: ComponentManifestOverride) -> None:
+    # No host camera platform exists to emit USE_CAMERA; define it here so
+    # the iterator CAMERA state compiles into the test binary.
+    async def to_code_testing(config):
+        cg.add_define("USE_CAMERA")
+
+    manifest.to_code = to_code_testing

tests/components/camera/test_component_iterator_camera.cpp

@@ -0,0 +1,79 @@
+#include <gtest/gtest.h>
+
+#include "esphome/core/component_iterator.h"
+
+#ifdef USE_CAMERA
+#include "esphome/components/camera/camera.h"
+
+namespace esphome::testing {
+
+class StubCamera : public camera::Camera {
+ public:
+  void add_listener(camera::CameraListener *listener) override {}
+  camera::CameraImageReader *create_image_reader() override { return nullptr; }
+  void request_image(camera::CameraRequester requester) override {}
+  void start_stream(camera::CameraRequester requester) override {}
+  void stop_stream(camera::CameraRequester requester) override {}
+};
+
+// Iterator that accepts everything except the camera, which can refuse a
+// configurable number of times. The CAMERA state is a singleton path
+// distinct from process_platform_item_; this pins the same contract:
+// a refused camera is re-offered, never skipped.
+class CameraRefusingIterator : public ComponentIterator {
+ public:
+// NOLINTBEGIN(bugprone-macro-parentheses)
+#define ENTITY_TYPE_(type, singular, plural, count, upper) \
+  bool on_##singular(type *obj) override { return true; }
+#define ENTITY_CONTROLLER_TYPE_(type, singular, plural, count, upper, callback) \
+  ENTITY_TYPE_(type, singular, plural, count, upper)
+#include "esphome/core/entity_types.h"
+#undef ENTITY_TYPE_
+#undef ENTITY_CONTROLLER_TYPE_
+  // NOLINTEND(bugprone-macro-parentheses)
+
+  bool on_camera(camera::Camera *obj) override {
+    this->camera_calls++;
+    if (this->camera_refusals > 0) {
+      this->camera_refusals--;
+      return false;
+    }
+    return true;
+  }
+
+  int camera_calls{0};
+  int camera_refusals{0};
+};
+
+// Far above the fixed number of iterator states
+static constexpr size_t BIG_BUDGET = 1000;
+
+class ComponentIteratorCameraTest : public ::testing::Test {
+ protected:
+  void SetUp() override {
+    // Constructing a Camera installs the process-wide singleton
+    static StubCamera stub_camera;
+    ASSERT_EQ(camera::Camera::instance(), &stub_camera);
+  }
+};
+
+TEST_F(ComponentIteratorCameraTest, RefusedCameraIsReofferedNotSkipped) {
+  CameraRefusingIterator it;
+  it.camera_refusals = 2;
+  it.begin();
+  // Runs until the camera refuses, which stops the pass
+  it.try_advance(BIG_BUDGET);
+  EXPECT_EQ(it.camera_calls, 1);
+  EXPECT_FALSE(it.completed());
+  // The camera is re-offered once per call, not skipped
+  it.try_advance(BIG_BUDGET);
+  EXPECT_EQ(it.camera_calls, 2);
+  EXPECT_FALSE(it.completed());
+  // Once accepted, the iteration completes
+  it.try_advance(BIG_BUDGET);
+  EXPECT_TRUE(it.completed());
+  EXPECT_EQ(it.camera_calls, 3);
+}
+
+}  // namespace esphome::testing
+#endif  // USE_CAMERA

tests/components/core/benchmark.yaml

@@ -0,0 +1,11 @@
+# Pulls in sensor so entity iteration paths compile (USE_SENSOR);
+# tests register their own instances. Plain yaml.safe_load, no ESPHome tags.
+# An alphabetically-earlier component's sensor: block shadows this one in
+# combined builds; the tests' sensor-count ASSERT catches a capacity drop.
+sensor:
+  - platform: template
+    id: bench_sensor_a
+    name: "Bench A"
+  - platform: template
+    id: bench_sensor_b
+    name: "Bench B"

tests/components/core/test_component_iterator.cpp

@@ -0,0 +1,195 @@
+#include <gtest/gtest.h>
+
+#include "esphome/core/component_iterator.h"
+
+#ifdef USE_SENSOR
+#include "esphome/components/sensor/sensor.h"
+#include "esphome/core/application.h"
+#endif
+
+namespace esphome::testing {
+
+// Iterator whose begin/end callbacks can refuse a configurable number of
+// times; all entity callbacks accept (any registered entities are accepted).
+class RefusingIterator : public ComponentIterator {
+ public:
+// NOLINTBEGIN(bugprone-macro-parentheses)
+#define ENTITY_TYPE_(type, singular, plural, count, upper) \
+  bool on_##singular(type *obj) override { return true; }
+#define ENTITY_CONTROLLER_TYPE_(type, singular, plural, count, upper, callback) \
+  ENTITY_TYPE_(type, singular, plural, count, upper)
+#include "esphome/core/entity_types.h"
+#undef ENTITY_TYPE_
+#undef ENTITY_CONTROLLER_TYPE_
+  // NOLINTEND(bugprone-macro-parentheses)
+
+  bool on_begin() override { return step(this->begin_calls, this->begin_refusals); }
+  bool on_end() override { return step(this->end_calls, this->end_refusals); }
+
+  int begin_calls{0};
+  int end_calls{0};
+  int begin_refusals{0};
+  int end_refusals{0};
+
+ protected:
+  static bool step(int &calls, int &refusals) {
+    calls++;
+    if (refusals > 0) {
+      refusals--;
+      return false;
+    }
+    return true;
+  }
+};
+
+// Far above the fixed number of iterator states
+static constexpr size_t BIG_BUDGET = 1000;
+
+TEST(ComponentIterator, NotRunningMakesNoProgress) {
+  RefusingIterator it;
+  it.try_advance(BIG_BUDGET);
+  EXPECT_TRUE(it.completed());
+  EXPECT_EQ(it.begin_calls, 0);
+  EXPECT_EQ(it.end_calls, 0);
+}
+
+TEST(ComponentIterator, CompletesInOneCallWithoutRefusals) {
+  RefusingIterator it;
+  it.begin();
+  it.try_advance(BIG_BUDGET);
+  EXPECT_TRUE(it.completed());
+  EXPECT_EQ(it.begin_calls, 1);
+  EXPECT_EQ(it.end_calls, 1);
+}
+
+TEST(ComponentIterator, StepBudgetIsHonored) {
+  RefusingIterator it;
+  it.begin();
+  it.try_advance(1);
+  EXPECT_EQ(it.begin_calls, 1);
+  EXPECT_EQ(it.end_calls, 0);
+  EXPECT_FALSE(it.completed());
+}
+
+TEST(ComponentIterator, RefusedStepStopsBatchAndRetriesSameStep) {
+  RefusingIterator it;
+  it.end_refusals = 3;
+  it.begin();
+  // First call runs until the refused end step, which stops the pass
+  it.try_advance(BIG_BUDGET);
+  EXPECT_EQ(it.end_calls, 1);
+  EXPECT_FALSE(it.completed());
+  // The refused step is retried once per call, not skipped
+  it.try_advance(BIG_BUDGET);
+  it.try_advance(BIG_BUDGET);
+  EXPECT_EQ(it.end_calls, 3);
+  EXPECT_FALSE(it.completed());
+  // Once accepted, the iteration completes
+  it.try_advance(BIG_BUDGET);
+  EXPECT_TRUE(it.completed());
+  EXPECT_EQ(it.end_calls, 4);
+}
+
+TEST(ComponentIterator, RefusedBeginStopsBatchAndRetries) {
+  RefusingIterator it;
+  it.begin_refusals = 2;
+  it.begin();
+  it.try_advance(BIG_BUDGET);
+  it.try_advance(BIG_BUDGET);
+  EXPECT_EQ(it.begin_calls, 2);
+  EXPECT_FALSE(it.completed());
+  it.try_advance(BIG_BUDGET);
+  EXPECT_TRUE(it.completed());
+  EXPECT_EQ(it.begin_calls, 3);
+}
+
+// The deprecated advance() wrapper must keep the legacy once-per-loop
+// pattern working during the deprecation window.
+#pragma GCC diagnostic push
+#pragma GCC diagnostic ignored "-Wdeprecated-declarations"
+TEST(ComponentIterator, DeprecatedAdvanceKeepsLegacyPatternWorking) {
+  RefusingIterator it;
+  it.end_refusals = 2;
+  it.begin();
+  size_t guard = 0;
+  while (!it.completed() && guard++ < BIG_BUDGET) {
+    it.advance();
+  }
+  EXPECT_TRUE(it.completed());
+  // Two refused end steps were retried, then accepted
+  EXPECT_EQ(it.end_calls, 3);
+}
+#pragma GCC diagnostic pop
+
+#ifdef USE_SENSOR
+// Iterator whose sensor callback can refuse or yield; pins the per-item
+// contract: a refused item is re-offered with at_ unchanged, never skipped.
+class ItemRefusingIterator : public RefusingIterator {
+ public:
+  bool on_sensor(sensor::Sensor *obj) override {
+    this->last_sensor = obj;
+    if (!step(this->sensor_calls, this->sensor_refusals))
+      return false;
+    if (this->yield_on_sensor)
+      this->yield_after_step_();
+    return true;
+  }
+  sensor::Sensor *last_sensor{nullptr};
+  int sensor_calls{0};
+  int sensor_refusals{0};
+  bool yield_on_sensor{false};
+};
+
+class ComponentIteratorSensorTest : public ::testing::Test {
+ protected:
+  void SetUp() override {
+    static sensor::Sensor sensor_a;
+    static sensor::Sensor sensor_b;
+    static bool registered = false;
+    if (!registered) {
+      App.register_sensor(&sensor_a);
+      App.register_sensor(&sensor_b);
+      registered = true;
+    }
+    // StaticVector drops silently when full; fail the fixture, not the contract
+    ASSERT_EQ(App.get_sensors().size(), 2u) << "benchmark.yaml sensor count too small";
+  }
+};
+
+TEST_F(ComponentIteratorSensorTest, RefusedItemIsReofferedNotSkipped) {
+  ItemRefusingIterator it;
+  it.sensor_refusals = 2;
+  it.begin();
+  // Runs until the first sensor refuses
+  it.try_advance(BIG_BUDGET);
+  EXPECT_EQ(it.sensor_calls, 1);
+  EXPECT_FALSE(it.completed());
+  // The refused item is re-offered, not skipped
+  it.try_advance(BIG_BUDGET);
+  EXPECT_EQ(it.sensor_calls, 2);
+  sensor::Sensor *refused = it.last_sensor;
+  // Once accepted, iteration continues through the second sensor to the end
+  it.try_advance(BIG_BUDGET);
+  EXPECT_TRUE(it.completed());
+  EXPECT_NE(it.last_sensor, refused);
+  EXPECT_EQ(it.sensor_calls, 4);
+}
+
+TEST_F(ComponentIteratorSensorTest, YieldAfterStepEndsPassAndResumes) {
+  ItemRefusingIterator it;
+  it.yield_on_sensor = true;
+  it.begin();
+  // The pass ends right after the first sensor despite a big budget
+  it.try_advance(BIG_BUDGET);
+  EXPECT_EQ(it.sensor_calls, 1);
+  EXPECT_FALSE(it.completed());
+  // The next pass ends after the second sensor
+  it.try_advance(BIG_BUDGET);
+  EXPECT_EQ(it.sensor_calls, 2);
+  // Remaining states then run to completion in one pass
+  it.try_advance(BIG_BUDGET);
+  EXPECT_TRUE(it.completed());
+}
+#endif  // USE_SENSOR
+
+}  // namespace esphome::testing

tests/components/mitsubishi_cn105/climate/mitsubishi_cn105_climate_tests.cpp

@@ -2,10 +2,19 @@
 #include <utility>
 #include "../common.h"
 
+#include "esphome/components/mitsubishi_cn105/mitsubishi_cn105_climate.h"
+
 namespace esphome::mitsubishi_cn105::testing {
 
+struct MitsubishiCN105ClimateTestContext {
+  MitsubishiCN105Component component;
+  MitsubishiCN105Climate sut;
+
+  MitsubishiCN105ClimateTestContext() { this->sut.set_parent(&this->component); }
+};
+
 TEST(MitsubishiCN105ClimateTests, CelsiusTemperatureMappingAndTraitsMatchExpectedValues) {
-  TestableMitsubishiCN105Climate sut;
+  MitsubishiCN105ClimateTestContext context;
   const auto mapping = TemperatureMapping();
 
   for (int temperature = 16; temperature <= 31; ++temperature) {
@@ -13,7 +22,7 @@ TEST(MitsubishiCN105ClimateTests, CelsiusTemperatureMappingAndTraitsMatchExpecte
     EXPECT_EQ(mapping.from_mitsubishi(temperature), temperature);
   }
 
-  const auto traits = sut.traits();
+  const auto traits = context.sut.traits();
   EXPECT_EQ(traits.get_temperature_unit(), TemperatureUnit::CELSIUS);
   EXPECT_FLOAT_EQ(traits.get_visual_min_temperature(), 16.0f);
   EXPECT_FLOAT_EQ(traits.get_visual_max_temperature(), 31.0f);
@@ -22,10 +31,10 @@ TEST(MitsubishiCN105ClimateTests, CelsiusTemperatureMappingAndTraitsMatchExpecte
 }
 
 TEST(MitsubishiCN105ClimateTests, FahrenheitTemperatureMappingAndTraitsMatchExpectedValues) {
-  TestableMitsubishiCN105Climate sut;
+  MitsubishiCN105ClimateTestContext context;
   auto mapping = TemperatureMapping();
   mapping.set_use_fahrenheit(true);
-  sut.set_use_fahrenheit(true);
+  context.component.set_use_fahrenheit(true);
 
   const std::array cases{
       std::pair{61, 16.0f}, std::pair{62, 16.5f}, std::pair{63, 17.0f}, std::pair{64, 17.5f}, std::pair{65, 18.0f},
@@ -40,7 +49,7 @@ TEST(MitsubishiCN105ClimateTests, FahrenheitTemperatureMappingAndTraitsMatchExpe
     EXPECT_FLOAT_EQ(mapping.to_mitsubishi(fahrenheit), mitsubishi_celsius);
     EXPECT_FLOAT_EQ(mapping.from_mitsubishi(mitsubishi_celsius), fahrenheit);
   }
-  const auto traits = sut.traits();
+  const auto traits = context.sut.traits();
   EXPECT_EQ(traits.get_temperature_unit(), TemperatureUnit::FAHRENHEIT);
   EXPECT_FLOAT_EQ(traits.get_visual_min_temperature(), 61.0f);
   EXPECT_FLOAT_EQ(traits.get_visual_max_temperature(), 88.0f);
@@ -63,163 +72,44 @@ TEST(MitsubishiCN105ClimateTests, FahrenheitTemperatureMappingUsesLinearConversi
 }
 
 TEST(MitsubishiCN105ClimateTests, SupportedSwingModeOffLeavesTraitsEmpty) {
-  TestableMitsubishiCN105Climate sut;
+  MitsubishiCN105ClimateTestContext context;
 
-  sut.set_supported_swing_mode(climate::CLIMATE_SWING_OFF);
+  context.sut.set_supported_swing_mode(climate::CLIMATE_SWING_OFF);
 
-  EXPECT_FALSE(sut.traits().get_supports_swing_modes());
+  EXPECT_FALSE(context.sut.traits().get_supports_swing_modes());
 }
 
 TEST(MitsubishiCN105ClimateTests, SupportedSwingModeVerticalExposesOffAndVertical) {
-  TestableMitsubishiCN105Climate sut;
+  MitsubishiCN105ClimateTestContext context;
 
-  sut.set_supported_swing_mode(climate::CLIMATE_SWING_VERTICAL);
+  context.sut.set_supported_swing_mode(climate::CLIMATE_SWING_VERTICAL);
 
-  EXPECT_TRUE(sut.traits().supports_swing_mode(climate::CLIMATE_SWING_OFF));
-  EXPECT_TRUE(sut.traits().supports_swing_mode(climate::CLIMATE_SWING_VERTICAL));
-  EXPECT_FALSE(sut.traits().supports_swing_mode(climate::CLIMATE_SWING_HORIZONTAL));
-  EXPECT_FALSE(sut.traits().supports_swing_mode(climate::CLIMATE_SWING_BOTH));
+  EXPECT_TRUE(context.sut.traits().supports_swing_mode(climate::CLIMATE_SWING_OFF));
+  EXPECT_TRUE(context.sut.traits().supports_swing_mode(climate::CLIMATE_SWING_VERTICAL));
+  EXPECT_FALSE(context.sut.traits().supports_swing_mode(climate::CLIMATE_SWING_HORIZONTAL));
+  EXPECT_FALSE(context.sut.traits().supports_swing_mode(climate::CLIMATE_SWING_BOTH));
 }
 
 TEST(MitsubishiCN105ClimateTests, SupportedSwingModeHorizontalExposesOffAndHorizontal) {
-  TestableMitsubishiCN105Climate sut;
+  MitsubishiCN105ClimateTestContext context;
 
-  sut.set_supported_swing_mode(climate::CLIMATE_SWING_HORIZONTAL);
+  context.sut.set_supported_swing_mode(climate::CLIMATE_SWING_HORIZONTAL);
 
-  EXPECT_TRUE(sut.traits().supports_swing_mode(climate::CLIMATE_SWING_OFF));
-  EXPECT_FALSE(sut.traits().supports_swing_mode(climate::CLIMATE_SWING_VERTICAL));
-  EXPECT_TRUE(sut.traits().supports_swing_mode(climate::CLIMATE_SWING_HORIZONTAL));
-  EXPECT_FALSE(sut.traits().supports_swing_mode(climate::CLIMATE_SWING_BOTH));
+  EXPECT_TRUE(context.sut.traits().supports_swing_mode(climate::CLIMATE_SWING_OFF));
+  EXPECT_FALSE(context.sut.traits().supports_swing_mode(climate::CLIMATE_SWING_VERTICAL));
+  EXPECT_TRUE(context.sut.traits().supports_swing_mode(climate::CLIMATE_SWING_HORIZONTAL));
+  EXPECT_FALSE(context.sut.traits().supports_swing_mode(climate::CLIMATE_SWING_BOTH));
 }
 
 TEST(MitsubishiCN105ClimateTests, SupportedSwingModeBothExposesAllExpectedModes) {
-  TestableMitsubishiCN105Climate sut;
+  MitsubishiCN105ClimateTestContext context;
 
-  sut.set_supported_swing_mode(climate::CLIMATE_SWING_BOTH);
+  context.sut.set_supported_swing_mode(climate::CLIMATE_SWING_BOTH);
 
-  EXPECT_TRUE(sut.traits().supports_swing_mode(climate::CLIMATE_SWING_OFF));
-  EXPECT_TRUE(sut.traits().supports_swing_mode(climate::CLIMATE_SWING_VERTICAL));
-  EXPECT_TRUE(sut.traits().supports_swing_mode(climate::CLIMATE_SWING_HORIZONTAL));
-  EXPECT_TRUE(sut.traits().supports_swing_mode(climate::CLIMATE_SWING_BOTH));
-}
-
-TEST(MitsubishiCN105ClimateTests, ApplyValuesMapsVerticalSwingWhenSupported) {
-  TestableMitsubishiCN105Climate sut;
-
-  sut.set_supported_swing_mode(climate::CLIMATE_SWING_VERTICAL);
-
-  sut.status().vane_mode = MitsubishiCN105::VaneMode::SWING;
-  sut.status().wide_vane_mode = MitsubishiCN105::WideVaneMode::CENTER;
-
-  sut.apply_values_();
-
-  EXPECT_EQ(sut.swing_mode, climate::CLIMATE_SWING_VERTICAL);
-}
-
-TEST(MitsubishiCN105ClimateTests, ApplyValuesMapsHorizontalSwingWhenSupported) {
-  TestableMitsubishiCN105Climate sut;
-
-  sut.set_supported_swing_mode(climate::CLIMATE_SWING_HORIZONTAL);
-
-  sut.status().vane_mode = MitsubishiCN105::VaneMode::AUTO;
-  sut.status().wide_vane_mode = MitsubishiCN105::WideVaneMode::SWING;
-
-  sut.apply_values_();
-
-  EXPECT_EQ(sut.swing_mode, climate::CLIMATE_SWING_HORIZONTAL);
-}
-
-TEST(MitsubishiCN105ClimateTests, ApplyValuesMapsBothSwingWhenSupported) {
-  TestableMitsubishiCN105Climate sut;
-
-  sut.set_supported_swing_mode(climate::CLIMATE_SWING_BOTH);
-
-  sut.status().vane_mode = MitsubishiCN105::VaneMode::SWING;
-  sut.status().wide_vane_mode = MitsubishiCN105::WideVaneMode::SWING;
-
-  sut.apply_values_();
-
-  EXPECT_EQ(sut.swing_mode, climate::CLIMATE_SWING_BOTH);
-}
-
-TEST(MitsubishiCN105ClimateTests, ApplyValuesMapsSwingOffWhenNoSwingActive) {
-  TestableMitsubishiCN105Climate sut;
-
-  sut.set_supported_swing_mode(climate::CLIMATE_SWING_BOTH);
-
-  sut.status().vane_mode = MitsubishiCN105::VaneMode::POSITION_3;
-  sut.status().wide_vane_mode = MitsubishiCN105::WideVaneMode::CENTER;
-
-  sut.apply_values_();
-
-  EXPECT_EQ(sut.swing_mode, climate::CLIMATE_SWING_OFF);
-}
-
-TEST(MitsubishiCN105ClimateTests, ApplyValuesRemembersLastNonSwingPositions) {
-  TestableMitsubishiCN105Climate sut;
-
-  sut.set_supported_swing_mode(climate::CLIMATE_SWING_BOTH);
-
-  sut.status().vane_mode = MitsubishiCN105::VaneMode::POSITION_4;
-  sut.status().wide_vane_mode = MitsubishiCN105::WideVaneMode::RIGHT;
-
-  sut.apply_values_();
-
-  EXPECT_EQ(sut.last_non_swing_vane_mode_, MitsubishiCN105::VaneMode::POSITION_4);
-  EXPECT_EQ(sut.last_non_swing_wide_vane_mode_, MitsubishiCN105::WideVaneMode::RIGHT);
-
-  sut.status().vane_mode = MitsubishiCN105::VaneMode::SWING;
-  sut.status().wide_vane_mode = MitsubishiCN105::WideVaneMode::SWING;
-
-  sut.apply_values_();
-
-  EXPECT_EQ(sut.last_non_swing_vane_mode_, MitsubishiCN105::VaneMode::POSITION_4);
-  EXPECT_EQ(sut.last_non_swing_wide_vane_mode_, MitsubishiCN105::WideVaneMode::RIGHT);
-  EXPECT_EQ(sut.swing_mode, climate::CLIMATE_SWING_BOTH);
-}
-
-TEST(MitsubishiCN105ClimateTests, ApplyValuesDoesNotOverwriteRememberedPositionWithUnknownValues) {
-  TestableMitsubishiCN105Climate sut;
-
-  sut.set_supported_swing_mode(climate::CLIMATE_SWING_BOTH);
-
-  sut.last_non_swing_vane_mode_ = MitsubishiCN105::VaneMode::POSITION_2;
-  sut.last_non_swing_wide_vane_mode_ = MitsubishiCN105::WideVaneMode::LEFT;
-
-  sut.status().vane_mode = MitsubishiCN105::VaneMode::UNKNOWN;
-  sut.status().wide_vane_mode = MitsubishiCN105::WideVaneMode::UNKNOWN;
-
-  sut.apply_values_();
-
-  EXPECT_EQ(sut.last_non_swing_vane_mode_, MitsubishiCN105::VaneMode::POSITION_2);
-  EXPECT_EQ(sut.last_non_swing_wide_vane_mode_, MitsubishiCN105::WideVaneMode::LEFT);
-  EXPECT_EQ(sut.swing_mode, climate::CLIMATE_SWING_OFF);
-}
-
-TEST(MitsubishiCN105ClimateTests, ApplyValuesIgnoresUnsupportedVerticalSwingState) {
-  TestableMitsubishiCN105Climate sut;
-
-  sut.set_supported_swing_mode(climate::CLIMATE_SWING_HORIZONTAL);
-
-  sut.status().vane_mode = MitsubishiCN105::VaneMode::SWING;
-  sut.status().wide_vane_mode = MitsubishiCN105::WideVaneMode::CENTER;
-
-  sut.apply_values_();
-
-  EXPECT_EQ(sut.swing_mode, climate::CLIMATE_SWING_OFF);
-}
-
-TEST(MitsubishiCN105ClimateTests, ApplyValuesIgnoresUnsupportedHorizontalSwingState) {
-  TestableMitsubishiCN105Climate sut;
-
-  sut.set_supported_swing_mode(climate::CLIMATE_SWING_VERTICAL);
-
-  sut.status().vane_mode = MitsubishiCN105::VaneMode::AUTO;
-  sut.status().wide_vane_mode = MitsubishiCN105::WideVaneMode::SWING;
-
-  sut.apply_values_();
-
-  EXPECT_EQ(sut.swing_mode, climate::CLIMATE_SWING_OFF);
+  EXPECT_TRUE(context.sut.traits().supports_swing_mode(climate::CLIMATE_SWING_OFF));
+  EXPECT_TRUE(context.sut.traits().supports_swing_mode(climate::CLIMATE_SWING_VERTICAL));
+  EXPECT_TRUE(context.sut.traits().supports_swing_mode(climate::CLIMATE_SWING_HORIZONTAL));
+  EXPECT_TRUE(context.sut.traits().supports_swing_mode(climate::CLIMATE_SWING_BOTH));
 }
 
 }  // namespace esphome::mitsubishi_cn105::testing

tests/components/mitsubishi_cn105/climate/mitsubishi_cn105_swing_mode_manager_tests.cpp

@@ -0,0 +1,99 @@
+#include "../common.h"
+
+#include "esphome/components/mitsubishi_cn105/mitsubishi_cn105_swing_mode_manager.h"
+
+namespace esphome::mitsubishi_cn105::testing {
+
+static SwingModeManager make_swing_mode_manager(std::initializer_list<climate::ClimateSwingMode> supported_modes) {
+  SwingModeManager manager;
+  climate::ClimateSwingModeMask supported_swing_modes;
+  for (const auto mode : supported_modes)
+    supported_swing_modes.insert(mode);
+  manager.set_supported_swing_modes(supported_swing_modes);
+  return manager;
+}
+
+TEST(SwingModeManagerTests, StatusMapsVerticalSwingWhenSupported) {
+  auto manager = make_swing_mode_manager({climate::CLIMATE_SWING_OFF, climate::CLIMATE_SWING_VERTICAL});
+  EXPECT_EQ(manager.update_and_get_swing_mode(MitsubishiCN105::VaneMode::SWING, MitsubishiCN105::WideVaneMode::CENTER),
+            std::optional{climate::CLIMATE_SWING_VERTICAL});
+}
+
+TEST(SwingModeManagerTests, StatusMapsHorizontalSwingWhenSupported) {
+  auto manager = make_swing_mode_manager({climate::CLIMATE_SWING_OFF, climate::CLIMATE_SWING_HORIZONTAL});
+  EXPECT_EQ(manager.update_and_get_swing_mode(MitsubishiCN105::VaneMode::AUTO, MitsubishiCN105::WideVaneMode::SWING),
+            std::optional{climate::CLIMATE_SWING_HORIZONTAL});
+}
+
+TEST(SwingModeManagerTests, StatusMapsBothSwingWhenSupported) {
+  auto manager = make_swing_mode_manager({climate::CLIMATE_SWING_OFF, climate::CLIMATE_SWING_VERTICAL,
+                                          climate::CLIMATE_SWING_HORIZONTAL, climate::CLIMATE_SWING_BOTH});
+  EXPECT_EQ(manager.update_and_get_swing_mode(MitsubishiCN105::VaneMode::SWING, MitsubishiCN105::WideVaneMode::SWING),
+            std::optional{climate::CLIMATE_SWING_BOTH});
+}
+
+TEST(SwingModeManagerTests, StatusMapsSwingOffWhenNoSwingActive) {
+  auto manager = make_swing_mode_manager({climate::CLIMATE_SWING_OFF, climate::CLIMATE_SWING_VERTICAL,
+                                          climate::CLIMATE_SWING_HORIZONTAL, climate::CLIMATE_SWING_BOTH});
+  EXPECT_EQ(
+      manager.update_and_get_swing_mode(MitsubishiCN105::VaneMode::POSITION_3, MitsubishiCN105::WideVaneMode::CENTER),
+      std::optional{climate::CLIMATE_SWING_OFF});
+}
+
+TEST(SwingModeManagerTests, RemembersLastNonSwingPositions) {
+  auto manager = make_swing_mode_manager({climate::CLIMATE_SWING_OFF, climate::CLIMATE_SWING_VERTICAL,
+                                          climate::CLIMATE_SWING_HORIZONTAL, climate::CLIMATE_SWING_BOTH});
+  manager.update_and_get_swing_mode(MitsubishiCN105::VaneMode::POSITION_4, MitsubishiCN105::WideVaneMode::RIGHT);
+  manager.update_and_get_swing_mode(MitsubishiCN105::VaneMode::SWING, MitsubishiCN105::WideVaneMode::SWING);
+  EXPECT_EQ(manager.vane_from(climate::CLIMATE_SWING_OFF), std::optional{MitsubishiCN105::VaneMode::POSITION_4});
+  EXPECT_EQ(manager.wide_vane_from(climate::CLIMATE_SWING_OFF), std::optional{MitsubishiCN105::WideVaneMode::RIGHT});
+}
+
+TEST(SwingModeManagerTests, UnknownValuesDoNotOverwriteRememberedPositions) {
+  auto manager = make_swing_mode_manager({climate::CLIMATE_SWING_OFF, climate::CLIMATE_SWING_VERTICAL,
+                                          climate::CLIMATE_SWING_HORIZONTAL, climate::CLIMATE_SWING_BOTH});
+  manager.update_and_get_swing_mode(MitsubishiCN105::VaneMode::POSITION_2, MitsubishiCN105::WideVaneMode::LEFT);
+  manager.update_and_get_swing_mode(MitsubishiCN105::VaneMode::UNKNOWN, MitsubishiCN105::WideVaneMode::UNKNOWN);
+  EXPECT_EQ(manager.vane_from(climate::CLIMATE_SWING_OFF), std::optional{MitsubishiCN105::VaneMode::POSITION_2});
+  EXPECT_EQ(manager.wide_vane_from(climate::CLIMATE_SWING_OFF), std::optional{MitsubishiCN105::WideVaneMode::LEFT});
+}
+
+TEST(SwingModeManagerTests, UnsupportedVerticalSwingStateIsIgnored) {
+  auto manager = make_swing_mode_manager({climate::CLIMATE_SWING_OFF, climate::CLIMATE_SWING_HORIZONTAL});
+  EXPECT_EQ(manager.update_and_get_swing_mode(MitsubishiCN105::VaneMode::SWING, MitsubishiCN105::WideVaneMode::CENTER),
+            std::optional{climate::CLIMATE_SWING_OFF});
+}
+
+TEST(SwingModeManagerTests, UnsupportedHorizontalSwingStateIsIgnored) {
+  auto manager = make_swing_mode_manager({climate::CLIMATE_SWING_OFF, climate::CLIMATE_SWING_VERTICAL});
+  EXPECT_EQ(manager.update_and_get_swing_mode(MitsubishiCN105::VaneMode::AUTO, MitsubishiCN105::WideVaneMode::SWING),
+            std::optional{climate::CLIMATE_SWING_OFF});
+}
+
+TEST(SwingModeManagerTests, SwingModeFromReturnsNulloptWhenNoSwingModesSupported) {
+  auto manager = make_swing_mode_manager({});
+  EXPECT_FALSE(manager.update_and_get_swing_mode(MitsubishiCN105::VaneMode::SWING, MitsubishiCN105::WideVaneMode::SWING)
+                   .has_value());
+}
+
+TEST(SwingModeManagerTests, VaneFromSwingModeReturnsNulloptWhenVerticalUnsupported) {
+  auto manager = make_swing_mode_manager({climate::CLIMATE_SWING_OFF, climate::CLIMATE_SWING_HORIZONTAL});
+  EXPECT_FALSE(manager.vane_from(climate::CLIMATE_SWING_VERTICAL).has_value());
+}
+
+TEST(SwingModeManagerTests, WideVaneFromSwingModeReturnsNulloptWhenHorizontalUnsupported) {
+  auto manager = make_swing_mode_manager({climate::CLIMATE_SWING_OFF, climate::CLIMATE_SWING_VERTICAL});
+  EXPECT_FALSE(manager.wide_vane_from(climate::CLIMATE_SWING_HORIZONTAL).has_value());
+}
+
+TEST(SwingModeManagerTests, VaneAndWideVaneFromSwingModeMapSwingModes) {
+  auto manager = make_swing_mode_manager({climate::CLIMATE_SWING_OFF, climate::CLIMATE_SWING_VERTICAL,
+                                          climate::CLIMATE_SWING_HORIZONTAL, climate::CLIMATE_SWING_BOTH});
+  EXPECT_EQ(manager.vane_from(climate::CLIMATE_SWING_VERTICAL), std::optional{MitsubishiCN105::VaneMode::SWING});
+  EXPECT_EQ(manager.vane_from(climate::CLIMATE_SWING_BOTH), std::optional{MitsubishiCN105::VaneMode::SWING});
+  EXPECT_EQ(manager.wide_vane_from(climate::CLIMATE_SWING_HORIZONTAL),
+            std::optional{MitsubishiCN105::WideVaneMode::SWING});
+  EXPECT_EQ(manager.wide_vane_from(climate::CLIMATE_SWING_BOTH), std::optional{MitsubishiCN105::WideVaneMode::SWING});
+}
+
+}  // namespace esphome::mitsubishi_cn105::testing

tests/components/mitsubishi_cn105/common.h

@@ -64,26 +64,4 @@ class TestableMitsubishiCN105 : public MitsubishiCN105 {
   void set_current_time(uint32_t ms) { test_loop_time_ms = ms; }
 };
 
-class TestableMitsubishiCN105Climate : public MitsubishiCN105Climate {
- public:
-  TestableMitsubishiCN105Climate() { this->set_parent(&this->component_); }
-
-  using MitsubishiCN105Climate::apply_values_;
-  using MitsubishiCN105Climate::last_non_swing_vane_mode_;
-  using MitsubishiCN105Climate::last_non_swing_wide_vane_mode_;
-
-  MitsubishiCN105::Status &status() { return const_cast<MitsubishiCN105::Status &>(this->component_.status()); }
-  void set_use_fahrenheit(bool value) { this->component_.set_use_fahrenheit(value); }
-
- protected:
-  MitsubishiCN105Component component_;
-};
-
-class TestableMitsubishiCN105Component : public MitsubishiCN105Component {
- public:
-  MitsubishiCN105::Status &mutable_status() { return const_cast<MitsubishiCN105::Status &>(this->status()); }
-
-  void notify_status() { this->status_callback_.call(); }
-};
-
 }  // namespace esphome::mitsubishi_cn105::testing

tests/components/mitsubishi_cn105/mitsubishi_cn105_component_tests.cpp

@@ -3,7 +3,7 @@
 namespace esphome::mitsubishi_cn105::testing {
 
 TEST(MitsubishiCN105ComponentTests, PublishesVaneStateForEveryValidSnapshot) {
-  TestableMitsubishiCN105Component hub;
+  MitsubishiCN105Component hub;
   size_t callback_count = 0;
   std::optional<VerticalVaneMode> callback_direction;
   hub.add_on_vane_state_callback([&](const VaneState &state) {
@@ -11,8 +11,9 @@ TEST(MitsubishiCN105ComponentTests, PublishesVaneStateForEveryValidSnapshot) {
     callback_direction = state.vertical.direction;
   });
 
-  hub.mutable_status().room_temperature = 20.0f;
-  hub.mutable_status().vane_mode = MitsubishiCN105::VaneMode::POSITION_4;
+  hub.set_telemetry_request_min_interval(SCHEDULER_DONT_RUN);
+  hub.set_target_temperature(20.0f);
+  hub.set_vane_mode(MitsubishiCN105::VaneMode::POSITION_4);
   hub.publish_status();
 
   EXPECT_EQ(callback_count, 1);
@@ -25,7 +26,7 @@ TEST(MitsubishiCN105ComponentTests, PublishesVaneStateForEveryValidSnapshot) {
 }
 
 TEST(MitsubishiCN105ComponentTests, PublishesUnknownVaneState) {
-  TestableMitsubishiCN105Component hub;
+  MitsubishiCN105Component hub;
   size_t status_callback_count = 0;
   size_t vane_callback_count = 0;
   std::optional<VerticalVaneMode> callback_direction;
@@ -35,15 +36,16 @@ TEST(MitsubishiCN105ComponentTests, PublishesUnknownVaneState) {
     callback_direction = state.vertical.direction;
   });
 
-  hub.mutable_status().room_temperature = 20.0f;
-  hub.mutable_status().vane_mode = MitsubishiCN105::VaneMode::UNKNOWN;
+  hub.set_telemetry_request_min_interval(SCHEDULER_DONT_RUN);
+  hub.set_target_temperature(20.0f);
+  ASSERT_EQ(hub.status().vane_mode, MitsubishiCN105::VaneMode::UNKNOWN);
   hub.publish_status();
 
   EXPECT_EQ(status_callback_count, 1);
   EXPECT_EQ(vane_callback_count, 1);
   EXPECT_EQ(callback_direction, std::optional{VERTICAL_VANE_MODE_UNKNOWN});
 
-  hub.mutable_status().vane_mode = MitsubishiCN105::VaneMode::POSITION_4;
+  hub.set_vane_mode(MitsubishiCN105::VaneMode::POSITION_4);
   hub.publish_status();
 
   EXPECT_EQ(status_callback_count, 2);
@@ -52,7 +54,7 @@ TEST(MitsubishiCN105ComponentTests, PublishesUnknownVaneState) {
 }
 
 TEST(MitsubishiCN105ComponentTests, VaneCallAppliesVerticalDirection) {
-  TestableMitsubishiCN105Component hub;
+  MitsubishiCN105Component hub;
 
   auto call = hub.make_vane_call();
   call.vertical.set_direction(VERTICAL_VANE_MODE_POSITION_5);
@@ -62,12 +64,11 @@ TEST(MitsubishiCN105ComponentTests, VaneCallAppliesVerticalDirection) {
 }
 
 TEST(MitsubishiCN105ComponentTests, VaneControlActionAppliesConfiguredFields) {
-  TestableMitsubishiCN105Component hub;
+  MitsubishiCN105Component hub;
   VaneControlAction<> action(&hub, [](VaneCall &call) { call.vertical.set_direction(VERTICAL_VANE_MODE_SWING); });
 
   action.play();
 
   EXPECT_EQ(hub.status().vane_mode, MitsubishiCN105::VaneMode::SWING);
 }
-
 }  // namespace esphome::mitsubishi_cn105::testing

tests/components/mitsubishi_cn105/select/mitsubishi_cn105_vane_select_vertical_tests.cpp

@@ -3,14 +3,9 @@
 
 namespace esphome::mitsubishi_cn105::testing {
 
-class TestableMitsubishiCN105VerticalVaneDirectionSelect : public MitsubishiCN105VerticalVaneDirectionSelect {
- public:
-  using MitsubishiCN105VerticalVaneDirectionSelect::control;
-};
-
 struct VerticalVaneDirectionSelectTestContext {
-  TestableMitsubishiCN105Component hub;
-  TestableMitsubishiCN105VerticalVaneDirectionSelect select;
+  MitsubishiCN105Component hub;
+  MitsubishiCN105VerticalVaneDirectionSelect select;
 
   VerticalVaneDirectionSelectTestContext() {
     this->select.traits.set_options({"Auto", "1", "2", "3", "4", "5", "Swing"});
@@ -31,13 +26,15 @@ TEST(MitsubishiCN105VerticalVaneDirectionSelectTests, MapsIndexesToVaneModes) {
 
   for (size_t i = 0; i < expected_modes.size(); ++i) {
     SCOPED_TRACE(i);
-    ctx.select.control(i);
+    ctx.select.make_call().set_index(i).perform();
     EXPECT_EQ(ctx.hub.status().vane_mode, expected_modes[i]);
   }
 }
 
 TEST(MitsubishiCN105VerticalVaneDirectionSelectTests, PublishesIncomingVaneModes) {
   VerticalVaneDirectionSelectTestContext ctx;
+  ctx.hub.set_telemetry_request_min_interval(SCHEDULER_DONT_RUN);
+  ctx.hub.set_target_temperature(20.0f);
 
   constexpr std::array modes{
       MitsubishiCN105::VaneMode::AUTO,       MitsubishiCN105::VaneMode::POSITION_1,
@@ -48,13 +45,12 @@ TEST(MitsubishiCN105VerticalVaneDirectionSelectTests, PublishesIncomingVaneModes
 
   for (size_t i = 0; i < modes.size(); ++i) {
     SCOPED_TRACE(i);
-    ctx.hub.mutable_status().vane_mode = modes[i];
-    ctx.hub.notify_status();
+    ctx.hub.set_vane_mode(modes[i]);
+    ctx.hub.publish_status();
     EXPECT_EQ(ctx.select.active_index(), std::optional{i});
   }
 
-  ctx.hub.mutable_status().vane_mode = MitsubishiCN105::VaneMode::UNKNOWN;
-  ctx.hub.notify_status();
+  ctx.select.publish_vane_state(MitsubishiCN105::VaneMode::UNKNOWN);
   EXPECT_EQ(ctx.select.active_index(), std::optional{modes.size() - 1});
 }
 
@@ -64,14 +60,15 @@ TEST(MitsubishiCN105VerticalVaneDirectionSelectTests, ControlPublishesSelectAndC
   climate_entity.set_parent(&ctx.hub);
   climate_entity.set_supported_swing_mode(climate::CLIMATE_SWING_VERTICAL);
 
-  ctx.hub.mutable_status().room_temperature = 20.0f;
+  ctx.hub.set_telemetry_request_min_interval(SCHEDULER_DONT_RUN);
+  ctx.hub.set_target_temperature(20.0f);
   climate_entity.setup();
 
-  ctx.select.control(6);
+  ctx.select.make_call().set_index(6).perform();
   EXPECT_EQ(ctx.select.active_index(), std::optional<size_t>{6});
   EXPECT_EQ(climate_entity.swing_mode, climate::CLIMATE_SWING_VERTICAL);
 
-  ctx.select.control(3);
+  ctx.select.make_call().set_index(3).perform();
   EXPECT_EQ(ctx.select.active_index(), std::optional<size_t>{3});
   EXPECT_EQ(climate_entity.swing_mode, climate::CLIMATE_SWING_OFF);
 }
@@ -82,7 +79,8 @@ TEST(MitsubishiCN105VerticalVaneDirectionSelectTests, ClimateControlPublishesSel
   climate_entity.set_parent(&ctx.hub);
   climate_entity.set_supported_swing_mode(climate::CLIMATE_SWING_VERTICAL);
 
-  ctx.hub.mutable_status().room_temperature = 20.0f;
+  ctx.hub.set_telemetry_request_min_interval(SCHEDULER_DONT_RUN);
+  ctx.hub.set_target_temperature(20.0f);
   climate_entity.setup();
 
   climate_entity.make_call().set_swing_mode(climate::CLIMATE_SWING_VERTICAL).perform();
@@ -95,10 +93,9 @@ TEST(MitsubishiCN105VerticalVaneDirectionSelectTests, ClimateControlPublishesSel
 TEST(MitsubishiCN105VerticalVaneDirectionSelectTests, BeforeInitializationDoesNotPublishSelectState) {
   VerticalVaneDirectionSelectTestContext ctx;
 
-  ctx.select.control(3);
+  ctx.select.make_call().set_index(3).perform();
 
   EXPECT_EQ(ctx.hub.status().vane_mode, MitsubishiCN105::VaneMode::POSITION_3);
   EXPECT_FALSE(ctx.select.has_state());
 }
-
 }  // namespace esphome::mitsubishi_cn105::testing

tests/components/online_image/common.yaml

@@ -57,6 +57,11 @@ image:
     url: http://www.faqs.org/images/library.jpg
     format: JPG
     type: RGB565
+  - platform: online_image
+    id: online_auto_image
+    url: http://www.faqs.org/images/library.jpg
+    format: AUTO
+    type: RGB565
 
 # Check the set_url action
 esphome:

tests/components/runtime_image/test_decoder_reuse.cpp

@@ -77,18 +77,12 @@ class TestableRuntimeImage : public RuntimeImage {
       : RuntimeImage(format, image::IMAGE_TYPE_RGB, image::TRANSPARENCY_OPAQUE, nullptr, false, 0, 0) {}
 
   ImageDecoder *decoder() { return this->decoder_.get(); }
-
-  /// Simulates the state a dynamic-format producer (PR #16337) would leave behind:
-  /// a cached decoder whose format no longer matches the image's format.
-  /// TODO: once #16337 adds a public way to change the format, drive the mismatch
-  /// through it and delete this seam.
-  void plant_decoder(ImageFormat format) { this->decoder_ = this->create_decoder_(format); }
 };
 
 /// Runs one full decode session. Returns true when every stage succeeded.
-static bool decode_all(TestableRuntimeImage &img, const uint8_t *data, size_t len) {
+static bool decode_all(TestableRuntimeImage &img, const uint8_t *data, size_t len, ImageFormat format = AUTO) {
   std::vector<uint8_t> buffer(data, data + len);  // feed_data needs mutable bytes
-  if (!img.begin_decode(len)) {
+  if (!img.begin_decode(len, format)) {
     return false;
   }
   size_t offset = 0;
@@ -203,25 +197,51 @@ TEST(RuntimeImageDecoder, ChunkedFeedDecodesLikeDownloadLoop) {
 }
 
 TEST(RuntimeImageDecoder, FormatSwitchEvictsMismatchedDecoder) {
-  // PNG image holding a stale BMP decoder: begin_decode must evict and recreate.
-  TestableRuntimeImage png_img(PNG);
-  png_img.plant_decoder(BMP);
-  ASSERT_NE(png_img.decoder(), nullptr);
-  ASSERT_EQ(png_img.decoder()->get_format(), BMP);
+  // Drive the format switch through begin_decode()'s format parameter, the way
+  // a dynamic-format producer (online_image MIME detection) does.
+  TestableRuntimeImage img(AUTO);
 
-  ASSERT_TRUE(decode_all(png_img, PNG_RGB, sizeof(PNG_RGB)));
-  EXPECT_EQ(png_img.decoder()->get_format(), PNG);
-  expect_pixels(png_img, PNG_RGB_EXPECTED);
+  ASSERT_TRUE(decode_all(img, BMP_24BPP, sizeof(BMP_24BPP), BMP));
+  ASSERT_NE(img.decoder(), nullptr);
+  ASSERT_EQ(img.decoder()->get_format(), BMP);
+  expect_pixels(img, BMP_24BPP_EXPECTED);
 
-  // And the other direction: BMP image holding a stale PNG decoder.
-  TestableRuntimeImage bmp_img(BMP);
-  bmp_img.plant_decoder(PNG);
-  ASSERT_NE(bmp_img.decoder(), nullptr);
-  ASSERT_EQ(bmp_img.decoder()->get_format(), PNG);
+  // Same explicit format again: the decoder must stay warm.
+  ImageDecoder *bmp_decoder = img.decoder();
+  ASSERT_TRUE(decode_all(img, BMP_8BPP, sizeof(BMP_8BPP), BMP));
+  expect_pixels(img, BMP_8BPP_EXPECTED);
+  EXPECT_EQ(img.decoder(), bmp_decoder);
 
-  ASSERT_TRUE(decode_all(bmp_img, BMP_24BPP, sizeof(BMP_24BPP)));
-  EXPECT_EQ(bmp_img.decoder()->get_format(), BMP);
-  expect_pixels(bmp_img, BMP_24BPP_EXPECTED);
+  // Different format: the stale decoder must be evicted and recreated.
+  ASSERT_TRUE(decode_all(img, PNG_RGB, sizeof(PNG_RGB), PNG));
+  EXPECT_EQ(img.decoder()->get_format(), PNG);
+  expect_pixels(img, PNG_RGB_EXPECTED);
+
+  // And back again.
+  ASSERT_TRUE(decode_all(img, BMP_24BPP, sizeof(BMP_24BPP), BMP));
+  EXPECT_EQ(img.decoder()->get_format(), BMP);
+  expect_pixels(img, BMP_24BPP_EXPECTED);
+}
+
+TEST(RuntimeImageDecoder, AutoFormatFallsBackToConfiguredAndKeepsDecoderWarm) {
+  // With a configured format, an AUTO begin_decode() must resolve to the
+  // configured format before the reuse check instead of evicting the decoder.
+  TestableRuntimeImage img(BMP);
+
+  ASSERT_TRUE(decode_all(img, BMP_24BPP, sizeof(BMP_24BPP), AUTO));
+  ImageDecoder *first = img.decoder();
+  ASSERT_NE(first, nullptr);
+  EXPECT_EQ(first->get_format(), BMP);
+
+  ASSERT_TRUE(decode_all(img, BMP_24BPP, sizeof(BMP_24BPP), AUTO));
+  expect_pixels(img, BMP_24BPP_EXPECTED);
+  EXPECT_EQ(img.decoder(), first) << "AUTO must not evict the configured-format decoder";
+}
+
+TEST(RuntimeImageDecoder, AutoWithoutConfiguredFormatFails) {
+  // Neither a configured format nor an explicit one: there is nothing to decode with.
+  TestableRuntimeImage img(AUTO);
+  EXPECT_FALSE(img.begin_decode(64));
 }
 
 TEST(RuntimeImageDecoder, ReleaseKeepsDecoderWarm) {

tests/integration/README.md

@@ -7,6 +7,7 @@ This directory contains end-to-end integration tests for ESPHome, focusing on te
 - `conftest.py` - Common fixtures and utilities
 - `const.py` - Constants used throughout the integration tests
 - `types.py` - Type definitions for fixtures and functions
+- `raw_api_client.py` - Minimal plaintext api client whose reads happen only on request (for backpressure tests)
 - `state_utils.py` - State handling utilities (e.g., `InitialStateHelper`, `find_entity`, `require_entity`)
 - `fixtures/` - YAML configuration files for tests
 - `test_*.py` - Individual test files
@@ -347,6 +348,7 @@ Create C++ components in `fixtures/external_components/` for:
 - Custom entity behaviors
 - Scheduler testing
 - Memory management tests
+- Deterministic network backpressure (`sndbuf_pin_component` pins socket send buffers; assert on its log line to prove the pin took effect)
 
 ##### Log Line Monitoring
 ```python

tests/integration/fixtures/api_list_entities_backpressure.yaml

@@ -0,0 +1,23 @@
+esphome:
+  name: api-backpressure-test
+
+host:
+
+api:
+  # Smallest queue so a non-draining client blocks the send path quickly
+  max_send_queue: 1
+  actions:
+# GENERATED_ACTIONS
+
+external_components:
+  - source:
+      type: local
+      path: EXTERNAL_COMPONENT_PATH
+    components: [sndbuf_pin_component]
+
+# Pins the device's socket send buffers for deterministic TCP backpressure
+sndbuf_pin_component:
+  buffer_size: SERVER_SNDBUF
+
+logger:
+  level: DEBUG

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

@@ -0,0 +1,20 @@
+import esphome.codegen as cg
+import esphome.config_validation as cv
+from esphome.const import CONF_BUFFER_SIZE, CONF_ID
+
+DEPENDENCIES = ["api"]
+
+sndbuf_pin_ns = cg.esphome_ns.namespace("sndbuf_pin")
+SndbufPinComponent = sndbuf_pin_ns.class_("SndbufPinComponent", cg.Component)
+
+CONFIG_SCHEMA = cv.Schema(
+    {
+        cv.GenerateID(): cv.declare_id(SndbufPinComponent),
+        cv.Required(CONF_BUFFER_SIZE): cv.int_range(min=1),
+    }
+).extend(cv.COMPONENT_SCHEMA)
+
+
+async def to_code(config):
+    var = cg.new_Pvariable(config[CONF_ID], config[CONF_BUFFER_SIZE])
+    await cg.register_component(var, config)

tests/integration/fixtures/external_components/sndbuf_pin_component/sndbuf_pin_component.cpp

@@ -0,0 +1,55 @@
+#include "sndbuf_pin_component.h"
+
+#include <netinet/in.h>
+#include <sys/socket.h>
+#include <cerrno>
+
+#include "esphome/components/api/api_server.h"
+#include "esphome/core/log.h"
+
+namespace esphome::sndbuf_pin {
+
+static const char *const TAG = "sndbuf_pin";
+
+// Skip stdio; scan the low fd range where the listeners land
+static constexpr int FIRST_USER_FD = 3;
+static constexpr int MAX_FD_SCAN = 128;
+
+void SndbufPinComponent::setup() {
+  int pinned = 0;
+  for (int fd = FIRST_USER_FD; fd < MAX_FD_SCAN; fd++) {
+    int type = 0;
+    socklen_t len = sizeof(type);
+    if (::getsockopt(fd, SOL_SOCKET, SO_TYPE, &type, &len) != 0 || type != SOCK_STREAM)
+      continue;
+    struct sockaddr_in addr {};
+    socklen_t addr_len = sizeof(addr);
+    if (::getsockname(fd, reinterpret_cast<struct sockaddr *>(&addr), &addr_len) != 0) {
+      ESP_LOGW(TAG, "fd %d: getsockname failed, errno %d", fd, errno);
+      continue;
+    }
+    if (ntohs(addr.sin_port) != api::global_api_server->get_port())
+      continue;
+    if (::setsockopt(fd, SOL_SOCKET, SO_SNDBUF, &this->buffer_size_, sizeof(this->buffer_size_)) != 0) {
+      ESP_LOGW(TAG, "fd %d: SO_SNDBUF pin failed, errno %d", fd, errno);
+      continue;
+    }
+    int applied = 0;
+    len = sizeof(applied);
+    if (::getsockopt(fd, SOL_SOCKET, SO_SNDBUF, &applied, &len) != 0 || applied < this->buffer_size_) {
+      // Linux doubles the requested value; anything below it means clamped
+      ESP_LOGW(TAG, "fd %d: SO_SNDBUF readback %d below requested %d", fd, applied, this->buffer_size_);
+      continue;
+    }
+    // Tests assert on this line; accepted sockets inherit the pinned size
+    ESP_LOGD(TAG, "fd %d port %d: SO_SNDBUF pinned to %d (effective %d)", fd, ntohs(addr.sin_port), this->buffer_size_,
+             applied);
+    pinned++;
+  }
+  if (pinned == 0) {
+    ESP_LOGE(TAG, "api listener socket was not pinned");
+    this->mark_failed();
+  }
+}
+
+}  // namespace esphome::sndbuf_pin

tests/integration/fixtures/external_components/sndbuf_pin_component/sndbuf_pin_component.h

@@ -0,0 +1,21 @@
+#pragma once
+
+#include "esphome/core/component.h"
+
+namespace esphome::sndbuf_pin {
+
+// Test-only (host): pins SO_SNDBUF on every open TCP socket so integration
+// tests get deterministic backpressure; an explicit SO_SNDBUF also disables
+// kernel autotuning, and accepted sockets inherit it from the listener.
+class SndbufPinComponent : public Component {
+ public:
+  explicit SndbufPinComponent(int buffer_size) : buffer_size_(buffer_size) {}
+  void setup() override;
+  // After the api server so its listening socket exists
+  float get_setup_priority() const override { return setup_priority::LATE; }
+
+ protected:
+  int buffer_size_;
+};
+
+}  // namespace esphome::sndbuf_pin

tests/integration/fixtures/online_image_auto_detects_image_bmp_mime.yaml

@@ -0,0 +1,28 @@
+esphome:
+  name: online-image-bmp
+
+host:
+
+http_request:
+
+display:
+
+image:
+  - platform: online_image
+    url: http://127.0.0.1:HTTP_PORT/foo.bmp
+    format: AUTO
+    id: myimg
+    type: RGB
+    on_download_finished:
+      logger.log:
+        format: "download finished. cache hit: %u"
+        args: [cached]
+
+api:
+  actions:
+    - action: fetch_image
+      then:
+        - component.update: myimg
+
+logger:
+  level: DEBUG

tests/integration/fixtures/online_image_auto_detects_redirected_image_bmp_mime.yaml

@@ -0,0 +1,28 @@
+esphome:
+  name: online-image-bmp
+
+host:
+
+http_request:
+
+display:
+
+image:
+  - platform: online_image
+    url: http://127.0.0.1:HTTP_PORT/foo.bmp
+    id: myimg
+    format: AUTO
+    type: RGB
+    on_download_finished:
+      logger.log:
+        format: "download finished. cache hit: %u"
+        args: [cached]
+
+api:
+  actions:
+    - action: fetch_image
+      then:
+        - component.update: myimg
+
+logger:
+  level: DEBUG

tests/integration/fixtures/online_image_bmp.yaml

@@ -7,8 +7,9 @@ http_request:
 
 display:
 
-online_image:
-  - url: http://127.0.0.1:HTTP_PORT/foo.bmp
+image:
+  - platform: online_image
+    url: http://127.0.0.1:HTTP_PORT/foo.bmp
     id: myimg
     format: BMP
     type: RGB

tests/integration/online_image_utils.py

@@ -0,0 +1,158 @@
+"""Shared fixture server and log helpers for the online_image integration tests."""
+
+from __future__ import annotations
+
+import asyncio
+from collections.abc import Callable
+import re
+
+# black 8x8 RGB BMP, generated with
+#   from PIL import Image
+#   from io import BytesIO
+#   b = BytesIO()
+#   img = Image.new("RGB", (8, 8))
+#   img.save(b, format="BMP")
+#   b.getvalue()
+BMP_IMAGE = b"BM\xf6\x00\x00\x00\x00\x00\x00\x006\x00\x00\x00(\x00\x00\x00\x08\x00\x00\x00\x08\x00\x00\x00\x01\x00\x18\x00\x00\x00\x00\x00\xc0\x00\x00\x00\xc4\x0e\x00\x00\xc4\x0e\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00"
+LEN_BMP_IMAGE = len(BMP_IMAGE)
+
+
+async def wait_for_download(
+    downloaded_bytes_future: asyncio.Future,
+    server_error_future: asyncio.Future,
+) -> int:
+    """Await the downloaded byte count, raising a server handler error first."""
+    await asyncio.wait(
+        {downloaded_bytes_future, server_error_future},
+        return_when=asyncio.FIRST_COMPLETED,
+    )
+    if server_error_future.done() and (exc := server_error_future.exception()):
+        raise exc
+    # Retrieve a late teardown error so asyncio does not log it at GC
+    server_error_future.add_done_callback(lambda f: f.exception())
+    return downloaded_bytes_future.result()
+
+
+def make_download_watcher(
+    downloaded_bytes_future: asyncio.Future,
+    download_finished_future: asyncio.Future,
+) -> Callable[[str], None]:
+    """Build a line callback resolving the futures from the device log."""
+
+    def check_output(line: str) -> None:
+        if (
+            match := re.search(r"Image fully downloaded, (\d+) bytes", line)
+        ) and not downloaded_bytes_future.done():
+            downloaded_bytes_future.set_result(int(match.group(1)))
+        if "download finished" in line and not download_finished_future.done():
+            download_finished_future.set_result(True)
+
+    return check_output
+
+
+def handle_http(
+    http_request_future,
+    content_type: str = "text/plain",
+    *,
+    request_path: str = "/foo.bmp",
+    request_line_consumed: bool = False,
+    server_error_future: asyncio.Future | None = None,
+):
+    async def handler(reader, writer):
+        try:
+            # Only read the request line if it hasn't been consumed by a caller
+            if not request_line_consumed:
+                async with asyncio.timeout(1.0):
+                    data = await reader.readuntil(b"\r\n")
+
+                expected_request = f"GET {request_path} HTTP/1.1\r\n".encode()
+                assert data[: len(expected_request)] == expected_request
+
+            async with asyncio.timeout(1.0):
+                await reader.readuntil(b"\r\n\r\n")
+
+            if not http_request_future.done():
+                http_request_future.set_result(True)
+
+            http_response = [
+                b"HTTP/1.1 200 OK",
+                b"Content-Length: %d" % LEN_BMP_IMAGE,
+                f"Content-Type: {content_type}".encode(),
+                b"Connection: close",
+                b"",
+                b"",
+            ]
+            writer.write(b"\r\n".join(http_response))
+            await writer.drain()
+
+            writer.write(BMP_IMAGE)
+
+            await writer.drain()
+        except Exception as exc:
+            if server_error_future is not None and not server_error_future.done():
+                server_error_future.set_exception(exc)
+            if not http_request_future.done():
+                http_request_future.set_exception(exc)
+            raise
+        finally:
+            writer.close()
+
+    return handler
+
+
+def handle_http_redirect(
+    http_request_future, final_request_future, server_error_future, port_holder
+):
+    async def handler(reader, writer):
+        try:
+            async with asyncio.timeout(1.0):
+                request = await reader.readuntil(b"\r\n")
+
+            if (
+                request[: len(b"GET /foo.bmp HTTP/1.1\r\n")]
+                == b"GET /foo.bmp HTTP/1.1\r\n"
+            ):
+                if not http_request_future.done():
+                    http_request_future.set_result(True)
+                async with asyncio.timeout(1.0):
+                    await reader.readuntil(b"\r\n\r\n")
+
+                http_response = [
+                    b"HTTP/1.1 302 Found",
+                    f"Location: http://127.0.0.1:{port_holder['port']}/final.bmp".encode(),
+                    b"Content-Type: text/html",
+                    b"Content-Length: 0",
+                    b"Connection: close",
+                    b"",
+                    b"",
+                ]
+                writer.write(b"\r\n".join(http_response))
+                await writer.drain()
+                return
+
+            assert (
+                request[: len(b"GET /final.bmp HTTP/1.1\r\n")]
+                == b"GET /final.bmp HTTP/1.1\r\n"
+            )
+            if not final_request_future.done():
+                final_request_future.set_result(True)
+            await handle_http(
+                final_request_future,
+                "image/bmp",
+                request_path="/final.bmp",
+                request_line_consumed=True,
+                server_error_future=server_error_future,
+            )(reader, writer)
+        except Exception as exc:
+            # Route handler failures to the dedicated error future so they're not silently lost
+            if not server_error_future.done():
+                server_error_future.set_exception(exc)
+            if not http_request_future.done():
+                http_request_future.set_exception(exc)
+            if not final_request_future.done():
+                final_request_future.set_exception(exc)
+            raise
+        finally:
+            writer.close()
+
+    return handler

tests/integration/raw_api_client.py

@@ -0,0 +1,148 @@
+"""Minimal plaintext native-api client over a raw socket.
+
+Reads only when told to, so tests control when the TCP pipe backs up toward
+the device; payloads are skipped and only message types are counted.
+"""
+
+from __future__ import annotations
+
+import asyncio
+from collections import Counter
+import socket
+from typing import Self
+
+from aioesphomeapi import api_pb2
+import aioesphomeapi.core as api_core
+from google.protobuf import message
+
+from .const import LOCALHOST
+
+# Message type ids are protocol constants; derive them from aioesphomeapi so
+# they cannot drift from the client library in use.
+MESSAGE_TYPE_OF = {cls: num for num, cls in api_core.MESSAGE_TYPE_TO_PROTO.items()}
+
+_READ_CHUNK = 4096
+
+
+def encode_varint(value: int) -> bytes:
+    out = bytearray()
+    while True:
+        byte = value & 0x7F
+        value >>= 7
+        if value:
+            out.append(byte | 0x80)
+        else:
+            out.append(byte)
+            return bytes(out)
+
+
+def decode_varint(buf: bytearray, pos: int) -> tuple[int, int] | None:
+    """Decode one varint at pos; return (value, new_pos) or None if short."""
+    value = shift = 0
+    while pos < len(buf):
+        byte = buf[pos]
+        pos += 1
+        value |= (byte & 0x7F) << shift
+        if not byte & 0x80:
+            return value, pos
+        shift += 7
+    return None
+
+
+def encode_frame(msg_type: int, payload: bytes) -> bytes:
+    """Encode one plaintext api frame: 0x00, payload length, message type."""
+    return b"\x00" + encode_varint(len(payload)) + encode_varint(msg_type) + payload
+
+
+class FrameParser:
+    """Incremental parser for the plaintext api frame stream."""
+
+    def __init__(self) -> None:
+        self._buf = bytearray()
+
+    def feed(self, data: bytes) -> list[int]:
+        self._buf.extend(data)
+        types: list[int] = []
+        while (msg_type := self._try_parse()) is not None:
+            types.append(msg_type)
+        return types
+
+    def _try_parse(self) -> int | None:
+        buf = self._buf
+        if not buf:
+            return None
+        assert buf[0] == 0, f"expected plaintext frame, got indicator {buf[0]}"
+        if (size_decoded := decode_varint(buf, 1)) is None:
+            return None
+        size, pos = size_decoded
+        if (type_decoded := decode_varint(buf, pos)) is None:
+            return None
+        msg_type, pos = type_decoded
+        if len(buf) - pos < size:
+            return None
+        del buf[: pos + size]
+        return msg_type
+
+
+class RawApiClient:
+    """Plaintext api client whose reads happen only on request."""
+
+    def __init__(self, port: int, recv_buffer_size: int | None = None) -> None:
+        self._port = port
+        self._parser = FrameParser()
+        self.bytes_received = 0
+        self.frame_counts: Counter[int] = Counter()
+        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+        try:
+            if recv_buffer_size is not None:
+                sock.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, recv_buffer_size)
+                # Kernels may round up (Linux doubles) but must not clamp below
+                applied = sock.getsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF)
+                assert applied >= recv_buffer_size, (
+                    f"SO_RCVBUF clamped to {applied}, requested {recv_buffer_size}"
+                )
+            sock.setblocking(False)
+        except Exception:
+            sock.close()
+            raise
+        self._sock = sock
+
+    async def __aenter__(self) -> Self:
+        return self
+
+    async def __aexit__(self, *exc_info: object) -> None:
+        self.close()
+
+    async def connect(self, client_info: str = "raw-api-client") -> None:
+        """Connect and complete the Hello handshake (no auth step since 2026.1.0)."""
+        loop = asyncio.get_running_loop()
+        await loop.sock_connect(self._sock, (LOCALHOST, self._port))
+        hello = api_pb2.HelloRequest()
+        hello.client_info = client_info
+        hello.api_version_major = 1
+        hello.api_version_minor = 10
+        await self.send_message(hello)
+        await self.read_until_frame(MESSAGE_TYPE_OF[api_pb2.HelloResponse])
+
+    async def send_message(self, msg: message.Message) -> None:
+        loop = asyncio.get_running_loop()
+        await loop.sock_sendall(
+            self._sock,
+            encode_frame(MESSAGE_TYPE_OF[type(msg)], msg.SerializeToString()),
+        )
+
+    async def read_until_frame(self, msg_type: int, timeout: float = 10.0) -> None:
+        """Read until at least one frame of msg_type has been received."""
+        loop = asyncio.get_running_loop()
+
+        async def _read_loop() -> None:
+            while not self.frame_counts[msg_type]:
+                data = await loop.sock_recv(self._sock, _READ_CHUNK)
+                assert data, "server closed the connection unexpectedly"
+                self.bytes_received += len(data)
+                self.frame_counts.update(self._parser.feed(data))
+
+        await asyncio.wait_for(_read_loop(), timeout)
+
+    def close(self) -> None:
+        self._sock.close()

tests/integration/test_api_list_entities_backpressure.py

@@ -0,0 +1,110 @@
+"""A client that stops reading the entity listing must not starve other clients.
+
+Service responses are sent directly (not via the deferred batch), so a full
+TCP pipe makes the send path refuse; the drive loop now lives in
+try_advance(), which stops on refusal instead of retrying forever. Not a
+before/after regression test: pre-fix builds survive here because the
+refusal path yields and pumps the socket each retry.
+
+The sndbuf_pin_component fixture pins the device's send buffers so the pipe
+fills deterministically regardless of kernel autotuning; the test waits for
+its log line before proceeding.
+"""
+
+from __future__ import annotations
+
+import asyncio
+
+from aioesphomeapi import api_pb2
+import pytest
+
+from .raw_api_client import MESSAGE_TYPE_OF, RawApiClient
+from .types import APIClientConnectedFactory, RunCompiledFunction
+
+SERVICES_RESPONSE = MESSAGE_TYPE_OF[api_pb2.ListEntitiesServicesResponse]
+LIST_DONE_RESPONSE = MESSAGE_TYPE_OF[api_pb2.ListEntitiesDoneResponse]
+
+# Both ends of the pipe are pinned small; only tens of KB fit in the kernel
+RECV_BUFFER_SIZE = 4096
+SERVER_SNDBUF = 8192  # substituted into the fixture yaml
+# Logged by the sndbuf_pin_component fixture when it pins a socket
+SNDBUF_PIN_LOG = "SO_SNDBUF pinned to"
+# One response (~6.4 KB) must stay smaller than the pinned send buffer; an
+# oversized message parks in the overflow buffer and reports as sent.
+ARGS_PER_SERVICE = 8
+ARG_NAME_LEN = 800
+# ~160 KB listing versus a tens-of-KB pipe guarantees a mid-services block
+NUM_SERVICES = 25
+assert ARGS_PER_SERVICE * ARG_NAME_LEN < SERVER_SNDBUF
+# The pipe fills in well under a second
+STALL_SECONDS = 0.5
+# Well above pipe capacity, well below the listing size
+MIN_DRAINED_BYTES = 60_000
+
+
+def _generated_actions() -> str:
+    """Build the api actions block: services with long argument names."""
+    lines: list[str] = []
+    for i in range(NUM_SERVICES):
+        lines.append(f"    - action: backpressure_service_{i:04d}")
+        lines.append("      variables:")
+        for j in range(ARGS_PER_SERVICE):
+            prefix = f"arg_{i:04d}_{j:02d}_"
+            lines.append(
+                f"        {prefix}{'x' * (ARG_NAME_LEN - len(prefix))}: string"
+            )
+        lines.append("      then:")
+        lines.append("        - logger.log: service called")
+    return "\n".join(lines)
+
+
+@pytest.mark.asyncio
+async def test_api_list_entities_backpressure(
+    yaml_config: str,
+    run_compiled: RunCompiledFunction,
+    api_client_connected: APIClientConnectedFactory,
+    unused_tcp_port: int,
+) -> None:
+    """A stalled reader mid-services must not block other api clients."""
+    assert "# GENERATED_ACTIONS" in yaml_config
+    config = yaml_config.replace("# GENERATED_ACTIONS", _generated_actions())
+    config = config.replace("SERVER_SNDBUF", str(SERVER_SNDBUF))
+
+    pin_applied = asyncio.Event()
+
+    def _on_log_line(line: str) -> None:
+        if SNDBUF_PIN_LOG in line:
+            pin_applied.set()
+
+    async with run_compiled(config, line_callback=_on_log_line):
+        # Fails loudly if the pin never applied
+        await asyncio.wait_for(pin_applied.wait(), 10)
+
+        async with RawApiClient(
+            unused_tcp_port, recv_buffer_size=RECV_BUFFER_SIZE
+        ) as stalled:
+            await stalled.connect(client_info="backpressure-stall-client")
+            await stalled.send_message(api_pb2.ListEntitiesRequest())
+            # The client now stops reading entirely.
+
+            # Let the server run against the full pipe
+            await asyncio.sleep(STALL_SECONDS)
+
+            # Other clients must still be served while the first is blocked
+            async with api_client_connected(timeout=20) as client:
+                device_info = await asyncio.wait_for(client.device_info(), 20)
+                assert device_info.name == "api-backpressure-test"
+                _, services = await asyncio.wait_for(
+                    client.list_entities_services(), 30
+                )
+                assert len(services) == NUM_SERVICES
+
+            # Fixture-size guard: the listing must dwarf the pinned pipe
+            before = stalled.bytes_received
+            await stalled.read_until_frame(LIST_DONE_RESPONSE, timeout=60)
+            drained = stalled.bytes_received - before
+            assert drained > MIN_DRAINED_BYTES, (
+                f"only {drained} bytes drained; the listing never backed up"
+            )
+            assert stalled.frame_counts[SERVICES_RESPONSE] == NUM_SERVICES
+            assert stalled.frame_counts[LIST_DONE_RESPONSE] == 1

tests/integration/test_online_image_auto_detects_image_bmp_mime.py

@@ -0,0 +1,71 @@
+"""Test that online_image AUTO format detection reads the Content-Type header."""
+
+from __future__ import annotations
+
+import asyncio
+
+import pytest
+
+from .online_image_utils import (
+    LEN_BMP_IMAGE,
+    handle_http,
+    make_download_watcher,
+    wait_for_download,
+)
+from .types import APIClientConnectedFactory, RunCompiledFunction
+
+
+@pytest.mark.asyncio
+async def test_online_image_auto_detects_image_bmp_mime(
+    yaml_config: str,
+    run_compiled: RunCompiledFunction,
+    api_client_connected: APIClientConnectedFactory,
+) -> None:
+    """AUTO format detection should honor the final response MIME type without explicit format."""
+    loop = asyncio.get_running_loop()
+    http_request_future = loop.create_future()
+    server_error_future = loop.create_future()
+    download_finished_future = loop.create_future()
+    downloaded_bytes_future = loop.create_future()
+
+    check_output = make_download_watcher(
+        downloaded_bytes_future, download_finished_future
+    )
+
+    server = await asyncio.start_server(
+        handle_http(
+            http_request_future,
+            "image/bmp",
+            server_error_future=server_error_future,
+        ),
+        "127.0.0.1",
+        0,
+    )
+    http_server_port = server.sockets[0].getsockname()[1]
+
+    config = yaml_config.replace("HTTP_PORT", str(http_server_port))
+
+    async with (
+        server,
+        run_compiled(config, line_callback=check_output),
+        api_client_connected() as client,
+    ):
+        device_info = await client.device_info()
+        assert device_info is not None
+        assert device_info.name == "online-image-bmp"
+
+        _, services = await client.list_entities_services()
+        request_service = next((s for s in services if s.name == "fetch_image"), None)
+        assert request_service is not None
+
+        await client.execute_service(request_service, {})
+
+        async with asyncio.timeout(0.1):
+            await http_request_future
+
+        async with asyncio.timeout(0.5):
+            numbytes = await wait_for_download(
+                downloaded_bytes_future, server_error_future
+            )
+            assert numbytes == LEN_BMP_IMAGE
+            await download_finished_future

tests/integration/test_online_image_auto_detects_redirected_image_bmp_mime.py

@@ -0,0 +1,71 @@
+"""Test that AUTO format detection uses the final Content-Type after redirects."""
+
+from __future__ import annotations
+
+import asyncio
+
+import pytest
+
+from .online_image_utils import (
+    LEN_BMP_IMAGE,
+    handle_http_redirect,
+    make_download_watcher,
+    wait_for_download,
+)
+from .types import APIClientConnectedFactory, RunCompiledFunction
+
+
+@pytest.mark.asyncio
+async def test_online_image_auto_detects_redirected_image_bmp_mime(
+    yaml_config: str,
+    run_compiled: RunCompiledFunction,
+    api_client_connected: APIClientConnectedFactory,
+) -> None:
+    """Redirect hops should not leave the 302 HTML Content-Type in place for the final image."""
+    loop = asyncio.get_running_loop()
+    http_request_future = loop.create_future()
+    final_request_future = loop.create_future()
+    server_error_future = loop.create_future()
+    download_finished_future = loop.create_future()
+    downloaded_bytes_future = loop.create_future()
+
+    check_output = make_download_watcher(
+        downloaded_bytes_future, download_finished_future
+    )
+
+    port_holder = {}
+    server = await asyncio.start_server(
+        handle_http_redirect(
+            http_request_future, final_request_future, server_error_future, port_holder
+        ),
+        "127.0.0.1",
+        0,
+    )
+    port_holder["port"] = server.sockets[0].getsockname()[1]
+
+    config = yaml_config.replace("HTTP_PORT", str(port_holder["port"]))
+
+    async with (
+        server,
+        run_compiled(config, line_callback=check_output),
+        api_client_connected() as client,
+    ):
+        device_info = await client.device_info()
+        assert device_info is not None
+        assert device_info.name == "online-image-bmp"
+
+        _, services = await client.list_entities_services()
+        request_service = next((s for s in services if s.name == "fetch_image"), None)
+        assert request_service is not None
+
+        await client.execute_service(request_service, {})
+
+        async with asyncio.timeout(0.1):
+            await http_request_future
+        async with asyncio.timeout(0.5):
+            await final_request_future
+            numbytes = await wait_for_download(
+                downloaded_bytes_future, server_error_future
+            )
+            assert numbytes == LEN_BMP_IMAGE
+            await download_finished_future

tests/integration/test_online_image_bmp.py

@@ -1,62 +1,12 @@
 from __future__ import annotations
 
 import asyncio
-import re
 
 import pytest
 
+from .online_image_utils import LEN_BMP_IMAGE, handle_http, make_download_watcher
 from .types import APIClientConnectedFactory, RunCompiledFunction
 
-# black 8x8 RGB BMP, generated with
-#   from PIL import Image
-#   from io import BytesIO
-#   b = BytesIO()
-#   img = Image.new("RGB", (8, 8))
-#   img.save(b, format="BMP")
-#   b.getvalue()
-BMP_IMAGE = b"BM\xf6\x00\x00\x00\x00\x00\x00\x006\x00\x00\x00(\x00\x00\x00\x08\x00\x00\x00\x08\x00\x00\x00\x01\x00\x18\x00\x00\x00\x00\x00\xc0\x00\x00\x00\xc4\x0e\x00\x00\xc4\x0e\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00"
-LEN_BMP_IMAGE = len(BMP_IMAGE)
-
-
-def handle_http(http_request_future):
-    async def handler(reader, writer):
-        try:
-            async with asyncio.timeout(1.0):
-                data = await reader.readuntil(b"\r\n")
-
-            # ensure our request matches the expectation
-            expected_request = b"GET /foo.bmp HTTP/1.1\r\n"
-            assert data[: len(expected_request)] == expected_request
-
-            # consume rest of request
-            async with asyncio.timeout(1.0):
-                data = await reader.readuntil(b"\r\n\r\n")
-
-            http_request_future.set_result(True)
-
-            http_response = [
-                b"HTTP/1.1 200 OK",
-                b"Content-Length: %d" % LEN_BMP_IMAGE,
-                b"Content-Type: text/plain",
-                b"Connection: close",
-                b"",
-                b"",
-            ]
-            writer.write(b"\r\n".join(http_response))
-            await writer.drain()
-
-            writer.write(BMP_IMAGE)
-
-            await writer.drain()
-        except Exception as exc:
-            if not http_request_future.done():
-                http_request_future.set_exception(exc)
-            raise
-        finally:
-            writer.close()
-
-    return handler
-
 
 @pytest.mark.asyncio
 async def test_online_image_bmp(
@@ -72,14 +22,9 @@ async def test_online_image_bmp(
     download_finished_future = loop.create_future()
     downloaded_bytes_future = loop.create_future()
 
-    def check_output(line: str) -> None:
-        """Check log output for expected messages."""
-
-        if match := re.search(r"Image fully downloaded, (\d+) bytes", line):
-            downloaded_bytes_future.set_result(int(match.group(1)))
-
-        if "download finished" in line:
-            download_finished_future.set_result(True)
+    check_output = make_download_watcher(
+        downloaded_bytes_future, download_finished_future
+    )
 
     server = await asyncio.start_server(
         handle_http(http_request_future), "127.0.0.1", 0

tests/unit_tests/test_external_files.py

@@ -4,7 +4,7 @@ import os
 from pathlib import Path
 import time
 from typing import Any
-from unittest.mock import MagicMock, patch
+from unittest.mock import MagicMock, call, patch
 
 import pytest
 import requests
@@ -81,6 +81,15 @@ def mock_download_content_many() -> MagicMock:
         yield m
 
 
+@pytest.fixture
+def mock_retry_sleep() -> MagicMock:
+    """Patch the retry backoff sleep (process-wide; net_retry.time is the
+    global module) so transient-error tests don't really wait 2s/4s.
+    """
+    with patch("esphome.net_retry.time.sleep") as m:
+        yield m
+
+
 def test_compute_local_file_dir(setup_core: Path) -> None:
     """Test compute_local_file_dir creates and returns correct path."""
     domain = "font"
@@ -495,6 +504,7 @@ class _BodyReadErrorResponse:
 def test_download_content_with_body_read_error_uses_cache(
     mock_has_remote_file_changed: MagicMock,
     mock_requests_get: MagicMock,
+    mock_retry_sleep: MagicMock,
     setup_core: Path,
 ) -> None:
     """Body-read errors (chunked-decode/gzip-decode/mid-stream connection
@@ -519,6 +529,7 @@ def test_download_content_with_body_read_error_uses_cache(
 def test_download_content_with_body_read_error_no_cache_fails(
     mock_has_remote_file_changed: MagicMock,
     mock_requests_get: MagicMock,
+    mock_retry_sleep: MagicMock,
     setup_core: Path,
 ) -> None:
     """A body-read failure with no cache available must surface as a
@@ -535,6 +546,131 @@ def test_download_content_with_body_read_error_no_cache_fails(
         external_files.download_content("https://example.com/file.txt", test_file)
 
 
+def test_download_content_retries_transient_error_then_succeeds(
+    mock_has_remote_file_changed: MagicMock,
+    mock_requests_get: MagicMock,
+    mock_retry_sleep: MagicMock,
+    setup_core: Path,
+) -> None:
+    """Transient failures (connection reset, timeout) are retried with 2s/4s
+    backoff before giving up; a late success downloads normally."""
+    test_file = setup_core / "downloads" / "file.txt"
+    mock_has_remote_file_changed.return_value = True
+
+    ok = MagicMock()
+    ok.content = b"downloaded"
+    ok.headers = {}
+    mock_requests_get.side_effect = [
+        requests.exceptions.ConnectionError("reset by peer"),
+        requests.exceptions.Timeout("timed out"),
+        ok,
+    ]
+
+    result = external_files.download_content("https://example.com/file.txt", test_file)
+
+    assert result == b"downloaded"
+    assert test_file.read_bytes() == b"downloaded"
+    assert mock_retry_sleep.call_args_list == [call(2), call(4)]
+
+
+def test_download_content_transient_error_exhausts_attempts(
+    mock_has_remote_file_changed: MagicMock,
+    mock_requests_get: MagicMock,
+    mock_retry_sleep: MagicMock,
+    setup_core: Path,
+) -> None:
+    """A persistent transient failure gives up after three attempts and then
+    follows the normal no-cache error path."""
+    test_file = setup_core / "nonexistent.txt"
+    mock_has_remote_file_changed.return_value = True
+    mock_requests_get.side_effect = requests.exceptions.ConnectionError("reset by peer")
+
+    with pytest.raises(Invalid, match="Could not download from.*reset by peer"):
+        external_files.download_content("https://example.com/file.txt", test_file)
+
+    assert mock_retry_sleep.call_args_list == [call(2), call(4)]
+
+
+def test_download_content_non_transient_error_not_retried(
+    mock_has_remote_file_changed: MagicMock,
+    mock_requests_get: MagicMock,
+    mock_retry_sleep: MagicMock,
+    setup_core: Path,
+) -> None:
+    """Permanent failures like a 404 fail on the first attempt."""
+    test_file = setup_core / "nonexistent.txt"
+    mock_has_remote_file_changed.return_value = True
+
+    response = MagicMock()
+    response.status_code = 404
+    mock_requests_get.side_effect = requests.exceptions.HTTPError(
+        "404 Client Error", response=response
+    )
+
+    with pytest.raises(Invalid, match="Could not download from.*404"):
+        external_files.download_content("https://example.com/file.txt", test_file)
+
+    assert mock_requests_get.call_count == 1
+    mock_retry_sleep.assert_not_called()
+
+
+def test_download_content_retries_body_read_error(
+    mock_has_remote_file_changed: MagicMock,
+    mock_requests_get: MagicMock,
+    mock_retry_sleep: MagicMock,
+    setup_core: Path,
+) -> None:
+    """Mid-stream failures surfacing from `.content` are retried too."""
+    test_file = setup_core / "downloads" / "file.txt"
+    mock_has_remote_file_changed.return_value = True
+
+    ok = MagicMock()
+    ok.content = b"downloaded"
+    ok.headers = {}
+    mock_requests_get.side_effect = [
+        _BodyReadErrorResponse(
+            requests.exceptions.ChunkedEncodingError("body truncated")
+        ),
+        ok,
+    ]
+
+    result = external_files.download_content("https://example.com/file.txt", test_file)
+
+    assert result == b"downloaded"
+    assert mock_requests_get.call_count == 2
+    assert mock_retry_sleep.call_args_list == [call(2)]
+
+
+def test_has_remote_file_changed_retries_transient_error(
+    mock_requests_head: MagicMock,
+    mock_retry_sleep: MagicMock,
+    setup_core: Path,
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """A HEAD revalidation that fails transiently then returns 304 does not
+    mark the cached copy stale, and the retry warning names the operation."""
+    test_file = setup_core / "cached.txt"
+    test_file.write_bytes(b"cached content")
+
+    ok = MagicMock()
+    ok.status_code = 304
+    ok.headers = {}
+    mock_requests_head.side_effect = [
+        requests.exceptions.ConnectionError("reset by peer"),
+        ok,
+    ]
+
+    changed = external_files.has_remote_file_changed(
+        "https://example.com/file.txt", test_file
+    )
+
+    assert changed is False
+    assert test_file not in external_files._run_data().stale_paths
+    assert mock_requests_head.call_count == 2
+    assert mock_retry_sleep.call_args_list == [call(2)]
+    assert "Revalidation of" in caplog.text
+
+
 def test_download_content_skip_external_update_uses_cache(
     mock_has_remote_file_changed: MagicMock,
     mock_requests_get: MagicMock,

tests/unit_tests/test_framework_helpers.py

@@ -26,7 +26,6 @@ from esphome.framework_helpers import (
     _7z_extract_all,
     _BatchDownloadProgress,
     _detect_archive_root,
-    _is_transient_download_error,
     _rename_with_retry,
     _tar_extract_all,
     _zip_extract_all,
@@ -1849,43 +1848,6 @@ class TestDownloadFromMirrors:
         mock_sleep.assert_not_called()
 
 
-def _http_error(status: int) -> req.HTTPError:
-    """An HTTPError carrying a response with the given status, as raised by
-    ``raise_for_status`` on a real response."""
-    resp = MagicMock()
-    resp.status_code = status
-    return req.HTTPError(str(status), response=resp)
-
-
-class TestIsTransientDownloadError:
-    def test_connection_errors_are_transient(self) -> None:
-        assert _is_transient_download_error(req.ConnectionError("reset"))
-        assert _is_transient_download_error(req.Timeout("timed out"))
-        assert _is_transient_download_error(
-            req.exceptions.ChunkedEncodingError("dropped")
-        )
-
-    def test_http_statuses(self) -> None:
-        assert not _is_transient_download_error(_http_error(404))
-        assert not _is_transient_download_error(_http_error(403))
-        assert _is_transient_download_error(_http_error(429))
-        assert _is_transient_download_error(_http_error(503))
-
-    def test_http_error_without_response_is_permanent(self) -> None:
-        assert not _is_transient_download_error(req.HTTPError("boom"))
-
-    def test_exhausted_resume_attempts_are_permanent(self) -> None:
-        """download_with_resume already spent its own resume attempts; its
-        EsphomeError wrapper is not retried again at the sweep level."""
-        wrapped = EsphomeError("Failed to download after 3 attempts")
-        wrapped.__cause__ = req.ConnectionError("down")
-        assert not _is_transient_download_error(wrapped)
-
-    def test_unrelated_errors_are_permanent(self) -> None:
-        assert not _is_transient_download_error(OSError("disk full"))
-        assert not _is_transient_download_error(EsphomeError("size mismatch"))
-
-
 def test_importing_framework_helpers_does_not_import_requests() -> None:
     """Importing framework_helpers must not drag in requests.
 

tests/unit_tests/test_net_retry.py

@@ -0,0 +1,143 @@
+"""Tests for esphome.net_retry."""
+
+import socket
+from unittest.mock import MagicMock, call, patch
+
+import pytest
+import requests as req
+
+from esphome.core import EsphomeError
+from esphome.net_retry import fetch_with_retry, is_transient_download_error
+
+
+def _http_error(status: int) -> req.HTTPError:
+    """An HTTPError carrying a response with the given status, as raised by
+    ``raise_for_status`` on a real response."""
+    resp = MagicMock()
+    resp.status_code = status
+    return req.HTTPError(str(status), response=resp)
+
+
+class TestIsTransientDownloadError:
+    def test_connection_errors_are_transient(self) -> None:
+        assert is_transient_download_error(req.ConnectionError("reset"))
+        assert is_transient_download_error(req.Timeout("timed out"))
+        assert is_transient_download_error(
+            req.exceptions.ChunkedEncodingError("dropped")
+        )
+        assert is_transient_download_error(
+            req.exceptions.ContentDecodingError("gzip stream truncated")
+        )
+
+    def test_http_statuses(self) -> None:
+        assert not is_transient_download_error(_http_error(404))
+        assert not is_transient_download_error(_http_error(403))
+        assert is_transient_download_error(_http_error(429))
+        assert is_transient_download_error(_http_error(503))
+
+    def test_http_error_without_response_is_permanent(self) -> None:
+        assert not is_transient_download_error(req.HTTPError("boom"))
+
+    def test_hard_dns_failures_are_permanent(self) -> None:
+        """Hard resolution failures are permanent via both the cause chain
+        and MaxRetryError.reason."""
+        from urllib3.exceptions import MaxRetryError, NameResolutionError
+
+        gai = socket.gaierror(socket.EAI_NONAME, "nodename nor servname provided")
+
+        chained = req.ConnectionError("resolution failed")
+        chained.__cause__ = gai
+        assert not is_transient_download_error(chained)
+
+        # The real urllib3 shape: gaierror on NameResolutionError.__cause__,
+        # carried by MaxRetryError.reason.
+        try:
+            raise NameResolutionError("example.invalid", None, gai) from gai
+        except NameResolutionError as nre:
+            wrapped = req.ConnectionError(
+                MaxRetryError(None, "http://example.invalid/", reason=nre)
+            )
+        assert not is_transient_download_error(wrapped)
+
+        # A garden-variety connection reset stays transient.
+        assert is_transient_download_error(req.ConnectionError("reset by peer"))
+
+    def test_temporary_dns_failure_stays_transient(self) -> None:
+        """EAI_AGAIN (flaky resolver) stays retryable."""
+        gai = socket.gaierror(socket.EAI_AGAIN, "temporary failure in name resolution")
+        chained = req.ConnectionError("resolution failed")
+        chained.__cause__ = gai
+
+        assert is_transient_download_error(chained)
+
+    def test_implicit_context_does_not_reclassify(self) -> None:
+        """A gaierror riding along as implicit __context__ must not turn a
+        genuine connection reset permanent."""
+        try:
+            try:
+                raise socket.gaierror(socket.EAI_NONAME, "first attempt")
+            except socket.gaierror:
+                raise req.ConnectionError("reset by peer") from None
+        except req.ConnectionError as reset:
+            assert reset.__context__ is not None
+            assert is_transient_download_error(reset)
+
+    def test_gaierror_without_errno_stays_transient(self) -> None:
+        """A gaierror carrying no EAI code cannot prove a hard failure."""
+        chained = req.ConnectionError("resolution failed")
+        chained.__cause__ = socket.gaierror("no errno")
+
+        assert is_transient_download_error(chained)
+
+    def test_mixed_chain_hard_failure_wins(self) -> None:
+        """EAI_AGAIN in the chain does not mask a hard failure elsewhere."""
+        again = socket.gaierror(socket.EAI_AGAIN, "temporary failure")
+        hard = socket.gaierror(socket.EAI_NONAME, "unknown host")
+
+        outer = req.ConnectionError(hard)
+        outer.__cause__ = again
+        assert not is_transient_download_error(outer)
+
+        outer = req.ConnectionError(again)
+        outer.__cause__ = hard
+        assert not is_transient_download_error(outer)
+
+    def test_dns_walk_survives_exception_cycles(self) -> None:
+        """A cyclic cause chain must terminate (and stay transient when no
+        resolution failure is present)."""
+        outer = req.ConnectionError("a")
+        inner = ValueError("b")
+        outer.__cause__ = inner
+        inner.__cause__ = outer
+
+        assert is_transient_download_error(outer)
+
+    def test_exhausted_resume_attempts_are_permanent(self) -> None:
+        """download_with_resume already spent its own resume attempts; its
+        EsphomeError wrapper is not retried again at the sweep level."""
+        wrapped = EsphomeError("Failed to download after 3 attempts")
+        wrapped.__cause__ = req.ConnectionError("down")
+        assert not is_transient_download_error(wrapped)
+
+    def test_unrelated_errors_are_permanent(self) -> None:
+        assert not is_transient_download_error(OSError("disk full"))
+        assert not is_transient_download_error(EsphomeError("size mismatch"))
+
+
+class TestFetchWithRetry:
+    def test_logs_the_upcoming_attempt_number(
+        self, caplog: pytest.LogCaptureFixture
+    ) -> None:
+        """The warning names the attempt about to run, not the failed one."""
+        with (
+            patch("esphome.net_retry.time.sleep") as mock_sleep,
+            pytest.raises(req.ConnectionError),
+        ):
+            fetch_with_retry(
+                "https://example.com/f",
+                lambda: (_ for _ in ()).throw(req.ConnectionError("reset")),
+            )
+
+        assert mock_sleep.call_args_list == [call(2), call(4)]
+        assert "(attempt 2/3)" in caplog.text
+        assert "(attempt 3/3)" in caplog.text

tests/unit_tests/test_platformio_extra_script.py

@@ -11,6 +11,7 @@ import pytest
 
 from esphome.core import EsphomeError
 from esphome.platformio.extra_script import (
+    CppDefine,
     ExtraScriptResult,
     _FakeSConsEnv,
     apply_extra_script,
@@ -51,8 +52,8 @@ def test_extra_script_captures_libpath_libs_and_defines(tmp_path):
 
     assert result.libpath == [str(Path("src") / "esp32")]
     assert result.libs == ["algobsec"]
-    assert ("BAR", "1") in result.cppdefines
-    assert "FOO" in result.cppdefines
+    assert CppDefine("BAR", "1") in result.cppdefines
+    assert CppDefine("FOO") in result.cppdefines
     assert result.linkflags == ["-Wl,--gc-sections"]
 
     # Lex like the consumer does: quoting makes raw strings platform-varying
@@ -476,7 +477,7 @@ def test_spaced_cppflag_survives_relexing(tmp_path) -> None:
     """A captured argv token with a space stays one token after lexing."""
     result = ExtraScriptResult(
         cppflags=["-include my hdr.h"],
-        cppdefines=[("MSG", '"hello world"'), "PLAIN"],
+        cppdefines=[CppDefine("MSG", '"hello world"'), CppDefine("PLAIN")],
     )
     flags = captured_as_build_flags(result, library_dir=tmp_path)
     assert lex_build_flags(flags, "test") == [