Merge branch 'dev' into sendspin-artwork

tests/benchmarks/core/bench_scheduler.cpp

@@ -101,8 +101,8 @@ static void Scheduler_SetTimeout(benchmark::State &state) {
   Component dummy_component;
 
   // Register 3 timeouts then call() — realistic worst case where multiple
-  // components schedule in the same loop iteration. Keeps item count within
-  // the recycling pool (MAX_POOL_SIZE=5) to avoid spurious malloc/free.
+  // components schedule in the same loop iteration. warm_pool fills the
+  // freelist so acquire/recycle never falls back to malloc.
   static constexpr int kBatchSize = 3;
   static_assert(kInnerIterations % kBatchSize == 0, "kInnerIterations must be divisible by kBatchSize");
   warm_pool(scheduler, &dummy_component, kBatchSize, 1000);
@@ -209,9 +209,9 @@ static void Scheduler_SetTimeout_ExceedPool(benchmark::State &state) {
   Scheduler scheduler;
   Component dummy_component;
 
-  // Register 10 timeouts then call() — exceeds MAX_POOL_SIZE=5 to measure
-  // the performance cliff when the recycling pool is exhausted and items
-  // must be malloc'd/freed.
+  // Register 10 timeouts then call() — larger working set than the 3-item
+  // batches above. With the unbounded freelist, warm_pool preallocates 10
+  // items so this measures steady-state, not malloc cliff.
   static constexpr int kBatchSize = 10;
   static_assert(kInnerIterations % kBatchSize == 0, "kInnerIterations must be divisible by kBatchSize");
   warm_pool(scheduler, &dummy_component, kBatchSize, 1000);

tests/component_tests/lvgl/config/line_points.yaml

@@ -0,0 +1,84 @@
+esphome:
+  name: test-line
+
+esp32:
+  board: lolin_c3_mini
+
+spi:
+  mosi_pin:
+    number: GPIO2
+    ignore_strapping_warning: true
+  clk_pin: GPIO1
+
+display:
+  - platform: mipi_spi
+    data_rate: 20MHz
+    model: st7735
+    cs_pin:
+      number: GPIO8
+      ignore_strapping_warning: true
+    dc_pin:
+      number: GPIO3
+
+lvgl:
+  widgets:
+    # Dict format
+    - line:
+        id: line_dict
+        points:
+          - x: 10
+            y: 20
+          - x: 100
+            y: 200
+          - x: 0
+            y: 0
+
+    # List format
+    - line:
+        id: line_list
+        points:
+          - [10, 20]
+          - [100, 200]
+          - [0, 0]
+
+    # String format
+    - line:
+        id: line_string
+        points:
+          - "10, 20"
+          - "100, 200"
+          - "0, 0"
+
+    # Percentage - dict format
+    - line:
+        id: line_pct_dict
+        points:
+          - x: "50%"
+            y: "75%"
+
+    # Percentage - list format
+    - line:
+        id: line_pct_list
+        points:
+          - ["50%", "75%"]
+
+    # Percentage - string format
+    - line:
+        id: line_pct_string
+        points:
+          - "50%, 75%"
+
+    # Mixed integer and percentage
+    - line:
+        id: line_mixed_dict
+        points:
+          - x: 10
+            y: "50%"
+          - x: "25%"
+            y: 200
+
+    - line:
+        id: line_mixed_list
+        points:
+          - [10, "50%"]
+          - ["25%", 200]

tests/component_tests/lvgl/test_line.py

@@ -0,0 +1,147 @@
+"""Tests for the LVGL line widget point schema and code generation."""
+
+from __future__ import annotations
+
+import re
+
+import pytest
+
+from esphome.components.lvgl.schemas import point_schema
+from esphome.config_validation import Invalid
+from esphome.const import CONF_X, CONF_Y
+
+# ---------------------------------------------------------------------------
+# Validation: point_schema normalises dict / list / string to same result
+# ---------------------------------------------------------------------------
+
+
+class TestPointSchemaValidation:
+    """Test that all point input formats normalise to the same dict."""
+
+    @pytest.mark.parametrize(
+        "dict_input,list_input,string_input",
+        [
+            ({CONF_X: 10, CONF_Y: 20}, [10, 20], "10, 20"),
+            ({CONF_X: 0, CONF_Y: 0}, [0, 0], "0, 0"),
+            ({CONF_X: 100, CONF_Y: 200}, [100, 200], "100, 200"),
+            ({CONF_X: -5, CONF_Y: -10}, [-5, -10], "-5, -10"),
+        ],
+    )
+    def test_integer_formats_produce_same_result(
+        self, dict_input, list_input, string_input
+    ):
+        result_dict = point_schema(dict_input)
+        result_list = point_schema(list_input)
+        result_string = point_schema(string_input)
+
+        assert result_dict == result_list
+        assert result_dict == result_string
+
+    def test_percentage_formats_produce_same_result(self):
+        result_dict = point_schema({CONF_X: "50%", CONF_Y: "75%"})
+        result_list = point_schema(["50%", "75%"])
+        result_string = point_schema("50%, 75%")
+
+        assert result_dict == result_list
+        assert result_dict == result_string
+
+    def test_pixel_suffix_matches_plain_integer(self):
+        result_px = point_schema({CONF_X: "10px", CONF_Y: "20px"})
+        result_int = point_schema({CONF_X: 10, CONF_Y: 20})
+
+        assert result_px == result_int
+
+    @pytest.mark.parametrize(
+        "value",
+        [
+            {CONF_X: 50, CONF_Y: 75},
+            [50, 75],
+            "50, 75",
+        ],
+    )
+    def test_output_contains_x_and_y(self, value):
+        result = point_schema(value)
+
+        assert CONF_X in result
+        assert CONF_Y in result
+
+    def test_list_wrong_length_raises(self):
+        with pytest.raises(Invalid, match="Invalid point"):
+            point_schema([1])
+
+        with pytest.raises(Invalid, match="Invalid point"):
+            point_schema([1, 2, 3])
+
+    def test_string_without_comma_raises(self):
+        with pytest.raises(Invalid, match="Invalid point"):
+            point_schema("garbage")
+
+    def test_string_extra_commas_raises(self):
+        with pytest.raises(Invalid, match="Invalid point"):
+            point_schema("1,2,3")
+
+
+# ---------------------------------------------------------------------------
+# Code generation: different point formats produce identical C++ output
+# ---------------------------------------------------------------------------
+
+_SET_POINTS_RE = re.compile(r"(\w+)->set_points\((.+?)\);")
+
+
+def _extract_set_points(main_cpp: str) -> dict[str, str]:
+    """Return {var_name: args_text} for every set_points() call found."""
+    return {m.group(1): m.group(2) for m in _SET_POINTS_RE.finditer(main_cpp)}
+
+
+class TestLineCodeGeneration:
+    """Verify that alternative point formats generate identical C++ code."""
+
+    @pytest.fixture()
+    def main_cpp(self, generate_main, component_config_path) -> str:
+        return generate_main(component_config_path("line_points.yaml"))
+
+    @pytest.fixture()
+    def set_points_calls(self, main_cpp) -> dict[str, str]:
+        return _extract_set_points(main_cpp)
+
+    def test_integer_points_all_formats_match(self, set_points_calls):
+        """Dict, list, and string formats with integer points produce same set_points call."""
+        assert set_points_calls["line_dict"] == set_points_calls["line_list"]
+        assert set_points_calls["line_dict"] == set_points_calls["line_string"]
+
+    def test_percentage_points_all_formats_match(self, set_points_calls):
+        """Dict, list, and string formats with percentage points produce same set_points call."""
+        assert set_points_calls["line_pct_dict"] == set_points_calls["line_pct_list"]
+        assert set_points_calls["line_pct_dict"] == set_points_calls["line_pct_string"]
+
+    def test_mixed_points_formats_match(self, set_points_calls):
+        """Dict and list formats with mixed int/percent points produce same set_points call."""
+        assert (
+            set_points_calls["line_mixed_dict"] == set_points_calls["line_mixed_list"]
+        )
+
+    def test_integer_points_contain_expected_values(self, set_points_calls):
+        """Integer points appear literally in the generated code."""
+        args = set_points_calls["line_dict"]
+        for val in ("10", "20", "100", "200"):
+            assert val in args
+
+    def test_percentage_points_use_lv_pct(self, set_points_calls):
+        """Percentage points are generated using the lv_pct() macro."""
+        args = set_points_calls["line_pct_dict"]
+        assert "lv_pct(50)" in args
+        assert "lv_pct(75)" in args
+
+    def test_all_lines_present(self, set_points_calls):
+        """All expected line IDs have a set_points call."""
+        expected = {
+            "line_dict",
+            "line_list",
+            "line_string",
+            "line_pct_dict",
+            "line_pct_list",
+            "line_pct_string",
+            "line_mixed_dict",
+            "line_mixed_list",
+        }
+        assert expected.issubset(set_points_calls.keys())

tests/component_tests/packages/test_packages.py

@@ -14,6 +14,7 @@ from esphome.components.packages import (
     do_packages_pass,
     is_package_definition,
     merge_packages,
+    resolve_packages,
 )
 from esphome.components.substitutions import ContextVars, do_substitution_pass
 import esphome.config as config_module
@@ -1621,3 +1622,122 @@ def test_remote_package_vars_resolved_against_sibling_package_substitutions(
     actual = packages_pass(config)
 
     assert actual[CONF_SENSOR][0]["pin"] == "GPIO5"
+
+
+# ---------------------------------------------------------------------------
+# resolve_packages — single-call wrapper around do_packages_pass + merge_packages
+# ---------------------------------------------------------------------------
+
+
+def test_resolve_packages_returns_config_unchanged_without_packages() -> None:
+    """No ``packages:`` key → no-op, same dict back."""
+    config = {CONF_ESPHOME: {CONF_NAME: "test"}, CONF_WIFI: {CONF_SSID: "x"}}
+    result = resolve_packages(config)
+    assert result is config
+    assert CONF_PACKAGES not in result
+
+
+def test_resolve_packages_loads_and_merges_in_one_call() -> None:
+    """End-to-end: a config with one local-dict package gets its blocks flattened."""
+    config = {
+        CONF_ESPHOME: {CONF_NAME: "main"},
+        CONF_PACKAGES: {
+            "shared": {
+                CONF_WIFI: {CONF_SSID: "from_package"},
+                CONF_SENSOR: [
+                    {CONF_PLATFORM: "template", CONF_NAME: "from_package_sensor"},
+                ],
+            }
+        },
+    }
+    result = resolve_packages(config)
+    # ``packages:`` is gone — it was consumed by the merge.
+    assert CONF_PACKAGES not in result
+    # Blocks contributed by the package are now top-level.
+    assert result[CONF_WIFI][CONF_SSID] == "from_package"
+    assert result[CONF_SENSOR][0][CONF_NAME] == "from_package_sensor"
+    # The main config's own keys survive untouched.
+    assert result[CONF_ESPHOME][CONF_NAME] == "main"
+
+
+def test_resolve_packages_preserves_main_config_overrides() -> None:
+    """Main-config values win over package values for the same key.
+
+    Pinning the precedence ESPHome's compiler uses so any future
+    refactor of the wrapper doesn't accidentally flip the order.
+    """
+    config = {
+        CONF_ESPHOME: {CONF_NAME: "main"},
+        CONF_WIFI: {CONF_SSID: "main_wins"},
+        CONF_PACKAGES: {
+            "shared": {CONF_WIFI: {CONF_SSID: "package_loses"}},
+        },
+    }
+    result = resolve_packages(config)
+    assert result[CONF_WIFI][CONF_SSID] == "main_wins"
+
+
+def test_resolve_packages_forwards_command_line_substitutions() -> None:
+    """``command_line_substitutions`` reaches the underlying ``do_packages_pass``.
+
+    The wrapper exists so external tools have one stable seam; if
+    that seam silently dropped a kwarg the underlying call accepts,
+    callers would see surprising behaviour. This pins the
+    pass-through.
+    """
+    config = {
+        CONF_ESPHOME: {CONF_NAME: "main"},
+        CONF_PACKAGES: {"shared": {CONF_WIFI: {CONF_SSID: "from_package"}}},
+    }
+    with patch(
+        "esphome.components.packages.do_packages_pass",
+        wraps=do_packages_pass,
+    ) as spy:
+        resolve_packages(config, command_line_substitutions={"foo": "bar"})
+    spy.assert_called_once()
+    _, kwargs = spy.call_args
+    assert kwargs.get("command_line_substitutions") == {"foo": "bar"}
+
+
+def test_resolve_packages_does_not_run_substitutions() -> None:
+    """``${var}`` placeholders inside package content stay literal.
+
+    The full ``validate_config`` pipeline runs ``do_substitution_pass``
+    BETWEEN ``do_packages_pass`` and ``merge_packages``; this wrapper
+    skips it on purpose. Pin that contract so a future refactor can't
+    silently start resolving substitutions and break callers that
+    deliberately compose the passes themselves.
+    """
+    config = {
+        CONF_ESPHOME: {CONF_NAME: "main"},
+        CONF_SUBSTITUTIONS: {"ssid_value": "resolved_ssid"},
+        CONF_PACKAGES: {
+            "shared": {CONF_WIFI: {CONF_SSID: "${ssid_value}"}},
+        },
+    }
+    result = resolve_packages(config)
+    # Without ``do_substitution_pass`` the placeholder is preserved.
+    assert result[CONF_WIFI][CONF_SSID] == "${ssid_value}"
+
+
+def test_resolve_packages_does_not_apply_extend_remove() -> None:
+    """Top-level ``!remove`` / ``!extend`` markers stay in the merged dict.
+
+    The full ``validate_config`` pipeline runs ``resolve_extend_remove``
+    AFTER ``merge_packages``; this wrapper skips it on purpose. Pin
+    that contract: a package-contributed block paired with a top-level
+    ``!remove`` is left as-is for callers to handle (or for them to
+    call ``resolve_extend_remove`` themselves).
+    """
+    config = {
+        CONF_ESPHOME: {CONF_NAME: "main"},
+        CONF_WIFI: Remove(),
+        CONF_PACKAGES: {
+            "shared": {CONF_WIFI: {CONF_SSID: "from_package"}},
+        },
+    }
+    result = resolve_packages(config)
+    # ``merge_packages`` keeps the top-level ``!remove`` (it wins
+    # over the package value during merge), and the marker is not
+    # resolved by this wrapper.
+    assert isinstance(result[CONF_WIFI], Remove)

tests/components/http_request/http_request.yaml

@@ -50,12 +50,33 @@ esphome:
                   format: "After delay, body still: %s"
                   args:
                     - body.c_str()
+              # Regression test for esphome/esphome#16224: a LightControlAction
+              # nested inside on_response with capture_response: true puts
+              # `std::string &` into the trigger's Ts..., which exposed a codegen
+              # bug where the apply lambda's parameter list did not match the
+              # ApplyFn signature.
+              - light.turn_on:
+                  id: test_regression_light
+                  brightness: 100%
+                  effect: "None"
 
 http_request:
   useragent: esphome/tagreader
   timeout: 10s
   verify_ssl: ${verify_ssl}
 
+output:
+  - platform: template
+    id: test_regression_output
+    type: float
+    write_action:
+      - logger.log: "set"
+
+light:
+  - platform: monochromatic
+    id: test_regression_light
+    output: test_regression_output
+
 script:
   - id: does_not_compile
     parameters:

tests/components/lvgl/lvgl-package.yaml

@@ -649,11 +649,15 @@ lvgl:
             on_scroll_begin:
               logger.log: Button clicked
             on_release:
-              logger.log: Button clicked
+              logger.log:
+                format: Button released at %d/%d
+                args: [point.x, point.y]
             on_long_press_repeat:
               logger.log: Button clicked
             on_pressing:
-              logger.log: Button pressing
+              logger.log:
+                format: Button pressing at %d/%d
+                args: [point.x, point.y]
             on_press_lost:
               logger.log: Button press lost
             on_single_click:
@@ -925,6 +929,10 @@ lvgl:
                   value: !lambda |-
                       static float yyy = 83.0;
                       return yyy + .8;
+            on_release:
+              logger.log:
+                format: Slider released at %d/%d with value %.0f
+                args: [point.x, point.y, x]
         - button:
             styles: spin_button
             id: spin_up
@@ -1038,7 +1046,10 @@ lvgl:
               - 5, 5
               - x: !lambda return random_uint32() % 100;
                 y: !lambda return random_uint32() % 100;
-              - 70, 70
+              - x: 10%
+                y: 50%
+              - 70%, 70%
+              - [75%, 75%]
               - 120, 10
               - 180, 60
               - 240, 10

tests/components/speaker/common-media_player.yaml

@@ -17,3 +17,16 @@ media_player:
     volume_max: 0.95
     volume_min: 0.0
     task_stack_in_psram: true
+    files:
+      - id: speaker_test_audio
+        file:
+          type: local
+          path: $component_dir/test.wav
+
+script:
+  - id: play_built_in_file
+    then:
+      - media_player.speaker.play_on_device_media_file:
+          id: speaker_media_player_id
+          media_file: speaker_test_audio
+          announcement: true

tests/components/template/common-base.yaml

@@ -202,6 +202,11 @@ sensor:
           value: last
       - timeout:
           timeout: 1d
+      - to_ntc_temperature:
+          calibration:
+            b_constant: 3950
+            reference_temperature: 25.0°C
+            reference_resistance: 10kOhm
       - to_ntc_resistance:
           calibration:
             - 10.0kOhm -> 25°C
@@ -270,8 +275,6 @@ cover:
     stop_action:
       - logger.log: stop_action
     optimistic: true
-    on_open:
-      - logger.log: "Cover on_open (deprecated)"
     on_opened:
       - logger.log: "Cover fully opened"
     on_closed:
@@ -369,6 +372,19 @@ number:
         - valve.control:
             id: template_valve
             position: !lambda "return x / 100.0f;"
+  # Same regression test for cover.control: forces the apply-lambda
+  # codegen to handle a non-empty trigger Ts (float).
+  - platform: template
+    id: template_cover_position_number
+    optimistic: true
+    min_value: 0
+    max_value: 100
+    step: 1
+    on_value:
+      then:
+        - cover.control:
+            id: template_cover_with_triggers
+            position: !lambda "return x / 100.0f;"
 
 select:
   - platform: template

tests/integration/conftest.py

@@ -501,14 +501,15 @@ async def _read_stream_lines(
 
 
 @asynccontextmanager
-async def run_binary_and_wait_for_port(
+async def run_binary(
     binary_path: Path,
-    host: str,
-    port: int,
-    timeout: float = PORT_WAIT_TIMEOUT,
     line_callback: Callable[[str], None] | None = None,
-) -> AsyncGenerator[None]:
-    """Run a binary, wait for it to open a port, and clean up on exit."""
+) -> AsyncGenerator[tuple[asyncio.subprocess.Process, list[str]]]:
+    """Run a binary under a PTY, capture log output, and clean up on exit.
+
+    Yields the running ``Process`` and a live list of captured log lines.
+    No port wait -- callers that need that should use
+    ``run_binary_and_wait_for_port``."""
     # Create a pseudo-terminal to make the binary think it's running interactively
     # This is needed because the ESPHome host logger checks isatty()
     controller_fd, device_fd = pty.openpty()
@@ -535,7 +536,6 @@ async def run_binary_and_wait_for_port(
     controller_transport, _ = await loop.connect_read_pipe(
         lambda: controller_protocol, os.fdopen(controller_fd, "rb", 0)
     )
-    output_reader = controller_reader
 
     if process.returncode is not None:
         raise RuntimeError(
@@ -543,27 +543,59 @@ async def run_binary_and_wait_for_port(
             "Ensure the binary is valid and can run successfully."
         )
 
-    # Wait for the API server to start listening
-    loop = asyncio.get_running_loop()
-    start_time = loop.time()
-
-    # Start collecting output
     stdout_lines: list[str] = []
-    output_tasks: list[asyncio.Task] = []
+    output_task = asyncio.create_task(
+        _read_stream_lines(controller_reader, stdout_lines, sys.stdout, line_callback)
+    )
 
     try:
-        # Read from output stream
-        output_tasks = [
-            asyncio.create_task(
-                _read_stream_lines(
-                    output_reader, stdout_lines, sys.stdout, line_callback
-                )
-            )
-        ]
-
         # Small yield to ensure the process has a chance to start
         await asyncio.sleep(0)
+        yield process, stdout_lines
+    finally:
+        output_task.cancel()
+        result = await asyncio.gather(output_task, return_exceptions=True)
+        if isinstance(result[0], Exception) and not isinstance(
+            result[0], asyncio.CancelledError
+        ):
+            print(f"Error reading from PTY: {result[0]}", file=sys.stderr)
 
+        # Close the PTY transport (Unix only)
+        if controller_transport is not None:
+            controller_transport.close()
+
+        # Cleanup: terminate the process gracefully
+        if process.returncode is None:
+            # Send SIGINT (Ctrl+C) for graceful shutdown
+            process.send_signal(signal.SIGINT)
+            try:
+                await asyncio.wait_for(process.wait(), timeout=SIGINT_TIMEOUT)
+            except TimeoutError:
+                # If SIGINT didn't work, try SIGTERM
+                process.terminate()
+                try:
+                    await asyncio.wait_for(process.wait(), timeout=SIGTERM_TIMEOUT)
+                except TimeoutError:
+                    # Last resort: SIGKILL
+                    process.kill()
+                    await process.wait()
+
+
+@asynccontextmanager
+async def run_binary_and_wait_for_port(
+    binary_path: Path,
+    host: str,
+    port: int,
+    timeout: float = PORT_WAIT_TIMEOUT,
+    line_callback: Callable[[str], None] | None = None,
+) -> AsyncGenerator[None]:
+    """Run a binary, wait for it to open a port, and clean up on exit."""
+    async with run_binary(binary_path, line_callback=line_callback) as (
+        process,
+        stdout_lines,
+    ):
+        loop = asyncio.get_running_loop()
+        start_time = loop.time()
         while loop.time() - start_time < timeout:
             try:
                 # Try to connect to the port
@@ -593,41 +625,6 @@ async def run_binary_and_wait_for_port(
 
         raise TimeoutError(error_msg)
 
-    finally:
-        # Cancel output collection tasks
-        for task in output_tasks:
-            task.cancel()
-        # Wait for tasks to complete and check for exceptions
-        results = await asyncio.gather(*output_tasks, return_exceptions=True)
-        for i, result in enumerate(results):
-            if isinstance(result, Exception) and not isinstance(
-                result, asyncio.CancelledError
-            ):
-                print(
-                    f"Error reading from PTY: {result}",
-                    file=sys.stderr,
-                )
-
-        # Close the PTY transport (Unix only)
-        if controller_transport is not None:
-            controller_transport.close()
-
-        # Cleanup: terminate the process gracefully
-        if process.returncode is None:
-            # Send SIGINT (Ctrl+C) for graceful shutdown
-            process.send_signal(signal.SIGINT)
-            try:
-                await asyncio.wait_for(process.wait(), timeout=SIGINT_TIMEOUT)
-            except TimeoutError:
-                # If SIGINT didn't work, try SIGTERM
-                process.terminate()
-                try:
-                    await asyncio.wait_for(process.wait(), timeout=SIGTERM_TIMEOUT)
-                except TimeoutError:
-                    # Last resort: SIGKILL
-                    process.kill()
-                    await process.wait()
-
 
 @asynccontextmanager
 async def run_compiled_context(

tests/integration/fixtures/safe_mode_loop_runs.yaml

@@ -0,0 +1,25 @@
+esphome:
+  name: safe-mode-loop-runs
+
+host:
+
+logger:
+
+safe_mode:
+  num_attempts: 10
+  on_safe_mode:
+    - lambda: |-
+        // Spawn a detached thread that logs a unique marker. The
+        // non-main-thread log goes through the task log buffer, which
+        // is only drained by Logger::loop(). If looping components
+        // weren't initialized (the bug fixed in #16269), the buffer is
+        // never read and the marker never reaches the console.
+        struct MarkerThread {
+          static void *thread_func(void *) {
+            ESP_LOGI("safe_mode_test", "looping component ran in safe mode");
+            return nullptr;
+          }
+        };
+        pthread_t t;
+        pthread_create(&t, nullptr, MarkerThread::thread_func, nullptr);
+        pthread_detach(t);

tests/integration/fixtures/scheduler_pool.yaml

@@ -221,14 +221,10 @@ script:
   - id: test_full_pool_reuse
     then:
       - lambda: |-
-          ESP_LOGI("test", "Phase 6: Testing pool size limits after Phase 5 items complete");
+          ESP_LOGI("test", "Phase 6: Testing pool reuse after Phase 5 items complete");
 
-          // At this point, all Phase 5 timeouts should have completed and been recycled.
-          // The pool should be at its maximum size (5).
-          // Creating 10 new items tests that:
-          // - First 5 items reuse from the pool
-          // - Remaining 5 items allocate new (pool empty)
-          // - Pool doesn't grow beyond MAX_POOL_SIZE of 5
+          // Phase 5 timeouts have completed and been recycled. The freelist is unbounded;
+          // creating 10 new items reuses from it and only allocates fresh when empty.
 
           auto *component = id(test_sensor);
           int full_reuse_count = 10;

tests/integration/host_prefs.py

@@ -0,0 +1,39 @@
+"""Helpers for manipulating the host platform's preferences file.
+
+ESPHome's host platform stores preferences in
+``~/.esphome/prefs/<app_name>.prefs`` using a simple binary layout that
+mirrors ``HostPreferences::sync()``:
+``[uint32_t key][uint8_t len][uint8_t data[len]]`` per entry.
+
+Tests use these helpers to pre-populate state the binary will see at
+boot (e.g. forcing safe mode) or to clear stale state between runs.
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+import struct
+
+
+def host_prefs_path(device_name: str) -> Path:
+    """Return the on-disk prefs file path for a host-platform device."""
+    return Path.home() / ".esphome" / "prefs" / f"{device_name}.prefs"
+
+
+def clear_host_prefs(device_name: str) -> None:
+    """Delete the prefs file for a host-platform device, if it exists."""
+    host_prefs_path(device_name).unlink(missing_ok=True)
+
+
+def write_host_pref(device_name: str, key: int, data: bytes) -> Path:
+    """Write a single preference entry, replacing the file's contents.
+
+    Returns the path that was written.
+    """
+    if len(data) > 255:
+        raise ValueError(f"Preference data too long: {len(data)} bytes (max 255)")
+    path = host_prefs_path(device_name)
+    path.parent.mkdir(parents=True, exist_ok=True)
+    payload = struct.pack("<IB", key, len(data)) + data
+    path.write_bytes(payload)
+    return path

tests/integration/test_safe_mode_loop_runs.py

@@ -0,0 +1,94 @@
+"""Regression test for safe_mode + looping_components init ordering.
+
+Reproduces the bug fixed in https://github.com/esphome/esphome/pull/16269:
+``App.looping_components_.init(...)`` was emitted at ``CoroPriority.FINAL``,
+which placed it *after* the ``safe_mode`` early-return in ``setup_app()``.
+When safe mode was entered, the ``FixedVector`` backing the looping-component
+list was never sized, ``looping_components_active_end_`` stayed at 0, and
+``loop()`` iterated zero components -- so any looping component above
+``CoroPriority.APPLICATION`` (e.g. wifi, logger) never ran.
+
+The test forces safe mode by writing ``ENTER_SAFE_MODE_MAGIC`` to the host
+preferences file before booting, then asserts that ``Logger::loop()`` runs
+by logging from a non-main thread. Non-main-thread logs are buffered in
+``TaskLogBuffer`` and only emitted to the console when ``Logger::loop()``
+drains the buffer. Without the fix, the marker stays in the buffer
+forever; with the fix, it reaches the console.
+
+The API server (``CoroPriority.WEB``, 40) is registered below safe_mode
+(``CoroPriority.APPLICATION``, 50), so it's never set up when safe mode
+is active and ``run_compiled`` would hang waiting for the API port.
+This test uses ``run_binary`` directly to skip the port wait.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import re
+import struct
+
+import pytest
+
+from .conftest import run_binary
+from .host_prefs import clear_host_prefs, write_host_pref
+from .types import CompileFunction, ConfigWriter
+
+# Must match esphome::safe_mode::RTC_KEY in safe_mode.h
+SAFE_MODE_RTC_KEY = 233825507
+# Must match esphome::safe_mode::SafeModeComponent::ENTER_SAFE_MODE_MAGIC
+ENTER_SAFE_MODE_MAGIC = 0x5AFE5AFE
+
+DEVICE_NAME = "safe-mode-loop-runs"
+THREAD_LOG_MARKER = "looping component ran in safe mode"
+
+
+@pytest.mark.asyncio
+async def test_safe_mode_loop_runs(
+    yaml_config: str,
+    write_yaml_config: ConfigWriter,
+    compile_esphome: CompileFunction,
+) -> None:
+    """When safe mode is active, ``App.loop()`` must still iterate looping
+    components -- proven here by a thread-logged marker reaching the
+    console (which requires ``Logger::loop()`` to run)."""
+    config_path = await write_yaml_config(yaml_config)
+    binary_path = await compile_esphome(config_path)
+
+    # Compile finished successfully; pre-populate prefs so the *next* run
+    # enters safe mode immediately.
+    write_host_pref(
+        DEVICE_NAME, SAFE_MODE_RTC_KEY, struct.pack("<I", ENTER_SAFE_MODE_MAGIC)
+    )
+
+    try:
+        loop = asyncio.get_running_loop()
+        safe_mode_active = loop.create_future()
+        thread_log_seen = loop.create_future()
+        safe_mode_pattern = re.compile(r"SAFE MODE IS ACTIVE")
+        thread_log_pattern = re.compile(re.escape(THREAD_LOG_MARKER))
+
+        def on_log(line: str) -> None:
+            if not safe_mode_active.done() and safe_mode_pattern.search(line):
+                safe_mode_active.set_result(True)
+            if not thread_log_seen.done() and thread_log_pattern.search(line):
+                thread_log_seen.set_result(True)
+
+        async with run_binary(binary_path, line_callback=on_log):
+            try:
+                await asyncio.wait_for(safe_mode_active, timeout=15.0)
+            except TimeoutError:
+                pytest.fail(
+                    "Did not observe 'SAFE MODE IS ACTIVE' -- safe mode "
+                    "didn't trigger, so this test isn't exercising the bug."
+                )
+            try:
+                await asyncio.wait_for(thread_log_seen, timeout=10.0)
+            except TimeoutError:
+                pytest.fail(
+                    f"Did not observe thread-logged marker {THREAD_LOG_MARKER!r} "
+                    "within timeout. Logger::loop() never drained the task "
+                    "log buffer, meaning App.looping_components_ was never "
+                    "sized -- this is the regression #16269 fixed."
+                )
+    finally:
+        clear_host_prefs(DEVICE_NAME)

tests/integration/test_scheduler_pool.py

@@ -180,16 +180,22 @@ async def test_scheduler_pool(
     # Verify pool behavior
     assert pool_recycle_count > 0, "Should have recycled items to pool"
 
-    # Check pool metrics
-    if pool_recycle_count > 0:
-        max_pool_size = 0
-        for line in log_lines:
-            if match := recycle_pattern.search(line):
-                size = int(match.group(1))
-                max_pool_size = max(max_pool_size, size)
+    # Pool is unbounded; the cap was the source of the churn it was meant to prevent.
+    assert pool_full_count == 0, (
+        f"Pool should never report full (got {pool_full_count})"
+    )
 
-        # Pool can grow up to its maximum of 5
-        assert max_pool_size <= 5, f"Pool grew beyond maximum ({max_pool_size})"
+    # Verify the pool actually grew past the old MAX_POOL_SIZE=5 cap.
+    # Phase 5 + Phase 6 schedule 8 + 10 same-component timeouts respectively, so the
+    # observed peak should comfortably exceed 5. Without this lower-bound check, a
+    # silent regression that re-introduced a small cap could pass the test above.
+    max_pool_size = 0
+    for line in log_lines:
+        if match := recycle_pattern.search(line):
+            max_pool_size = max(max_pool_size, int(match.group(1)))
+    assert max_pool_size > 5, (
+        f"Pool should grow past the old cap of 5; observed peak {max_pool_size}"
+    )
 
     # Log summary for debugging
     print("\nScheduler Pool Test Summary (Python Orchestrated):")

tests/integration/test_template_text_save.py

@@ -9,7 +9,6 @@ Tests that:
 from __future__ import annotations
 
 import asyncio
-from pathlib import Path
 import socket
 from typing import Any
 
@@ -17,9 +16,12 @@ from aioesphomeapi import TextInfo, TextState
 import pytest
 
 from .conftest import run_binary_and_wait_for_port, wait_and_connect_api_client
+from .host_prefs import clear_host_prefs
 from .state_utils import InitialStateHelper, require_entity
 from .types import CompileFunction, ConfigWriter
 
+DEVICE_NAME = "host-template-text-save-test"
+
 
 @pytest.mark.asyncio
 async def test_template_text_save(
@@ -32,11 +34,7 @@ async def test_template_text_save(
     port, port_socket = reserved_tcp_port
 
     # Clean up any stale preference file from previous runs
-    prefs_file = (
-        Path.home() / ".esphome" / "prefs" / "host-template-text-save-test.prefs"
-    )
-    if prefs_file.exists():
-        prefs_file.unlink()
+    clear_host_prefs(DEVICE_NAME)
 
     # Write and compile once
     config_path = await write_yaml_config(yaml_config)
@@ -59,7 +57,7 @@ async def test_template_text_save(
         wait_and_connect_api_client(port=port) as client,
     ):
         device_info = await client.device_info()
-        assert device_info.name == "host-template-text-save-test"
+        assert device_info.name == DEVICE_NAME
 
         entities, _ = await client.list_entities_services()
         text_entity = require_entity(
@@ -127,5 +125,4 @@ async def test_template_text_save(
         )
 
     # Clean up preference file
-    if prefs_file.exists():
-        prefs_file.unlink()
+    clear_host_prefs(DEVICE_NAME)

tests/unit_tests/components/api/test_client.py

@@ -4,6 +4,7 @@ from __future__ import annotations
 
 from unittest.mock import patch
 
+from esphome.components import esp32
 from esphome.components.api import client as api_client
 from esphome.core import EsphomeError
 
@@ -18,11 +19,11 @@ def test_decoder_swallows_esphome_error() -> None:
     reconnect.
     """
     config = {"esphome": {"name": "test"}}
-    processor = api_client._LogLineProcessor(config, None)
 
     with patch.object(
-        api_client, "process_stacktrace", side_effect=EsphomeError("no idedata")
+        esp32, "process_stacktrace", side_effect=EsphomeError("no idedata")
     ) as mock_process:
+        processor = api_client._LogLineProcessor(config, esp32.process_stacktrace)
         processor.process_line("PC: 0x4010496e")
 
     assert mock_process.called
@@ -47,9 +48,9 @@ def test_decoder_warning_uses_fallback_for_empty_error(caplog) -> None:
     must show a useful explanation rather than empty parens.
     """
     config = {"esphome": {"name": "test"}}
-    processor = api_client._LogLineProcessor(config, None)
 
-    with patch.object(api_client, "process_stacktrace", side_effect=EsphomeError()):
+    with patch.object(esp32, "process_stacktrace", side_effect=EsphomeError()):
+        processor = api_client._LogLineProcessor(config, esp32.process_stacktrace)
         processor.process_line("PC: 0x4010496e")
 
     warnings = [r.message for r in caplog.records if r.levelname == "WARNING"]
@@ -65,11 +66,11 @@ def test_decoder_short_circuits_after_failure() -> None:
     stall log streaming.
     """
     config = {"esphome": {"name": "test"}}
-    processor = api_client._LogLineProcessor(config, None)
 
     with patch.object(
-        api_client, "process_stacktrace", side_effect=EsphomeError("no idedata")
+        esp32, "process_stacktrace", side_effect=EsphomeError("no idedata")
     ) as mock_process:
+        processor = api_client._LogLineProcessor(config, esp32.process_stacktrace)
         processor.process_line("PC: 0x4010496e")
         processor.process_line("BT0: 0x4010496e")
         processor.process_line("BT1: 0x401049aa")
@@ -80,18 +81,18 @@ def test_decoder_short_circuits_after_failure() -> None:
 def test_decoder_threads_backtrace_state() -> None:
     """When decoding succeeds, backtrace_state is threaded across calls."""
     config = {"esphome": {"name": "test"}}
-    processor = api_client._LogLineProcessor(config, None)
 
     with patch.object(
-        api_client, "process_stacktrace", side_effect=[True, False]
+        esp32, "process_stacktrace", side_effect=[True, False]
     ) as mock_process:
+        processor = api_client._LogLineProcessor(config, esp32.process_stacktrace)
         processor.process_line(">>>stack>>>")
         assert processor.backtrace_state is True
         processor.process_line("<<<stack<<<")
         assert processor.backtrace_state is False
 
-    assert mock_process.call_args_list[0].kwargs == {"backtrace_state": False}
-    assert mock_process.call_args_list[1].kwargs == {"backtrace_state": True}
+    assert not mock_process.call_args_list[0].args[-1]
+    assert mock_process.call_args_list[1].args[-1]
 
 
 def test_decoder_uses_platform_handler_when_provided() -> None:
@@ -105,7 +106,7 @@ def test_decoder_uses_platform_handler_when_provided() -> None:
 
     processor = api_client._LogLineProcessor(config, platform_handler)
 
-    with patch.object(api_client, "process_stacktrace") as mock_generic:
+    with patch.object(esp32, "process_stacktrace") as mock_generic:
         processor.process_line("BT0: 0x4010496e")
 
     assert calls == [(config, "BT0: 0x4010496e", False)]

tests/unit_tests/components/test_esp_stacktrace.py

@@ -0,0 +1,109 @@
+"""Tests for ESP32 component."""
+
+from pathlib import Path
+from unittest.mock import Mock
+
+
+def test_process_stacktrace_esp8266_exception(setup_core: Path, caplog) -> None:
+    """Test process_stacktrace handles ESP8266 exceptions."""
+    from esphome.components.esp8266 import process_stacktrace
+
+    config = {"name": "test"}
+
+    # Test exception type parsing
+    line = "Exception (28):"
+    backtrace_state = False
+
+    result = process_stacktrace(config, line, backtrace_state)
+
+    assert "Access to invalid address: LOAD (wild pointer?)" in caplog.text
+    assert result is False
+
+
+def test_process_stacktrace_esp8266_backtrace(
+    setup_core: Path, mock_esp8266_decode_pc: Mock
+) -> None:
+    """Test process_stacktrace handles ESP8266 multi-line backtrace."""
+    from esphome.components.esp8266 import process_stacktrace
+
+    config = {"name": "test"}
+
+    # Start of backtrace
+    line1 = ">>>stack>>>"
+    state = process_stacktrace(config, line1, False)
+    assert state is True
+
+    # Backtrace content with addresses
+    line2 = "40201234 40205678"
+    state = process_stacktrace(config, line2, state)
+    assert state is True
+    assert mock_esp8266_decode_pc.call_count == 2
+
+    # End of backtrace
+    line3 = "<<<stack<<<"
+    state = process_stacktrace(config, line3, state)
+    assert state is False
+
+
+def test_process_stacktrace_esp32_backtrace(
+    setup_core: Path, mock_esp32_decode_pc: Mock
+) -> None:
+    """Test process_stacktrace handles ESP32 single-line backtrace."""
+    from esphome.components.esp32 import process_stacktrace
+
+    config = {"name": "test"}
+
+    line = "Backtrace: 0x40081234:0x3ffb1234 0x40085678:0x3ffb5678"
+    state = process_stacktrace(config, line, False)
+
+    # Should decode both addresses
+    assert mock_esp32_decode_pc.call_count == 2
+    mock_esp32_decode_pc.assert_any_call(config, "40081234")
+    mock_esp32_decode_pc.assert_any_call(config, "40085678")
+    assert state is False
+
+
+def test_process_stacktrace_bad_alloc(
+    setup_core: Path, mock_esp32_decode_pc: Mock, caplog
+) -> None:
+    """Test process_stacktrace handles bad alloc messages."""
+    from esphome.components.esp32 import process_stacktrace
+
+    config = {"name": "test"}
+
+    line = "last failed alloc call: 40201234(512)"
+    state = process_stacktrace(config, line, False)
+
+    assert "Memory allocation of 512 bytes failed at 40201234" in caplog.text
+    mock_esp32_decode_pc.assert_called_once_with(config, "40201234")
+    assert state is False
+
+
+def test_process_stacktrace_esp32_crash_handler(
+    setup_core: Path, mock_esp32_decode_pc: Mock
+) -> None:
+    """Test process_stacktrace handles ESP32 crash handler backtrace lines."""
+    from esphome.components.esp32 import process_stacktrace
+
+    config = {"name": "test"}
+
+    # Simulate crash handler log lines as they appear from the API/serial
+    line_pc = "[E][esp32.crash:078]:   PC:  0x400D1234  (fault location)"
+    state = process_stacktrace(config, line_pc, False)
+    # PC line is matched by existing STACKTRACE_ESP32_PC_RE
+    mock_esp32_decode_pc.assert_called_with(config, "400D1234")
+    assert state is False
+
+    mock_esp32_decode_pc.reset_mock()
+
+    line_bt0 = "[E][esp32.crash:080]:   BT0: 0x400D5678  (backtrace)"
+    state = process_stacktrace(config, line_bt0, False)
+    mock_esp32_decode_pc.assert_called_once_with(config, "400D5678")
+    assert state is False
+
+    mock_esp32_decode_pc.reset_mock()
+
+    line_bt1 = "[E][esp32.crash:080]:   BT1: 0x42005ABC  (backtrace)"
+    state = process_stacktrace(config, line_bt1, False)
+    mock_esp32_decode_pc.assert_called_once_with(config, "42005ABC")
+    assert state is False

tests/unit_tests/conftest.py

@@ -77,9 +77,16 @@ def mock_run_platformio_cli_run() -> Generator[Mock, None, None]:
 
 
 @pytest.fixture
-def mock_decode_pc() -> Generator[Mock, None, None]:
-    """Mock _decode_pc for platformio_api."""
-    with patch("esphome.platformio_api._decode_pc") as mock:
+def mock_esp32_decode_pc() -> Generator[Mock, None, None]:
+    """Mock _decode_pc for esp32."""
+    with patch("esphome.components.esp32._decode_pc") as mock:
+        yield mock
+
+
+@pytest.fixture
+def mock_esp8266_decode_pc() -> Generator[Mock, None, None]:
+    """Mock _decode_pc for esp8266."""
+    with patch("esphome.components.esp8266._decode_pc") as mock:
         yield mock
 
 

tests/unit_tests/core/test_config.py

@@ -10,6 +10,7 @@ from unittest.mock import MagicMock, Mock, patch
 import pytest
 
 from esphome import config_validation as cv, core
+from esphome.components.safe_mode import to_code as safe_mode_to_code
 from esphome.const import (
     CONF_AREA,
     CONF_AREAS,
@@ -312,6 +313,75 @@ def test_add_platform_defines_priority() -> None:
     )
 
 
+def test_to_code_priority_above_safe_mode() -> None:
+    """Test that core to_code emits the looping_components_ init before safe_mode.
+
+    Regression test for https://github.com/esphome/esphome/issues/16262.
+    safe_mode emits an `if (should_enter_safe_mode(...)) return;` line in main()
+    at APPLICATION priority. The `App.looping_components_.init(...)` call must be
+    emitted at a higher priority than APPLICATION so it lands in main() before
+    the early return; otherwise the FixedVector is never sized when safe mode is
+    active and loop() never runs (Wi-Fi never connects).
+    """
+    assert config.to_code.priority > safe_mode_to_code.priority, (
+        f"core to_code priority ({config.to_code.priority}) must be greater than "
+        f"safe_mode to_code priority ({safe_mode_to_code.priority}) so that "
+        "App.looping_components_.init() is emitted before safe_mode's early return"
+    )
+
+
+@pytest.mark.asyncio
+async def test_add_looping_components_handles_empty_entries() -> None:
+    """Test that _add_looping_components emits a valid constexpr when there are
+    no looping component entries.
+
+    With zero entries the generated constexpr must still be syntactically valid
+    C++ (`= 0;`), not an empty expression (`= ;`). This guards the empty-list
+    case that would otherwise produce uncompilable main.cpp output.
+    """
+    CORE.data["looping_component_entries"] = []
+
+    await config._add_looping_components()
+
+    constexpr_lines = [
+        str(s)
+        for s in CORE.global_statements
+        if "ESPHOME_LOOPING_COMPONENT_COUNT" in str(s)
+    ]
+    assert len(constexpr_lines) == 1
+    text = constexpr_lines[0]
+    assert "static constexpr size_t ESPHOME_LOOPING_COMPONENT_COUNT" in text
+    # The right-hand side must contain a literal `0`, not be empty.
+    rhs = text.split("=", 1)[1]
+    assert "0" in rhs
+    assert rhs.strip().rstrip(";").strip(), (
+        f"constexpr right-hand side must not be empty, got: {text!r}"
+    )
+
+
+@pytest.mark.asyncio
+async def test_add_looping_components_with_entries() -> None:
+    """Test that _add_looping_components builds a HasLoopOverride sum from entries."""
+    CORE.data["looping_component_entries"] = [
+        "esphome::wifi::WiFiComponent",
+        "esphome::logger::Logger",
+        "esphome::wifi::WiFiComponent",
+    ]
+
+    await config._add_looping_components()
+
+    constexpr_lines = [
+        str(s)
+        for s in CORE.global_statements
+        if "ESPHOME_LOOPING_COMPONENT_COUNT" in str(s)
+    ]
+    assert len(constexpr_lines) == 1
+    text = constexpr_lines[0]
+    # Deduplicated by type, with per-type counts as multiplier.
+    assert "(2 * HasLoopOverride<esphome::wifi::WiFiComponent>::value)" in text
+    assert "(1 * HasLoopOverride<esphome::logger::Logger>::value)" in text
+
+
 def test_valid_include_with_angle_brackets() -> None:
     """Test valid_include accepts angle bracket includes."""
     assert valid_include("<ArduinoJson.h>") == "<ArduinoJson.h>"

tests/unit_tests/fixtures/substitutions/18-dict_self_reference.approved.yaml

@@ -0,0 +1,16 @@
+substitutions:
+  device:
+    manufacturer: espressif
+    model: esp32
+    mac_suffix: ffffff
+    name: espressif-esp32-ffffff
+  network:
+    host: example.com
+    port: 8080
+    url: http://example.com:8080/api
+esphome:
+  name: espressif-esp32-ffffff
+test_list:
+  - espressif-esp32-ffffff
+  - http://example.com:8080/api
+  - espressif/esp32

tests/unit_tests/fixtures/substitutions/18-dict_self_reference.input.yaml

@@ -0,0 +1,18 @@
+substitutions:
+  device:
+    manufacturer: "espressif"
+    model: "esp32"
+    mac_suffix: "ffffff"
+    name: ${device.manufacturer}-${device.model}-${device.mac_suffix}
+  network:
+    host: "example.com"
+    port: 8080
+    url: "http://${network.host}:${network.port}/api"
+
+esphome:
+  name: ${device.name}
+
+test_list:
+  - ${device.name}
+  - ${network.url}
+  - "${device.manufacturer}/${device.model}"

tests/unit_tests/fixtures/substitutions/18-package_vars_in_subs.approved.yaml

@@ -0,0 +1,9 @@
+binary_sensor:
+  - platform: template
+    id: front_door_enrolling
+    name: Front Door Enrolling
+substitutions:
+  enrolling_id: front_door_enrolling
+  enrolling_name: Front Door Enrolling
+esphome:
+  name: test

tests/unit_tests/fixtures/substitutions/18-package_vars_in_subs.input.yaml

@@ -0,0 +1,9 @@
+esphome:
+  name: test
+
+packages:
+  fingerprint: !include
+    file: 18-package_vars_in_subs_inc.yaml
+    vars:
+      sensor_name: "Front Door"
+      sensor_id_prefix: "front_door"

tests/unit_tests/fixtures/substitutions/18-package_vars_in_subs_inc.yaml

@@ -0,0 +1,8 @@
+substitutions:
+  enrolling_id: ${sensor_id_prefix}_enrolling
+  enrolling_name: ${sensor_name} Enrolling
+
+binary_sensor:
+  - platform: template
+    id: ${enrolling_id}
+    name: ${enrolling_name}

tests/unit_tests/test_bundle.py

@@ -170,6 +170,23 @@ def test_find_used_secret_keys_deduplicates(tmp_path: Path) -> None:
     assert keys == {"key1"}
 
 
+def test_find_used_secret_keys_quoted(tmp_path: Path) -> None:
+    """Quoted !secret keys should resolve to the same key as unquoted form.
+
+    YAML strips surrounding quotes during parsing, so the secrets.yaml
+    lookup uses the unquoted key. The bundle scan must do the same.
+    """
+    yaml1 = tmp_path / "a.yaml"
+    yaml1.write_text(
+        "single: !secret 'wifi_ssid'\n"
+        'double: !secret "wifi_pw"\n'
+        "bare: !secret api_key\n"
+    )
+
+    keys = _find_used_secret_keys([yaml1])
+    assert keys == {"wifi_ssid", "wifi_pw", "api_key"}
+
+
 # ---------------------------------------------------------------------------
 # _add_bytes_to_tar
 # ---------------------------------------------------------------------------
@@ -1217,6 +1234,35 @@ def test_create_bundle_filters_secrets(tmp_path: Path) -> None:
     assert "should_not_appear" not in secrets_data
 
 
+def test_create_bundle_filters_secrets_quoted(tmp_path: Path) -> None:
+    """Bundling must include secrets.yaml when !secret keys are quoted.
+
+    Regression test for issue 16259: quoted !secret references previously
+    captured the quotes as part of the key, so no key matched secrets.yaml
+    entries and the secrets file was dropped from the bundle entirely.
+    """
+    config_dir = _setup_config_dir(tmp_path)
+
+    secrets = config_dir / "secrets.yaml"
+    secrets.write_text("ota_password: hunter2\nunused: should_not_appear\n")
+
+    config_yaml = "ota:\n  password: !secret 'ota_password'\n"
+    (config_dir / "test.yaml").write_text(config_yaml)
+
+    creator = ConfigBundleCreator({})
+    result = creator.create_bundle()
+
+    assert result.manifest[ManifestKey.HAS_SECRETS] is True
+
+    buf = io.BytesIO(result.data)
+    with tarfile.open(fileobj=buf, mode="r:gz") as tar:
+        secrets_data = tar.extractfile("secrets.yaml").read().decode()
+
+    assert "ota_password" in secrets_data
+    assert "hunter2" in secrets_data
+    assert "unused" not in secrets_data
+
+
 def test_create_bundle_no_secrets(tmp_path: Path) -> None:
     _setup_config_dir(tmp_path)
 

tests/unit_tests/test_main.py

@@ -43,6 +43,7 @@ from esphome.__main__ import (
     has_non_ip_address,
     has_ota,
     has_resolvable_address,
+    has_web_server_ota,
     mqtt_get_ip,
     run_esphome,
     run_miniterm,
@@ -54,9 +55,11 @@ from esphome.__main__ import (
 )
 from esphome.address_cache import AddressCache
 from esphome.bundle import BUNDLE_EXTENSION, BundleFile, BundleResult
+from esphome.components import esp32
 from esphome.components.esp32 import KEY_ESP32, KEY_VARIANT, VARIANT_ESP32
 from esphome.const import (
     CONF_API,
+    CONF_AUTH,
     CONF_BAUD_RATE,
     CONF_BROKER,
     CONF_DISABLED,
@@ -75,6 +78,8 @@ from esphome.const import (
     CONF_SUBSTITUTIONS,
     CONF_TOPIC,
     CONF_USE_ADDRESS,
+    CONF_USERNAME,
+    CONF_WEB_SERVER,
     CONF_WIFI,
     KEY_CORE,
     KEY_TARGET_PLATFORM,
@@ -85,7 +90,7 @@ from esphome.const import (
 )
 from esphome.core import CORE, EsphomeError
 from esphome.espota2 import OTA_TYPE_UPDATE_APP, OTA_TYPE_UPDATE_PARTITION_TABLE
-from esphome.util import BootselResult
+from esphome.util import BootselResult, FlashImage
 from esphome.zeroconf import _await_discovery, discover_mdns_devices
 
 
@@ -212,6 +217,13 @@ def mock_run_ota() -> Generator[Mock]:
         yield mock
 
 
+@pytest.fixture
+def mock_run_web_server_ota() -> Generator[Mock]:
+    """Mock web_server_ota.run_ota for testing."""
+    with patch("esphome.web_server_ota.run_ota") as mock:
+        yield mock
+
+
 @pytest.fixture
 def mock_is_ip_address() -> Generator[Mock]:
     """Mock is_ip_address for testing."""
@@ -1113,6 +1125,7 @@ class MockArgs:
     reset: bool = False
     list_only: bool = False
     output: str | None = None
+    ota_platform: str | None = None
     partition_table: bool = False
 
 
@@ -1181,8 +1194,8 @@ def test_upload_using_esptool_path_conversion(
     mock_idedata = MagicMock(spec=platformio_api.IDEData)
     mock_idedata.firmware_bin_path = tmp_path / "firmware.bin"
     mock_idedata.extra_flash_images = [
-        platformio_api.FlashImage(path=tmp_path / "bootloader.bin", offset="0x1000"),
-        platformio_api.FlashImage(path=tmp_path / "partitions.bin", offset="0x8000"),
+        FlashImage(path=tmp_path / "bootloader.bin", offset="0x1000"),
+        FlashImage(path=tmp_path / "partitions.bin", offset="0x8000"),
     ]
 
     mock_get_idedata.return_value = mock_idedata
@@ -1259,8 +1272,8 @@ def test_upload_using_esptool_skips_missing_extra_flash_images(
     mock_idedata = MagicMock(spec=platformio_api.IDEData)
     mock_idedata.firmware_bin_path = tmp_path / "firmware.bin"
     mock_idedata.extra_flash_images = [
-        platformio_api.FlashImage(path=tmp_path / "bootloader.bin", offset="0x1000"),
-        platformio_api.FlashImage(path=missing_path, offset="0x2d0000"),
+        FlashImage(path=tmp_path / "bootloader.bin", offset="0x1000"),
+        FlashImage(path=missing_path, offset="0x2d0000"),
     ]
     mock_get_idedata.return_value = mock_idedata
 
@@ -1877,6 +1890,277 @@ def test_upload_program_ota_no_config(
         upload_program(config, args, devices)
 
 
+def test_has_web_server_ota_detects_platform() -> None:
+    """has_web_server_ota returns True when web_server OTA platform is configured."""
+    setup_core(
+        config={
+            CONF_OTA: [{CONF_PLATFORM: CONF_WEB_SERVER}],
+        }
+    )
+    assert has_web_server_ota() is True
+    assert has_ota() is True
+
+
+def test_has_web_server_ota_returns_false_without_config() -> None:
+    """has_web_server_ota returns False when only native OTA is configured."""
+    setup_core(
+        config={
+            CONF_OTA: [{CONF_PLATFORM: CONF_ESPHOME}],
+        }
+    )
+    assert has_web_server_ota() is False
+    assert has_ota() is True
+
+
+def test_upload_program_web_server_only_auto_dispatches(
+    mock_run_web_server_ota: Mock,
+    mock_run_ota: Mock,
+    mock_get_port_type: Mock,
+    tmp_path: Path,
+) -> None:
+    """When only web_server OTA is configured, upload_program picks it automatically."""
+    setup_core(platform=PLATFORM_ESP32, tmp_path=tmp_path)
+    mock_get_port_type.return_value = "NETWORK"
+    mock_run_web_server_ota.return_value = (0, "192.168.1.100")
+
+    config = {
+        CONF_OTA: [{CONF_PLATFORM: CONF_WEB_SERVER}],
+        CONF_WEB_SERVER: {
+            CONF_PORT: 80,
+            CONF_AUTH: {CONF_USERNAME: "admin", CONF_PASSWORD: "pw"},
+        },
+    }
+    args = MockArgs()
+    devices = ["192.168.1.100"]
+
+    exit_code, host = upload_program(config, args, devices)
+
+    assert exit_code == 0
+    assert host == "192.168.1.100"
+    expected_firmware = (
+        tmp_path / ".esphome" / "build" / "test" / ".pioenvs" / "test" / "firmware.bin"
+    )
+    mock_run_web_server_ota.assert_called_once_with(
+        ["192.168.1.100"], 80, "admin", "pw", expected_firmware
+    )
+    mock_run_ota.assert_not_called()
+
+
+def test_upload_program_web_server_no_auth(
+    mock_run_web_server_ota: Mock,
+    mock_get_port_type: Mock,
+    tmp_path: Path,
+) -> None:
+    """web_server OTA works without an auth block (passes None for credentials)."""
+    setup_core(platform=PLATFORM_ESP32, tmp_path=tmp_path)
+    mock_get_port_type.return_value = "NETWORK"
+    mock_run_web_server_ota.return_value = (0, "192.168.1.100")
+
+    config = {
+        CONF_OTA: [{CONF_PLATFORM: CONF_WEB_SERVER}],
+        CONF_WEB_SERVER: {CONF_PORT: 8080},
+    }
+    args = MockArgs()
+    devices = ["192.168.1.100"]
+
+    exit_code, host = upload_program(config, args, devices)
+
+    assert exit_code == 0
+    assert host == "192.168.1.100"
+    expected_firmware = (
+        tmp_path / ".esphome" / "build" / "test" / ".pioenvs" / "test" / "firmware.bin"
+    )
+    mock_run_web_server_ota.assert_called_once_with(
+        ["192.168.1.100"], 8080, None, None, expected_firmware
+    )
+
+
+def test_upload_program_both_platforms_default_prefers_native(
+    mock_run_ota: Mock,
+    mock_run_web_server_ota: Mock,
+    mock_get_port_type: Mock,
+    tmp_path: Path,
+) -> None:
+    """When both OTA platforms are configured, default selection is native API."""
+    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")
+
+    config = {
+        CONF_OTA: [
+            {
+                CONF_PLATFORM: CONF_ESPHOME,
+                CONF_PORT: 3232,
+                CONF_PASSWORD: "secret",
+            },
+            {CONF_PLATFORM: CONF_WEB_SERVER},
+        ],
+        CONF_WEB_SERVER: {CONF_PORT: 80},
+    }
+    args = MockArgs()
+    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()
+    mock_run_web_server_ota.assert_not_called()
+
+
+def test_upload_program_ota_platform_override_to_web_server(
+    mock_run_ota: Mock,
+    mock_run_web_server_ota: Mock,
+    mock_get_port_type: Mock,
+    tmp_path: Path,
+) -> None:
+    """--ota-platform web_server forces web_server OTA even when native is configured."""
+    setup_core(platform=PLATFORM_ESP32, tmp_path=tmp_path)
+    mock_get_port_type.return_value = "NETWORK"
+    mock_run_web_server_ota.return_value = (0, "192.168.1.100")
+
+    config = {
+        CONF_OTA: [
+            {
+                CONF_PLATFORM: CONF_ESPHOME,
+                CONF_PORT: 3232,
+                CONF_PASSWORD: "secret",
+            },
+            {CONF_PLATFORM: CONF_WEB_SERVER},
+        ],
+        CONF_WEB_SERVER: {CONF_PORT: 80},
+    }
+    args = MockArgs(ota_platform=CONF_WEB_SERVER)
+    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_not_called()
+    mock_run_web_server_ota.assert_called_once()
+
+
+def test_upload_program_ota_platform_unavailable(
+    mock_get_port_type: Mock,
+) -> None:
+    """--ota-platform must reference a platform that is actually configured."""
+    setup_core(platform=PLATFORM_ESP32)
+    mock_get_port_type.return_value = "NETWORK"
+
+    config = {
+        CONF_OTA: [
+            {
+                CONF_PLATFORM: CONF_ESPHOME,
+                CONF_PORT: 3232,
+                CONF_PASSWORD: "secret",
+            }
+        ],
+    }
+    args = MockArgs(ota_platform=CONF_WEB_SERVER)
+    devices = ["192.168.1.100"]
+
+    with pytest.raises(EsphomeError, match="--ota-platform web_server"):
+        upload_program(config, args, devices)
+
+
+def test_upload_program_web_server_missing_component(
+    mock_get_port_type: Mock,
+    tmp_path: Path,
+) -> None:
+    """web_server OTA without a web_server component fails with a clear error."""
+    setup_core(platform=PLATFORM_ESP32, tmp_path=tmp_path)
+    mock_get_port_type.return_value = "NETWORK"
+
+    config = {
+        CONF_OTA: [{CONF_PLATFORM: CONF_WEB_SERVER}],
+        # No CONF_WEB_SERVER
+    }
+    args = MockArgs()
+    devices = ["192.168.1.100"]
+
+    with pytest.raises(EsphomeError, match="web_server.*not configured"):
+        upload_program(config, args, devices)
+
+
+def test_upload_program_unrelated_ota_platform_ignored(
+    mock_run_ota: Mock,
+    mock_get_port_type: Mock,
+    tmp_path: Path,
+) -> None:
+    """OTA list entries that are neither esphome nor web_server are ignored.
+
+    Covers the false branch in _choose_ota_platform's filter loop and the
+    no-match branch in _upload_via_native_api's lookup loop.
+    """
+    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")
+
+    config = {
+        CONF_OTA: [
+            {CONF_PLATFORM: "http_request"},  # unrelated platform; ignored
+            {
+                CONF_PLATFORM: CONF_ESPHOME,
+                CONF_PORT: 3232,
+                CONF_PASSWORD: "secret",
+            },
+        ],
+    }
+    args = MockArgs()
+    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()
+
+
+def test_upload_program_duplicate_platform_dedup_in_error(
+    mock_get_port_type: Mock,
+    tmp_path: Path,
+) -> None:
+    """Duplicate same-platform OTA entries don't repeat in --ota-platform errors."""
+    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},
+            {CONF_PLATFORM: CONF_ESPHOME, CONF_PORT: 3233},
+        ],
+    }
+    args = MockArgs(ota_platform=CONF_WEB_SERVER)
+    devices = ["192.168.1.100"]
+
+    with pytest.raises(EsphomeError) as excinfo:
+        upload_program(config, args, devices)
+
+    # Error mentions esphome once in the platform list, not "esphome, esphome".
+    msg = str(excinfo.value)
+    assert "esphome, esphome" not in msg
+    assert msg.endswith(": esphome")
+
+
+def test_upload_program_only_unrelated_ota_platforms(
+    mock_get_port_type: Mock,
+    tmp_path: Path,
+) -> None:
+    """Only unrelated OTA platforms configured -> raises like missing OTA."""
+    setup_core(platform=PLATFORM_ESP32, tmp_path=tmp_path)
+    mock_get_port_type.return_value = "NETWORK"
+
+    config = {
+        CONF_OTA: [{CONF_PLATFORM: "http_request"}],
+    }
+    args = MockArgs()
+    devices = ["192.168.1.100"]
+
+    with pytest.raises(EsphomeError, match="Cannot upload Over the Air"):
+        upload_program(config, args, devices)
+
+
 def test_upload_program_ota_with_mqtt_resolution(
     mock_mqtt_get_ip: Mock,
     mock_is_ip_address: Mock,
@@ -4225,7 +4509,7 @@ def test_run_miniterm_batches_lines_with_same_timestamp(
 
     with (
         patch("serial.Serial", return_value=mock_serial),
-        patch.object(platformio_api, "process_stacktrace") as mock_bt,
+        patch.object(esp32, "process_stacktrace") as mock_bt,
     ):
         mock_bt.return_value = False
         result = run_miniterm(config, "/dev/ttyUSB0", args)
@@ -4264,7 +4548,7 @@ def test_run_miniterm_different_chunks_different_timestamps(
 
     with (
         patch("serial.Serial", return_value=mock_serial),
-        patch.object(platformio_api, "process_stacktrace") as mock_bt,
+        patch.object(esp32, "process_stacktrace") as mock_bt,
     ):
         mock_bt.return_value = False
         result = run_miniterm(config, "/dev/ttyUSB0", args)
@@ -4295,7 +4579,7 @@ def test_run_miniterm_handles_split_lines() -> None:
 
     with (
         patch("serial.Serial", return_value=mock_serial),
-        patch.object(platformio_api, "process_stacktrace") as mock_bt,
+        patch.object(esp32, "process_stacktrace") as mock_bt,
         patch("esphome.__main__.safe_print") as mock_print,
     ):
         mock_bt.return_value = False
@@ -4349,7 +4633,7 @@ def test_run_miniterm_backtrace_state_maintained() -> None:
     with (
         patch("serial.Serial", return_value=mock_serial),
         patch.object(
-            platformio_api,
+            esp32,
             "process_stacktrace",
             side_effect=track_backtrace_state,
         ),
@@ -4400,7 +4684,7 @@ def test_run_miniterm_handles_empty_reads(
 
     with (
         patch("serial.Serial", return_value=mock_serial),
-        patch.object(platformio_api, "process_stacktrace") as mock_bt,
+        patch.object(esp32, "process_stacktrace") as mock_bt,
     ):
         mock_bt.return_value = False
         result = run_miniterm(config, "/dev/ttyUSB0", args)
@@ -4473,7 +4757,7 @@ def test_run_miniterm_buffer_limit_prevents_unbounded_growth() -> None:
 
     with (
         patch("serial.Serial", return_value=mock_serial),
-        patch.object(platformio_api, "process_stacktrace") as mock_bt,
+        patch.object(esp32, "process_stacktrace") as mock_bt,
         patch("esphome.__main__.safe_print") as mock_print,
         patch("esphome.__main__.SERIAL_BUFFER_MAX_SIZE", test_buffer_limit),
     ):
@@ -4754,6 +5038,32 @@ def test_run_esphome_non_bundle_skips_extraction(tmp_path: Path) -> None:
     assert result == 2
 
 
+@pytest.mark.parametrize(
+    ("command", "expected_skip"),
+    [
+        ("logs", True),
+        ("clean", True),
+        ("compile", False),
+        ("config", False),
+        ("run", False),
+        ("clean-mqtt", False),
+    ],
+)
+def test_run_esphome_skip_external_update_per_command(
+    tmp_path: Path, command: str, expected_skip: bool
+) -> None:
+    """read_config is invoked with skip_external_update=True only for commands
+    that don't need fresh external components (logs, clean)."""
+    yaml_file = tmp_path / "device.yaml"
+    yaml_file.write_text("esphome:\n  name: test\n")
+
+    with patch("esphome.__main__.read_config", return_value=None) as mock_read:
+        run_esphome(["esphome", command, str(yaml_file)])
+
+    mock_read.assert_called_once()
+    assert mock_read.call_args.kwargs["skip_external_update"] is expected_skip
+
+
 def test_get_configured_xtal_freq_reads_sdkconfig(tmp_path: Path) -> None:
     """Test reading XTAL_FREQ from sdkconfig."""
     CORE.name = "test-device"

tests/unit_tests/test_platformio_api.py

@@ -13,6 +13,7 @@ import pytest
 
 from esphome import platformio_api, platformio_runner
 from esphome.core import CORE, EsphomeError
+from esphome.util import FlashImage
 
 
 def test_idedata_firmware_elf_path(setup_core: Path) -> None:
@@ -70,7 +71,7 @@ def test_idedata_extra_flash_images(setup_core: Path) -> None:
 
     images = idedata.extra_flash_images
     assert len(images) == 2
-    assert all(isinstance(img, platformio_api.FlashImage) for img in images)
+    assert all(isinstance(img, FlashImage) for img in images)
     assert images[0].path == Path("/path/to/bootloader.bin")
     assert images[0].offset == "0x1000"
     assert images[1].path == Path("/path/to/partition.bin")
@@ -106,7 +107,7 @@ def test_idedata_cc_path(setup_core: Path) -> None:
 
 def test_flash_image_dataclass() -> None:
     """Test FlashImage dataclass stores path and offset correctly."""
-    image = platformio_api.FlashImage(path=Path("/path/to/image.bin"), offset="0x10000")
+    image = FlashImage(path=Path("/path/to/image.bin"), offset="0x10000")
 
     assert image.path == Path("/path/to/image.bin")
     assert image.offset == "0x10000"
@@ -708,101 +709,6 @@ def test_patched_clean_build_dir_creates_missing(setup_core: Path) -> None:
         assert build_dir.exists()
 
 
-def test_process_stacktrace_esp8266_exception(setup_core: Path, caplog) -> None:
-    """Test process_stacktrace handles ESP8266 exceptions."""
-    config = {"name": "test"}
-
-    # Test exception type parsing
-    line = "Exception (28):"
-    backtrace_state = False
-
-    result = platformio_api.process_stacktrace(config, line, backtrace_state)
-
-    assert "Access to invalid address: LOAD (wild pointer?)" in caplog.text
-    assert result is False
-
-
-def test_process_stacktrace_esp8266_backtrace(
-    setup_core: Path, mock_decode_pc: Mock
-) -> None:
-    """Test process_stacktrace handles ESP8266 multi-line backtrace."""
-    config = {"name": "test"}
-
-    # Start of backtrace
-    line1 = ">>>stack>>>"
-    state = platformio_api.process_stacktrace(config, line1, False)
-    assert state is True
-
-    # Backtrace content with addresses
-    line2 = "40201234 40205678"
-    state = platformio_api.process_stacktrace(config, line2, state)
-    assert state is True
-    assert mock_decode_pc.call_count == 2
-
-    # End of backtrace
-    line3 = "<<<stack<<<"
-    state = platformio_api.process_stacktrace(config, line3, state)
-    assert state is False
-
-
-def test_process_stacktrace_esp32_backtrace(
-    setup_core: Path, mock_decode_pc: Mock
-) -> None:
-    """Test process_stacktrace handles ESP32 single-line backtrace."""
-    config = {"name": "test"}
-
-    line = "Backtrace: 0x40081234:0x3ffb1234 0x40085678:0x3ffb5678"
-    state = platformio_api.process_stacktrace(config, line, False)
-
-    # Should decode both addresses
-    assert mock_decode_pc.call_count == 2
-    mock_decode_pc.assert_any_call(config, "40081234")
-    mock_decode_pc.assert_any_call(config, "40085678")
-    assert state is False
-
-
-def test_process_stacktrace_bad_alloc(
-    setup_core: Path, mock_decode_pc: Mock, caplog
-) -> None:
-    """Test process_stacktrace handles bad alloc messages."""
-    config = {"name": "test"}
-
-    line = "last failed alloc call: 40201234(512)"
-    state = platformio_api.process_stacktrace(config, line, False)
-
-    assert "Memory allocation of 512 bytes failed at 40201234" in caplog.text
-    mock_decode_pc.assert_called_once_with(config, "40201234")
-    assert state is False
-
-
-def test_process_stacktrace_esp32_crash_handler(
-    setup_core: Path, mock_decode_pc: Mock
-) -> None:
-    """Test process_stacktrace handles ESP32 crash handler backtrace lines."""
-    config = {"name": "test"}
-
-    # Simulate crash handler log lines as they appear from the API/serial
-    line_pc = "[E][esp32.crash:078]:   PC:  0x400D1234  (fault location)"
-    state = platformio_api.process_stacktrace(config, line_pc, False)
-    # PC line is matched by existing STACKTRACE_ESP32_PC_RE
-    mock_decode_pc.assert_called_with(config, "400D1234")
-    assert state is False
-
-    mock_decode_pc.reset_mock()
-
-    line_bt0 = "[E][esp32.crash:080]:   BT0: 0x400D5678  (backtrace)"
-    state = platformio_api.process_stacktrace(config, line_bt0, False)
-    mock_decode_pc.assert_called_once_with(config, "400D5678")
-    assert state is False
-
-    mock_decode_pc.reset_mock()
-
-    line_bt1 = "[E][esp32.crash:080]:   BT1: 0x42005ABC  (backtrace)"
-    state = platformio_api.process_stacktrace(config, line_bt1, False)
-    mock_decode_pc.assert_called_once_with(config, "42005ABC")
-    assert state is False
-
-
 def test_patch_file_downloader_succeeds_first_try() -> None:
     """Test patch_file_downloader succeeds on first attempt."""
     mock_exception_cls = type("PackageException", (Exception,), {})

tests/unit_tests/test_util.py

@@ -709,3 +709,119 @@ def test_detect_rp2040_bootsel_timeout() -> None:
         result = util.detect_rp2040_bootsel("/usr/bin/picotool")
     assert result.device_count == 0
     assert result.permission_error is False
+
+
+class TestSafePrint:
+    """Tests for ``safe_print`` and its UnicodeEncodeError fallback chain."""
+
+    @pytest.fixture(autouse=True)
+    def _no_dashboard(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        """Default ``CORE.dashboard`` to False so each test starts hermetic."""
+        from esphome.core import CORE
+
+        monkeypatch.setattr(CORE, "dashboard", False)
+
+    def test_prints_plain_message(self, capsys: pytest.CaptureFixture[str]) -> None:
+        """ASCII-only messages take the fast path through native ``print``."""
+        util.safe_print("hello world")
+        assert capsys.readouterr().out == "hello world\n"
+
+    def test_prints_unicode_on_utf8_stdout(
+        self, capsys: pytest.CaptureFixture[str]
+    ) -> None:
+        """Non-ASCII goes straight through when stdout can encode it."""
+        util.safe_print("bars: \u2582\u2584\u2586\u2588")
+        assert capsys.readouterr().out == "bars: \u2582\u2584\u2586\u2588\n"
+
+    def test_dashboard_escapes_esc_byte(
+        self,
+        capsys: pytest.CaptureFixture[str],
+        monkeypatch: pytest.MonkeyPatch,
+    ) -> None:
+        r"""Dashboard mode escapes raw ``\033`` ESC bytes to literal ``\\033``."""
+        from esphome.core import CORE
+
+        monkeypatch.setattr(CORE, "dashboard", True)
+        util.safe_print("\033[0;32mhi\033[0m")
+        assert capsys.readouterr().out == "\\033[0;32mhi\\033[0m\n"
+
+    def test_fallback_writes_string_not_bytes_repr(
+        self, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        """Regression: cp1252 fallback must produce a printable str, not ``b'...'``.
+
+        On Windows, when stdout is a redirected pipe (e.g. the dashboard),
+        Python uses cp1252, which cannot encode the wifi signal-bar block
+        characters (U+2582..U+2588). The previous fallback path called
+        ``print(message.encode(...))`` with a ``bytes`` object, which
+        Python's ``print`` rendered as a literal ``b'...'`` repr — visible
+        in the user's dashboard output. The fix re-encodes through the
+        stream's encoding with ``backslashreplace`` and decodes back to
+        ``str``.
+        """
+        buf = io.BytesIO()
+        cp1252_stream = io.TextIOWrapper(buf, encoding="cp1252", errors="strict")
+        monkeypatch.setattr(sys, "stdout", cp1252_stream)
+
+        util.safe_print("bars: \u2582\u2584\u2586\u2588 done")
+        cp1252_stream.flush()
+        output = buf.getvalue().decode("cp1252")
+
+        # Output is a clean line, not the bytes repr.
+        assert not output.startswith("b'")
+        assert "b'bars" not in output
+        # Unencodable codepoints become readable backslash escapes.
+        assert "\\u2582\\u2584\\u2586\\u2588" in output
+        # Encodable parts survive unchanged.
+        assert "bars: " in output
+        assert " done" in output
+        assert output.endswith("\n")
+
+    def test_fallback_with_dashboard_escaped_message(
+        self, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        """Dashboard ESC escaping + cp1252 fallback compose correctly."""
+        from esphome.core import CORE
+
+        monkeypatch.setattr(CORE, "dashboard", True)
+        buf = io.BytesIO()
+        cp1252_stream = io.TextIOWrapper(buf, encoding="cp1252", errors="strict")
+        monkeypatch.setattr(sys, "stdout", cp1252_stream)
+
+        util.safe_print("\033[0;32m\u2582\u2584\u2586\u2588\033[0m")
+        cp1252_stream.flush()
+        output = buf.getvalue().decode("cp1252")
+
+        # Dashboard escaping turned ESC into literal "\033" (5 chars), which
+        # cp1252 can encode, so it survives the round-trip verbatim.
+        assert "\\033[0;32m" in output
+        assert "\\033[0m" in output
+        # Block characters became backslash escapes via backslashreplace.
+        assert "\\u2582\\u2584\\u2586\\u2588" in output
+
+    def test_final_message_when_locale_is_invalid(
+        self,
+        monkeypatch: pytest.MonkeyPatch,
+        capsys: pytest.CaptureFixture[str],
+    ) -> None:
+        """If every encoding path fails, surface the locale-error sentinel."""
+        original_print = print
+        call_count = 0
+
+        def fake_print(*args: Any, **kwargs: Any) -> None:
+            nonlocal call_count
+            call_count += 1
+            # The first three calls are: native print, stream-encoding
+            # fallback, ASCII fallback. Make all three raise so we reach
+            # the final sentinel "Cannot print line..." which is expected
+            # to succeed (no encoding required).
+            if call_count <= 3:
+                raise UnicodeEncodeError("ascii", "x", 0, 1, "boom")
+            original_print(*args, **kwargs)
+
+        monkeypatch.setattr("builtins.print", fake_print)
+        util.safe_print("x")
+        assert call_count == 4
+        assert (
+            capsys.readouterr().out == "Cannot print line because of invalid locale!\n"
+        )

tests/unit_tests/test_web_server_ota.py

@@ -0,0 +1,670 @@
+"""Unit tests for esphome.web_server_ota module."""
+
+from __future__ import annotations
+
+import io
+import logging
+from pathlib import Path
+import socket
+from unittest.mock import MagicMock, patch
+
+import pytest
+import requests
+from requests.auth import HTTPBasicAuth
+
+from esphome.core import CORE, EsphomeError
+from esphome.helpers import ProgressBar
+from esphome.web_server_ota import (
+    OTA_PATH,
+    WebServerOTAError,
+    _MultipartStreamer,
+    run_ota,
+)
+
+
+@pytest.fixture
+def firmware(tmp_path: Path) -> Path:
+    binary = tmp_path / "firmware.bin"
+    binary.write_bytes(b"\x00\x01\x02FIRMWARE\xff" * 64)
+    return binary
+
+
+def _make_response(status: int, body: str) -> MagicMock:
+    response = MagicMock(spec=requests.Response)
+    response.status_code = status
+    response.text = body
+    response.reason = ""
+    return response
+
+
+def _patch_resolve(
+    monkeypatch: pytest.MonkeyPatch, hosts: list[tuple[str, int]]
+) -> None:
+    """Replace resolve_ip_address so tests don't actually do DNS."""
+    addr_infos = [
+        (socket.AF_INET, socket.SOCK_STREAM, 0, "", (host, port))
+        for host, port in hosts
+    ]
+    monkeypatch.setattr(
+        "esphome.web_server_ota.resolve_ip_address", lambda *a, **kw: addr_infos
+    )
+
+
+# ---------------------------------------------------------------------------
+# _MultipartStreamer
+# ---------------------------------------------------------------------------
+
+
+def test_multipart_streamer_emits_full_body() -> None:
+    """Streaming the whole body in one call yields prefix + file + suffix."""
+    data = b"abcdef" * 100
+    streamer = _MultipartStreamer(io.BytesIO(data), len(data), "fw.bin")
+
+    body = streamer.read()
+    while True:
+        chunk = streamer.read()
+        if not chunk:
+            break
+        body += chunk
+
+    assert body.startswith(f"--{streamer.boundary}\r\n".encode())
+    assert b'name="update"' in body
+    assert b'filename="fw.bin"' in body
+    assert data in body
+    assert body.endswith(f"\r\n--{streamer.boundary}--\r\n".encode())
+
+
+def test_multipart_streamer_chunked_read_matches_full_read() -> None:
+    """Chunked reads (urllib3 calls read(8192) repeatedly) yield the same body."""
+    data = b"abcdef" * 1000  # 6000 bytes
+    full = _MultipartStreamer(io.BytesIO(data), len(data), "fw.bin").read()
+
+    streamed = bytearray()
+    s = _MultipartStreamer(io.BytesIO(data), len(data), "fw.bin")
+    # Same boundary lengths -> identical total length.
+    while True:
+        chunk = s.read(64)
+        if not chunk:
+            break
+        streamed += chunk
+    # Boundaries are random per instance, so compare lengths and structure.
+    assert len(streamed) == len(full)
+    assert streamed.startswith(f"--{s.boundary}\r\n".encode())
+    assert streamed.endswith(f"\r\n--{s.boundary}--\r\n".encode())
+
+
+def test_multipart_streamer_len_matches_emitted_bytes() -> None:
+    """``__len__`` is what urllib3 uses to set Content-Length, so it must
+    equal the total bytes emitted by ``read``."""
+    data = b"x" * 12345
+    s = _MultipartStreamer(io.BytesIO(data), len(data), "fw.bin")
+    declared = len(s)
+
+    emitted = 0
+    while True:
+        chunk = s.read(1024)
+        if not chunk:
+            break
+        emitted += len(chunk)
+
+    assert emitted == declared
+
+
+def test_multipart_streamer_progress_ticks_during_read() -> None:
+    """Each read advances the progress bar (this is the whole point of
+    streaming via ``data=``: progress reflects bytes leaving the host)."""
+    data = b"x" * 1000
+    s = _MultipartStreamer(io.BytesIO(data), len(data), "fw.bin")
+
+    updates: list[float] = []
+    s.progress.update = updates.append  # type: ignore[method-assign]
+
+    while True:
+        chunk = s.read(128)
+        if not chunk:
+            break
+
+    assert updates, "progress.update was never called"
+    # Strictly non-decreasing.
+    assert updates == sorted(updates)
+    # Final update reaches (within FP) 1.0 because all bytes were read.
+    assert updates[-1] == pytest.approx(1.0, abs=1e-9)
+
+
+def test_multipart_streamer_content_type_includes_boundary() -> None:
+    s = _MultipartStreamer(io.BytesIO(b""), 0, "fw.bin")
+    assert s.content_type == f"multipart/form-data; boundary={s.boundary}"
+
+
+def test_multipart_streamer_zero_size_file() -> None:
+    """A zero-byte file still produces a well-formed body and progress is
+    skipped (avoiding a divide-by-zero on the empty file segment)."""
+    s = _MultipartStreamer(io.BytesIO(b""), 0, "empty.bin")
+    body = b""
+    while True:
+        chunk = s.read(64)
+        if not chunk:
+            break
+        body += chunk
+    assert body.startswith(f"--{s.boundary}".encode())
+    assert body.endswith(f"--{s.boundary}--\r\n".encode())
+
+
+def test_multipart_streamer_unique_boundary_per_instance() -> None:
+    a = _MultipartStreamer(io.BytesIO(b""), 0, "a")
+    b = _MultipartStreamer(io.BytesIO(b""), 0, "a")
+    assert a.boundary != b.boundary
+
+
+def test_multipart_streamer_zero_size_read_returns_empty() -> None:
+    """``read(0)`` short-circuits without touching state."""
+    s = _MultipartStreamer(io.BytesIO(b"x" * 10), 10, "fw.bin")
+    assert s.read(0) == b""
+    # No bytes consumed.
+    assert s._sent == 0
+
+
+# ---------------------------------------------------------------------------
+# run_ota
+# ---------------------------------------------------------------------------
+
+
+def test_run_ota_success(monkeypatch: pytest.MonkeyPatch, firmware: Path) -> None:
+    _patch_resolve(monkeypatch, [("192.168.1.50", 80)])
+
+    with patch(
+        "esphome.web_server_ota.requests.post",
+        return_value=_make_response(200, "Update Successful!"),
+    ) as post:
+        exit_code, host = run_ota(["device.local"], 80, None, None, firmware)
+
+    assert exit_code == 0
+    assert host == "192.168.1.50"
+    post.assert_called_once()
+    args, kwargs = post.call_args
+    assert args == (f"http://192.168.1.50:80{OTA_PATH}",)
+    assert kwargs["auth"] is None
+    # Streaming body, not files=, so progress fires during transmission.
+    assert "files" not in kwargs
+    assert isinstance(kwargs["data"], _MultipartStreamer)
+    assert kwargs["headers"]["Content-Type"] == kwargs["data"].content_type
+    assert kwargs["headers"]["Connection"] == "close"
+
+
+def test_run_ota_logs_device_response_body(
+    monkeypatch: pytest.MonkeyPatch, firmware: Path, caplog: pytest.LogCaptureFixture
+) -> None:
+    """The device's HTTP response body is surfaced on success."""
+    _patch_resolve(monkeypatch, [("192.168.1.50", 80)])
+    caplog.set_level(logging.INFO, logger="esphome.web_server_ota")
+
+    with patch(
+        "esphome.web_server_ota.requests.post",
+        return_value=_make_response(200, "Update Successful!"),
+    ):
+        run_ota(["192.168.1.50"], 80, None, None, firmware)
+
+    assert "Device response: Update Successful!" in caplog.text
+    assert "OTA successful" in caplog.text
+
+
+def test_run_ota_log_says_via_web_server(
+    monkeypatch: pytest.MonkeyPatch, firmware: Path, caplog: pytest.LogCaptureFixture
+) -> None:
+    """The upload-start log line names the transport explicitly."""
+    _patch_resolve(monkeypatch, [("192.168.1.50", 80)])
+    caplog.set_level(logging.INFO, logger="esphome.web_server_ota")
+
+    with patch(
+        "esphome.web_server_ota.requests.post",
+        return_value=_make_response(200, "Update Successful!"),
+    ):
+        run_ota(["192.168.1.50"], 80, None, None, firmware)
+
+    assert "via web_server OTA" in caplog.text
+
+
+def test_run_ota_sends_basic_auth(
+    monkeypatch: pytest.MonkeyPatch, firmware: Path
+) -> None:
+    _patch_resolve(monkeypatch, [("192.168.1.50", 80)])
+
+    with patch(
+        "esphome.web_server_ota.requests.post",
+        return_value=_make_response(200, "Update Successful!"),
+    ) as post:
+        exit_code, _ = run_ota(["192.168.1.50"], 80, "admin", "secret", firmware)
+
+    assert exit_code == 0
+    auth = post.call_args.kwargs["auth"]
+    assert isinstance(auth, HTTPBasicAuth)
+    assert auth.username == "admin"
+    assert auth.password == "secret"
+
+
+def test_run_ota_skips_auth_when_no_credentials(
+    monkeypatch: pytest.MonkeyPatch, firmware: Path
+) -> None:
+    _patch_resolve(monkeypatch, [("192.168.1.50", 80)])
+
+    with patch(
+        "esphome.web_server_ota.requests.post",
+        return_value=_make_response(200, "Update Successful!"),
+    ) as post:
+        run_ota(["192.168.1.50"], 80, None, None, firmware)
+
+    assert post.call_args.kwargs["auth"] is None
+
+
+def test_run_ota_skips_auth_when_only_username(
+    monkeypatch: pytest.MonkeyPatch, firmware: Path
+) -> None:
+    """Both username and password are required to send Basic auth."""
+    _patch_resolve(monkeypatch, [("192.168.1.50", 80)])
+
+    with patch(
+        "esphome.web_server_ota.requests.post",
+        return_value=_make_response(200, "Update Successful!"),
+    ) as post:
+        run_ota(["192.168.1.50"], 80, "admin", None, firmware)
+
+    assert post.call_args.kwargs["auth"] is None
+
+
+def test_run_ota_uses_update_url(
+    monkeypatch: pytest.MonkeyPatch, firmware: Path
+) -> None:
+    _patch_resolve(monkeypatch, [("192.168.1.50", 8080)])
+
+    with patch(
+        "esphome.web_server_ota.requests.post",
+        return_value=_make_response(200, "Update Successful!"),
+    ) as post:
+        run_ota(["192.168.1.50"], 8080, None, None, firmware)
+
+    url = post.call_args.args[0]
+    assert url == f"http://192.168.1.50:8080{OTA_PATH}"
+    assert OTA_PATH == "/update"
+
+
+def test_run_ota_failure_response(
+    monkeypatch: pytest.MonkeyPatch, firmware: Path, caplog: pytest.LogCaptureFixture
+) -> None:
+    _patch_resolve(monkeypatch, [("192.168.1.50", 80)])
+
+    with patch(
+        "esphome.web_server_ota.requests.post",
+        return_value=_make_response(200, "Update Failed!"),
+    ):
+        exit_code, host = run_ota(["192.168.1.50"], 80, None, None, firmware)
+
+    assert exit_code == 1
+    assert host is None
+    assert "OTA failure" in caplog.text
+
+
+def test_run_ota_failure_response_empty_body(
+    monkeypatch: pytest.MonkeyPatch, firmware: Path, caplog: pytest.LogCaptureFixture
+) -> None:
+    _patch_resolve(monkeypatch, [("192.168.1.50", 80)])
+
+    with patch(
+        "esphome.web_server_ota.requests.post",
+        return_value=_make_response(200, ""),
+    ):
+        exit_code, host = run_ota(["192.168.1.50"], 80, None, None, firmware)
+
+    assert exit_code == 1
+    assert host is None
+    assert "no response body" in caplog.text
+
+
+def test_run_ota_auth_failed(
+    monkeypatch: pytest.MonkeyPatch, firmware: Path, caplog: pytest.LogCaptureFixture
+) -> None:
+    _patch_resolve(monkeypatch, [("192.168.1.50", 80)])
+
+    with patch(
+        "esphome.web_server_ota.requests.post",
+        return_value=_make_response(401, "Unauthorized"),
+    ):
+        exit_code, host = run_ota(["192.168.1.50"], 80, "user", "wrong", firmware)
+
+    assert exit_code == 1
+    assert host is None
+    assert "Authentication failed" in caplog.text
+
+
+def test_run_ota_unexpected_status_code(
+    monkeypatch: pytest.MonkeyPatch, firmware: Path, caplog: pytest.LogCaptureFixture
+) -> None:
+    _patch_resolve(monkeypatch, [("192.168.1.50", 80)])
+
+    with patch(
+        "esphome.web_server_ota.requests.post",
+        return_value=_make_response(500, "Internal Error"),
+    ):
+        exit_code, host = run_ota(["192.168.1.50"], 80, None, None, firmware)
+
+    assert exit_code == 1
+    assert host is None
+    assert "Unexpected HTTP 500" in caplog.text
+
+
+def test_run_ota_unexpected_status_empty_body_falls_back(
+    monkeypatch: pytest.MonkeyPatch, firmware: Path, caplog: pytest.LogCaptureFixture
+) -> None:
+    """Empty response body uses response.reason / a fallback in the error."""
+    _patch_resolve(monkeypatch, [("192.168.1.50", 80)])
+
+    response = _make_response(503, "")
+    response.reason = "Service Unavailable"
+
+    with patch(
+        "esphome.web_server_ota.requests.post",
+        return_value=response,
+    ):
+        exit_code, host = run_ota(["192.168.1.50"], 80, None, None, firmware)
+
+    assert exit_code == 1
+    assert host is None
+    assert "Service Unavailable" in caplog.text
+
+
+def test_run_ota_unexpected_status_no_body_no_reason(
+    monkeypatch: pytest.MonkeyPatch, firmware: Path, caplog: pytest.LogCaptureFixture
+) -> None:
+    """Empty body and empty reason still produce a usable error message."""
+    _patch_resolve(monkeypatch, [("192.168.1.50", 80)])
+
+    response = _make_response(599, "")
+    response.reason = ""
+
+    with patch(
+        "esphome.web_server_ota.requests.post",
+        return_value=response,
+    ):
+        run_ota(["192.168.1.50"], 80, None, None, firmware)
+
+    assert "no response body" in caplog.text
+
+
+def test_run_ota_connection_error_then_success(
+    monkeypatch: pytest.MonkeyPatch, firmware: Path
+) -> None:
+    """First resolved address fails to connect, second succeeds."""
+    _patch_resolve(
+        monkeypatch,
+        [("192.168.1.10", 80), ("192.168.1.50", 80)],
+    )
+
+    with patch(
+        "esphome.web_server_ota.requests.post",
+        side_effect=[
+            requests.ConnectionError("refused"),
+            _make_response(200, "Update Successful!"),
+        ],
+    ) as post:
+        exit_code, host = run_ota(["device.local"], 80, None, None, firmware)
+
+    assert exit_code == 0
+    assert host == "192.168.1.50"
+    assert post.call_count == 2
+
+
+def test_run_ota_request_exception_falls_through(
+    monkeypatch: pytest.MonkeyPatch, firmware: Path
+) -> None:
+    """A non-ConnectionError RequestException (e.g. timeout) falls through too."""
+    _patch_resolve(
+        monkeypatch,
+        [("192.168.1.10", 80), ("192.168.1.50", 80)],
+    )
+
+    with patch(
+        "esphome.web_server_ota.requests.post",
+        side_effect=[
+            requests.Timeout("read timeout"),
+            _make_response(200, "Update Successful!"),
+        ],
+    ):
+        exit_code, host = run_ota(["device.local"], 80, None, None, firmware)
+
+    assert exit_code == 0
+    assert host == "192.168.1.50"
+
+
+def test_run_ota_all_addresses_unreachable(
+    monkeypatch: pytest.MonkeyPatch, firmware: Path, caplog: pytest.LogCaptureFixture
+) -> None:
+    """When every resolved address fails to connect, run_ota returns failure."""
+    _patch_resolve(
+        monkeypatch,
+        [("192.168.1.10", 80), ("192.168.1.20", 80)],
+    )
+
+    with patch(
+        "esphome.web_server_ota.requests.post",
+        side_effect=requests.ConnectionError("refused"),
+    ):
+        exit_code, host = run_ota(["device.local"], 80, None, None, firmware)
+
+    assert exit_code == 1
+    assert host is None
+    # Per-address failure is logged for each attempt; final summary follows.
+    assert caplog.text.count("OTA upload to ") >= 2
+    assert "OTA upload failed." in caplog.text
+
+
+def test_run_ota_no_resolved_addresses(
+    monkeypatch: pytest.MonkeyPatch, firmware: Path, caplog: pytest.LogCaptureFixture
+) -> None:
+    """If resolve_ip_address returns no candidates, log and return failure."""
+    _patch_resolve(monkeypatch, [])
+
+    exit_code, host = run_ota(["192.168.1.50"], 80, None, None, firmware)
+
+    assert exit_code == 1
+    assert host is None
+    assert "Could not resolve 192.168.1.50" in caplog.text
+
+
+def test_run_ota_resolution_failure(
+    monkeypatch: pytest.MonkeyPatch, firmware: Path
+) -> None:
+    def _raise(*_args, **_kwargs):
+        raise EsphomeError("dns failed")
+
+    monkeypatch.setattr("esphome.web_server_ota.resolve_ip_address", _raise)
+
+    exit_code, host = run_ota(["does.not.exist"], 80, None, None, firmware)
+
+    assert exit_code == 1
+    assert host is None
+
+
+def test_run_ota_resolution_failure_dashboard_mode(
+    monkeypatch: pytest.MonkeyPatch, firmware: Path, caplog: pytest.LogCaptureFixture
+) -> None:
+    """Dashboard mode skips the '--device <IP>' tip on resolution failure."""
+
+    def _raise(*_args, **_kwargs):
+        raise EsphomeError("dns failed")
+
+    monkeypatch.setattr("esphome.web_server_ota.resolve_ip_address", _raise)
+    monkeypatch.setattr(CORE, "dashboard", True)
+    try:
+        exit_code, host = run_ota(["does.not.exist"], 80, None, None, firmware)
+    finally:
+        monkeypatch.setattr(CORE, "dashboard", False)
+
+    assert exit_code == 1
+    assert host is None
+    assert "--device <IP>" not in caplog.text
+
+
+def test_run_ota_empty_hosts(firmware: Path) -> None:
+    exit_code, host = run_ota([], 80, None, None, firmware)
+    assert exit_code == 1
+    assert host is None
+
+
+def test_run_ota_string_host_accepted(
+    monkeypatch: pytest.MonkeyPatch, firmware: Path
+) -> None:
+    """A bare string is accepted in addition to a list of hosts."""
+    _patch_resolve(monkeypatch, [("10.0.0.5", 80)])
+
+    with patch(
+        "esphome.web_server_ota.requests.post",
+        return_value=_make_response(200, "Update Successful!"),
+    ):
+        exit_code, host = run_ota("10.0.0.5", 80, None, None, firmware)
+
+    assert exit_code == 0
+    assert host == "10.0.0.5"
+
+
+def test_run_ota_multiple_hosts_first_fails(
+    monkeypatch: pytest.MonkeyPatch, firmware: Path
+) -> None:
+    """Multi-host fallthrough: first host's addresses all fail, second host wins."""
+    addr_lookup = {
+        "primary.local": [
+            (socket.AF_INET, socket.SOCK_STREAM, 0, "", ("192.168.1.10", 80)),
+        ],
+        "secondary.local": [
+            (socket.AF_INET, socket.SOCK_STREAM, 0, "", ("192.168.1.50", 80)),
+        ],
+    }
+
+    def _resolve(host, port, address_cache=None):  # noqa: ARG001
+        return addr_lookup[host]
+
+    monkeypatch.setattr("esphome.web_server_ota.resolve_ip_address", _resolve)
+
+    with patch(
+        "esphome.web_server_ota.requests.post",
+        side_effect=[
+            requests.ConnectionError("refused"),
+            _make_response(200, "Update Successful!"),
+        ],
+    ):
+        exit_code, host = run_ota(
+            ["primary.local", "secondary.local"], 80, None, None, firmware
+        )
+
+    assert exit_code == 0
+    assert host == "192.168.1.50"
+
+
+def test_run_ota_all_hosts_return_failure_no_exception(
+    monkeypatch: pytest.MonkeyPatch, firmware: Path, caplog: pytest.LogCaptureFixture
+) -> None:
+    """All hosts resolve to no addresses; run_ota cleanly returns failure."""
+    addr_lookup = {
+        "a.local": [],
+        "b.local": [],
+    }
+
+    def _resolve(host, port, address_cache=None):  # noqa: ARG001
+        return addr_lookup[host]
+
+    monkeypatch.setattr("esphome.web_server_ota.resolve_ip_address", _resolve)
+
+    exit_code, host = run_ota(["a.local", "b.local"], 80, None, None, firmware)
+
+    assert exit_code == 1
+    assert host is None
+    # Each host gets its own "Could not resolve" log line + final summary.
+    assert caplog.text.count("Could not resolve") == 2
+    assert "OTA upload failed." in caplog.text
+
+
+def test_web_server_ota_error_is_esphome_error() -> None:
+    assert issubclass(WebServerOTAError, EsphomeError)
+
+
+def test_run_ota_finalizes_progress_bar_on_success(
+    monkeypatch: pytest.MonkeyPatch, firmware: Path
+) -> None:
+    """progress.done() fires on the success path (finally block)."""
+    _patch_resolve(monkeypatch, [("192.168.1.50", 80)])
+
+    done_called: list[bool] = []
+
+    with (
+        patch(
+            "esphome.web_server_ota.requests.post",
+            return_value=_make_response(200, "Update Successful!"),
+        ),
+        patch.object(ProgressBar, "done", lambda self: done_called.append(True)),
+    ):
+        run_ota(["192.168.1.50"], 80, None, None, firmware)
+
+    assert done_called
+
+
+def test_run_ota_finalizes_progress_bar_on_failure(
+    monkeypatch: pytest.MonkeyPatch, firmware: Path
+) -> None:
+    """progress.done() fires when the request itself raises (finally block)."""
+    _patch_resolve(monkeypatch, [("192.168.1.50", 80)])
+
+    done_called: list[bool] = []
+
+    with (
+        patch(
+            "esphome.web_server_ota.requests.post",
+            side_effect=requests.ConnectionError("boom"),
+        ),
+        patch.object(ProgressBar, "done", lambda self: done_called.append(True)),
+    ):
+        run_ota(["192.168.1.50"], 80, None, None, firmware)
+
+    assert done_called
+
+
+def test_run_ota_ipv6_url_brackets_host(
+    monkeypatch: pytest.MonkeyPatch, firmware: Path
+) -> None:
+    """IPv6 candidates are bracketed in the URL so the port parses correctly."""
+    addr_infos = [
+        (socket.AF_INET6, socket.SOCK_STREAM, 0, "", ("2001:db8::1", 80, 0, 0)),
+    ]
+    monkeypatch.setattr(
+        "esphome.web_server_ota.resolve_ip_address", lambda *a, **kw: addr_infos
+    )
+
+    with patch(
+        "esphome.web_server_ota.requests.post",
+        return_value=_make_response(200, "Update Successful!"),
+    ) as post:
+        exit_code, host = run_ota(["device.local"], 80, None, None, firmware)
+
+    assert exit_code == 0
+    assert host == "2001:db8::1"
+    url = post.call_args.args[0]
+    assert url == f"http://[2001:db8::1]:80{OTA_PATH}"
+
+
+def test_run_ota_ipv6_link_local_includes_scope_id(
+    monkeypatch: pytest.MonkeyPatch, firmware: Path
+) -> None:
+    """Link-local IPv6 candidates include the percent-encoded zone index."""
+    addr_infos = [
+        (socket.AF_INET6, socket.SOCK_STREAM, 0, "", ("fe80::1", 80, 0, 3)),
+    ]
+    monkeypatch.setattr(
+        "esphome.web_server_ota.resolve_ip_address", lambda *a, **kw: addr_infos
+    )
+
+    with patch(
+        "esphome.web_server_ota.requests.post",
+        return_value=_make_response(200, "Update Successful!"),
+    ) as post:
+        exit_code, _ = run_ota(["device.local"], 80, None, None, firmware)
+
+    assert exit_code == 0
+    url = post.call_args.args[0]
+    assert url == f"http://[fe80::1%253]:80{OTA_PATH}"