Merge remote-tracking branch 'origin/dev' into web-server-offline-hint

tests/unit_tests/build_gen/test_build_tool.py

@@ -0,0 +1,246 @@
+"""Tests for the ninja build-tool helper script."""
+
+from __future__ import annotations
+
+from pathlib import Path
+import subprocess
+import sys
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from esphome.build_gen import build_tool
+
+
+def test_ar_removes_stale_archive(tmp_path: Path) -> None:
+    archive = tmp_path / "lib.a"
+    archive.write_text("stale")
+    rsp = tmp_path / "lib.a.rsp"
+    rsp.write_text("a.o\n")
+    with (
+        patch.object(
+            build_tool.sys,
+            "argv",
+            ["build_tool", "ar", "ar-bin", str(archive), str(rsp)],
+        ),
+        patch.object(
+            build_tool.subprocess, "run", return_value=MagicMock(returncode=0)
+        ) as mock_run,
+    ):
+        assert build_tool.main() == 0
+    assert not archive.exists()
+    # The rspfile is expanded by the shim (GNU ar would escape backslashes)
+    assert mock_run.call_args[0][0] == ["ar-bin", "rcs", str(archive), "a.o"]
+
+
+def test_copy(tmp_path: Path) -> None:
+    src = tmp_path / "firmware.bin"
+    src.write_text("data")
+    dst = tmp_path / "firmware.factory.bin"
+    with patch.object(
+        build_tool.sys, "argv", ["build_tool", "copy", str(src), str(dst)]
+    ):
+        assert build_tool.main() == 0
+    assert dst.read_text() == "data"
+
+
+def test_unknown_mode(capsys: pytest.CaptureFixture[str]) -> None:
+    with patch.object(build_tool.sys, "argv", ["build_tool", "bogus"]):
+        assert build_tool.main() == 1
+    assert "unknown build_tool mode" in capsys.readouterr().err
+
+
+def test_runs_as_script(tmp_path: Path) -> None:
+    """The ninja rules invoke the file as a plain script."""
+
+    src = tmp_path / "a.bin"
+    src.write_text("x")
+    dst = tmp_path / "b.bin"
+    result = subprocess.run(
+        [sys.executable, build_tool.__file__, "copy", str(src), str(dst)],
+        check=False,
+    )
+    assert result.returncode == 0
+    assert dst.read_text() == "x"
+
+
+def test_ar_expands_rspfile_without_escaping(tmp_path) -> None:
+    """Backslash paths survive: the shim expands the rspfile itself instead
+    of letting GNU ar treat backslashes as escapes."""
+    rsp = tmp_path / "objs.rsp"
+    rsp.write_text("obj/a.o\nsub\\b.o\n")
+    with (
+        patch.object(
+            build_tool.sys,
+            "argv",
+            ["build_tool", "ar", "ar-bin", str(tmp_path / "lib.a"), str(rsp)],
+        ),
+        patch.object(
+            build_tool.subprocess, "run", return_value=MagicMock(returncode=0)
+        ) as mock_run,
+    ):
+        assert build_tool.main() == 0
+    assert mock_run.call_args[0][0] == [
+        "ar-bin",
+        "rcs",
+        str(tmp_path / "lib.a"),
+        "obj/a.o",
+        "sub\\b.o",
+    ]
+
+
+def test_ar_unquotes_ninja_escaped_paths(tmp_path: Path) -> None:
+    """The shim strips a simple surrounding quote, since ninja shell-
+    quotes special rsp paths, so ar sees the real filename."""
+    rsp = tmp_path / "t.rsp"
+    rsp.write_text("'obj/a b.o'\nobj/c.o\n")
+    with (
+        patch.object(
+            build_tool.sys, "argv", ["bt", "ar", "/usr/bin/ar", "lib.a", str(rsp)]
+        ),
+        patch.object(build_tool.subprocess, "run") as mock_run,
+    ):
+        mock_run.return_value.returncode = 0
+        rc = build_tool.main()
+    assert rc == 0
+    assert mock_run.call_args.args[0] == [
+        "/usr/bin/ar",
+        "rcs",
+        "lib.a",
+        "obj/a b.o",
+        "obj/c.o",
+    ]
+
+
+def test_ar_empty_object_list_fails(
+    tmp_path: Path, capsys: pytest.CaptureFixture[str]
+) -> None:
+    """A lost object list is an error here, not undefined symbols at link."""
+    rsp = tmp_path / "t.rsp"
+    rsp.write_text("\n\n")
+    with patch.object(
+        build_tool.sys, "argv", ["bt", "ar", "/usr/bin/ar", "lib.a", str(rsp)]
+    ):
+        rc = build_tool.main()
+    assert rc == 1
+    assert "no objects listed" in capsys.readouterr().err
+
+
+def test_ar_batches_long_object_lists(tmp_path: Path) -> None:
+    """The expanded argv must stay under the Windows 32767-char limit: a
+    long object list creates with rcs, then appends with qs."""
+    archive = tmp_path / "lib.a"
+    rsp = tmp_path / "lib.a.rsp"
+    objects = [f"dir/{'x' * 120}_{i}.o" for i in range(400)]
+    rsp.write_text("\n".join(objects) + "\n")
+    with (
+        patch.object(
+            build_tool.sys,
+            "argv",
+            ["build_tool", "ar", "ar-bin", str(archive), str(rsp)],
+        ),
+        patch.object(
+            build_tool.subprocess, "run", return_value=MagicMock(returncode=0)
+        ) as mock_run,
+    ):
+        assert build_tool.main() == 0
+    calls = [c[0][0] for c in mock_run.call_args_list]
+    assert len(calls) > 1
+    assert calls[0][1] == "rcs"
+    assert all(c[1] == "qs" for c in calls[1:])
+    assert [o for c in calls for o in c[3:]] == objects
+    assert all(sum(len(a) + 1 for a in c) < 32000 for c in calls)
+
+
+def test_ar_batch_failure_stops(tmp_path: Path) -> None:
+    """A failing batch propagates its exit code without running the rest."""
+    archive = tmp_path / "lib.a"
+    rsp = tmp_path / "lib.a.rsp"
+    rsp.write_text("\n".join(f"{'y' * 200}_{i}.o" for i in range(300)) + "\n")
+    with (
+        patch.object(
+            build_tool.sys,
+            "argv",
+            ["build_tool", "ar", "ar-bin", str(archive), str(rsp)],
+        ),
+        patch.object(
+            build_tool.subprocess,
+            "run",
+            side_effect=lambda cmd, **kw: (
+                archive.write_text("partial"),
+                MagicMock(returncode=3),
+            )[1],
+        ) as mock_run,
+    ):
+        assert build_tool.main() == 3
+    assert mock_run.call_count == 1
+    # The failed batch must not leave a truncated archive behind
+    assert not archive.exists()
+
+
+def test_ar_exception_leaves_no_partial_archive(tmp_path: Path) -> None:
+    """A missing ar binary mid-loop must not leave a truncated archive from
+    earlier successful batches."""
+    archive = tmp_path / "lib.a"
+    rsp = tmp_path / "lib.a.rsp"
+    rsp.write_text("a.o\n")
+    with (
+        patch.object(
+            build_tool.sys,
+            "argv",
+            ["build_tool", "ar", "ar-bin", str(archive), str(rsp)],
+        ),
+        patch.object(
+            build_tool.subprocess,
+            "run",
+            side_effect=lambda cmd, **kw: (
+                archive.write_text("partial"),
+                (_ for _ in ()).throw(FileNotFoundError("no ar")),
+            ),
+        ),
+        pytest.raises(FileNotFoundError),
+    ):
+        build_tool.main()
+    assert not archive.exists()
+
+
+def test_surplus_arguments_error(capsys: pytest.CaptureFixture[str]) -> None:
+    """A mis-specified ninja rule passing extra operands errors instead of
+    silently dropping them."""
+    with patch.object(
+        build_tool.sys, "argv", ["build_tool", "copy", "a", "b", "extra"]
+    ):
+        assert build_tool.main() == 1
+    assert "expected 2 arguments, got 3" in capsys.readouterr().err
+
+
+def test_copy_same_file_keeps_the_input(
+    tmp_path: Path, capsys: pytest.CaptureFixture[str]
+) -> None:
+    """A same-file copy (dst IS src) must not unlink the input, and fails
+    with a message and exit code like the other shim paths."""
+    src = tmp_path / "firmware.bin"
+    src.write_bytes(b"image")
+    with patch.object(
+        build_tool.sys, "argv", ["build_tool", "copy", str(src), str(src)]
+    ):
+        assert build_tool.main() == 1
+    assert src.read_bytes() == b"image"
+    assert "failed" in capsys.readouterr().err
+
+
+def test_copy_failure_leaves_no_partial_output(tmp_path: Path) -> None:
+    """A failed copy unlinks the destination; a partial firmware image must
+    never be left on disk."""
+    dst = tmp_path / "firmware.factory.bin"
+    dst.write_text("stale")
+    with (
+        patch.object(build_tool.shutil, "copyfile", side_effect=OSError("disk full")),
+        patch.object(
+            build_tool.sys,
+            "argv",
+            ["build_tool", "copy", str(tmp_path / "src.bin"), str(dst)],
+        ),
+    ):
+        assert build_tool.main() == 1
+    assert not dst.exists()

tests/unit_tests/components/esp32/test_sdkconfig.py

@@ -0,0 +1,72 @@
+"""Tests for the esp32 sdkconfig write and its toolchain-gated clean."""
+
+from __future__ import annotations
+
+import os
+from pathlib import Path
+import time
+from unittest.mock import patch
+
+import pytest
+
+from esphome.components.esp32 import _write_sdkconfig
+from esphome.components.esp32.const import KEY_SDKCONFIG_OPTIONS
+from esphome.const import KEY_CORE, KEY_ESP32, KEY_FRAMEWORK_VERSION, Toolchain
+from esphome.core import CORE
+from esphome.espidf.toolchain import has_outdated_files
+
+
+def _setup_core(tmp_path: Path, toolchain: Toolchain | None) -> None:
+    CORE.config_path = tmp_path / "test.yaml"
+    CORE.build_path = tmp_path
+    CORE.toolchain = toolchain
+    CORE.data[KEY_ESP32] = {KEY_SDKCONFIG_OPTIONS: {"CONFIG_X": "y"}}
+    CORE.data[KEY_CORE] = {KEY_FRAMEWORK_VERSION: "5.5.5"}
+
+
+def _seed_configured_build(tmp_path: Path) -> None:
+    """A settled native build: configure outputs predate what comes next."""
+    build = tmp_path / "build"
+    (build / "config").mkdir(parents=True)
+    (build / "config" / "sdkconfig.h").write_text("")
+    (build / "CMakeCache.txt").write_text("")
+    (build / "build.ninja").write_text("")
+    # Explicitly older than what the test writes next: has_outdated_files()
+    # compares st_mtime with a strict >, so same-tick writes would pass
+    past = time.time() - 60
+    for f in build.rglob("*"):
+        os.utime(f, (past, past))
+
+
+@pytest.mark.parametrize(
+    ("toolchain", "clean_expected"),
+    [(Toolchain.ESP_IDF, False), (Toolchain.PLATFORMIO, True), (None, True)],
+)
+def test_write_sdkconfig_cleans_only_on_platformio(
+    tmp_path: Path, toolchain: Toolchain | None, clean_expected: bool
+) -> None:
+    """A changed sdkconfig forces a full clean only under PlatformIO; the
+    esp-idf toolchain reconfigures via has_outdated_files() instead; an
+    unresolved toolchain fails safe onto the clean."""
+    _setup_core(tmp_path, toolchain)
+    _seed_configured_build(tmp_path)
+    with (
+        patch.object(CORE, "name", "test"),
+        patch("esphome.components.esp32.clean_build") as clean,
+    ):
+        _write_sdkconfig()
+        assert "CONFIG_X" in CORE.relative_build_path("sdkconfig.test").read_text()
+        assert clean.called is clean_expected
+        if clean_expected:
+            clean.assert_called_once_with(clear_pio_cache=False)
+        # The change must still trigger a reconfigure: the internal
+        # sdkconfig snapshot is now newer than build/CMakeCache.txt
+        assert has_outdated_files() is True
+        clean.reset_mock()
+        # A settled configure restamps the cache; an unchanged rewrite
+        # must then neither clean nor mark the build stale
+        future = time.time() + 60
+        os.utime(CORE.relative_build_path("build/CMakeCache.txt"), (future, future))
+        _write_sdkconfig()
+        clean.assert_not_called()
+        assert has_outdated_files() is False

tests/unit_tests/components/light/test_gamma_table.py

@@ -53,9 +53,12 @@ def test_nonzero_indices_are_nonzero(gamma: float) -> None:
         assert table[i] >= 1, f"gamma={gamma}, index {i}: got {table[i]}"
 
 
-@pytest.mark.parametrize("gamma", [1.0, 2.0, 2.2, 2.8, 3.0])
+@pytest.mark.parametrize("gamma", [1.0, 1.8, 2.0, 2.2, 2.8, 3.0, 4.0])
 def test_table_monotonically_nondecreasing(gamma: float) -> None:
-    """The gamma table must be monotonically non-decreasing."""
+    """The gamma table must be monotonically non-decreasing.
+
+    gamma_table_reverse_search()'s binary search depends on this.
+    """
     table = generate_gamma_table(gamma)
     for i in range(1, 256):
         assert table[i] >= table[i - 1], (
@@ -115,3 +118,13 @@ def test_lut_output_monotonically_nondecreasing() -> None:
         result = _simulate_gamma_correct_lut(table, value)
         assert result >= prev, f"value={value}: result {result} < previous {prev}"
         prev = result
+
+
+def test_table_matches_raw_power_curve() -> None:
+    """Check the gamma table against known good values for gamma=2.8."""
+    table = generate_gamma_table(2.8)
+    golden = {1: 1, 5: 1, 15: 24, 27: 122, 28: 135, 100: 4766, 200: 33193, 254: 64818}
+    for i, expected in golden.items():
+        assert table[i] == expected, (
+            f"index {i}: table[{i}]={table[i]} expected {expected}"
+        )

tests/unit_tests/core/test_config.py

@@ -3,7 +3,6 @@
 from collections.abc import Callable
 import os
 from pathlib import Path
-import types
 from typing import Any
 from unittest.mock import MagicMock, Mock, patch
 
@@ -705,33 +704,6 @@ def test_include_file_with_c_header(
         assert '#include "c_library.h"' in mock_raw_statement.text
 
 
-def test_get_usable_cpu_count() -> None:
-    """Test get_usable_cpu_count returns CPU count."""
-    count = config.get_usable_cpu_count()
-    assert isinstance(count, int)
-    assert count > 0
-
-
-def test_get_usable_cpu_count_with_process_cpu_count() -> None:
-    """Test get_usable_cpu_count uses process_cpu_count when available."""
-    # Test with process_cpu_count (Python 3.13+)
-    # Create a mock os module with process_cpu_count
-
-    mock_os = types.SimpleNamespace(process_cpu_count=lambda: 8, cpu_count=lambda: 4)
-
-    with patch("esphome.core.config.os", mock_os):
-        # When process_cpu_count exists, it should be used
-        count = config.get_usable_cpu_count()
-        assert count == 8
-
-    # Test fallback to cpu_count when process_cpu_count not available
-    mock_os_no_process = types.SimpleNamespace(cpu_count=lambda: 4)
-
-    with patch("esphome.core.config.os", mock_os_no_process):
-        count = config.get_usable_cpu_count()
-        assert count == 4
-
-
 def test_list_target_platforms(tmp_path: Path) -> None:
     """Test _list_target_platforms returns available platforms."""
     # Create mock components directory structure
@@ -1155,6 +1127,34 @@ def test_config_hash_same_for_different_config_dirs(tmp_path: Path) -> None:
     assert hash1 == hash2
 
 
+def test_config_hash_same_for_different_data_dirs(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """Test that downloaded file paths hash the same wherever data_dir lives."""
+    config_dir = tmp_path / "config"
+    config_dir.mkdir()
+
+    CORE.reset()
+    CORE.config_path = config_dir / "device.yaml"
+    CORE.config = {
+        "esphome": {"name": "test"},
+        "file": config_dir / ".esphome" / "image" / "c44630d6",
+    }
+    hash1 = CORE.config_hash
+
+    other_data_dir = tmp_path / "data"
+    CORE.reset()
+    monkeypatch.setenv("ESPHOME_DATA_DIR", str(other_data_dir))
+    CORE.config_path = config_dir / "device.yaml"
+    CORE.config = {
+        "esphome": {"name": "test"},
+        "file": other_data_dir / "image" / "c44630d6",
+    }
+    hash2 = CORE.config_hash
+
+    assert hash1 == hash2
+
+
 def test_make_app_name_cpp_no_mac_simple() -> None:
     """Test simple name without MAC suffix returns string literal."""
     cpp_expr, global_decl, byte_len = make_app_name_cpp(

tests/unit_tests/test_espidf_framework.py

@@ -520,6 +520,33 @@ def test_check_esp_idf_install_feature_failure(espidf_mocks: SimpleNamespace) ->
         check_esp_idf_install(_IDF_VERSION, force=True, features=["fb"])
 
 
+def test_python_deps_use_uv_when_available(
+    espidf_mocks: SimpleNamespace, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """The python env installs go through uv when on the PATH, pip otherwise."""
+    monkeypatch.delenv("UV_HTTP_RETRIES", raising=False)
+    with patch(
+        "esphome.espidf.framework.shutil.which",
+        # Keyed on the name: the same which() also probes the default tools
+        side_effect=lambda name: "/usr/bin/uv" if name == "uv" else None,
+    ):
+        check_esp_idf_install(_IDF_VERSION, force=True, features=["fb"])
+    upgrade_call, feature_call = espidf_mocks.run_ok.call_args_list[1:3]
+    upgrade_cmd, feature_cmd = upgrade_call.args[0], feature_call.args[0]
+    assert upgrade_cmd[:3] == ["/usr/bin/uv", "pip", "install"]
+    assert "--python" in upgrade_cmd
+    assert feature_cmd[:3] == ["/usr/bin/uv", "pip", "install"]
+    assert upgrade_call.kwargs["env"]["UV_HTTP_RETRIES"] == "10"
+
+    espidf_mocks.run_ok.reset_mock()
+    monkeypatch.setenv("UV_HTTP_RETRIES", "3")  # an explicit user value wins
+    with patch("esphome.espidf.framework.shutil.which", return_value=None):
+        check_esp_idf_install(_IDF_VERSION, force=True, features=["fb"])
+    upgrade_call = espidf_mocks.run_ok.call_args_list[1]
+    assert upgrade_call.args[0][1:4] == ["-m", "pip", "install"]
+    assert upgrade_call.kwargs["env"]["UV_HTTP_RETRIES"] == "3"
+
+
 def _mark_installed() -> None:
     """Create the extracted marker and python-env interpreter so the install
     check takes the already-installed path rather than force-installing."""
@@ -911,7 +938,7 @@ def test_prefetch_leaves_unverifiable_entries_to_the_installer(
             "esphome.espidf.framework.run_command",
             return_value=(True, json.dumps(entries), ""),
         ),
-        patch("esphome.espidf.framework.download_with_resume") as download,
+        patch("esphome.framework_helpers.download_with_resume") as download,
         patch("esphome.espidf.framework.get_system_python_path", return_value="python"),
         patch("esphome.framework_helpers._BatchDownloadProgress") as progress_cls,
     ):
@@ -934,7 +961,7 @@ def test_prefetch_all_entries_unverifiable_is_a_noop(tmp_path: Path) -> None:
             "esphome.espidf.framework.run_command",
             return_value=(True, json.dumps(entries), ""),
         ),
-        patch("esphome.espidf.framework.download_with_resume") as download,
+        patch("esphome.framework_helpers.download_with_resume") as download,
         patch("esphome.espidf.framework.get_system_python_path", return_value="python"),
     ):
         _prefetch_idf_tool_archives(tmp_path, "esp32", ["required"], None)
@@ -952,7 +979,7 @@ def test_prefetch_dedupes_entries_by_dest(tmp_path: Path) -> None:
             "esphome.espidf.framework.run_command",
             return_value=(True, json.dumps(entries), ""),
         ),
-        patch("esphome.espidf.framework.download_with_resume") as download,
+        patch("esphome.framework_helpers.download_with_resume") as download,
         patch("esphome.espidf.framework.get_system_python_path", return_value="python"),
         patch("esphome.framework_helpers._BatchDownloadProgress"),
     ):
@@ -967,7 +994,7 @@ def test_prefetch_downloads_each_archive_with_resume(tmp_path: Path) -> None:
             "esphome.espidf.framework.run_command",
             return_value=(True, _PREFETCH_JSON, ""),
         ),
-        patch("esphome.espidf.framework.download_with_resume") as download,
+        patch("esphome.framework_helpers.download_with_resume") as download,
         patch("esphome.espidf.framework.get_system_python_path", return_value="python"),
         patch("esphome.framework_helpers._BatchDownloadProgress") as progress_cls,
     ):
@@ -1011,7 +1038,7 @@ def test_prefetch_downloads_archives_concurrently(tmp_path: Path) -> None:
             "esphome.espidf.framework.run_command",
             return_value=(True, json.dumps(entries), ""),
         ),
-        patch("esphome.espidf.framework.download_with_resume") as download,
+        patch("esphome.framework_helpers.download_with_resume") as download,
         patch("esphome.espidf.framework.get_system_python_path", return_value="python"),
         patch(
             "esphome.framework_helpers.ThreadPoolExecutor", wraps=ThreadPoolExecutor
@@ -1032,7 +1059,7 @@ def test_prefetch_skips_already_downloaded_archives(tmp_path: Path) -> None:
             "esphome.espidf.framework.run_command",
             return_value=(True, _PREFETCH_JSON, ""),
         ),
-        patch("esphome.espidf.framework.download_with_resume") as download,
+        patch("esphome.framework_helpers.download_with_resume") as download,
         patch("esphome.espidf.framework.get_system_python_path", return_value="python"),
     ):
         _prefetch_idf_tool_archives(tmp_path, "esp32", ["required"], None)
@@ -1065,7 +1092,7 @@ def test_prefetch_failures_never_raise(
     with (
         patch("esphome.espidf.framework.run_command", return_value=run_result),
         patch(
-            "esphome.espidf.framework.download_with_resume",
+            "esphome.framework_helpers.download_with_resume",
             side_effect=download_error,
         ),
         patch("esphome.espidf.framework.get_system_python_path", return_value="python"),
@@ -1087,7 +1114,7 @@ def test_prefetch_total_failure_logs_error(
             return_value=(True, _PREFETCH_JSON, ""),
         ),
         patch(
-            "esphome.espidf.framework.download_with_resume",
+            "esphome.framework_helpers.download_with_resume",
             side_effect=OSError("proxy refuses everything"),
         ),
         patch("esphome.espidf.framework.get_system_python_path", return_value="python"),
@@ -1112,7 +1139,7 @@ def test_prefetch_one_failed_archive_does_not_stop_the_rest(
             return_value=(True, _PREFETCH_JSON, ""),
         ),
         patch(
-            "esphome.espidf.framework.download_with_resume",
+            "esphome.framework_helpers.download_with_resume",
             side_effect=_fail_cmake_download,
         ) as download,
         patch("esphome.espidf.framework.get_system_python_path", return_value="python"),
@@ -1133,7 +1160,7 @@ def test_prefetch_finishes_progress_bar_and_cancels_queue(tmp_path: Path) -> Non
             "esphome.espidf.framework.run_command",
             return_value=(True, _PREFETCH_JSON, ""),
         ),
-        patch("esphome.espidf.framework.download_with_resume"),
+        patch("esphome.framework_helpers.download_with_resume"),
         patch("esphome.espidf.framework.get_system_python_path", return_value="python"),
         patch("esphome.framework_helpers._BatchDownloadProgress") as progress_cls,
         patch("esphome.framework_helpers.ThreadPoolExecutor") as pool_cls,

tests/unit_tests/test_espota2_noise.py

@@ -0,0 +1,407 @@
+"""Unit tests for encrypted OTA uploads in esphome.espota2.
+
+A fake device implementing the responder side of the wire protocol (via
+noiseprotocol, which esphome already has through aioesphomeapi) serves a real
+TCP loopback connection, so these exercise the actual handshake, framing, and
+cipher interop of the client code. Tests that need the client-side crypto skip
+when the installed aioesphomeapi predates the noise module.
+"""
+
+from __future__ import annotations
+
+import base64
+import hashlib
+import io
+from pathlib import Path
+import socket
+import sys
+import threading
+from unittest.mock import Mock, patch
+
+import pytest
+
+from esphome import espota2
+
+PSK = base64.b64encode(bytes(range(32))).decode()
+OTHER_PSK = base64.b64encode(bytes(range(1, 33))).decode()
+
+MAGIC = bytes(espota2.MAGIC_BYTES)
+
+
+def _recv_exact(sock: socket.socket, amount: int) -> bytes:
+    data = b""
+    while len(data) < amount:
+        chunk = sock.recv(amount - len(data))
+        if not chunk:
+            raise ConnectionError("client closed")
+        data += chunk
+    return data
+
+
+def _frame(payload: bytes) -> bytes:
+    return (
+        bytes([espota2.NOISE_FRAME_INDICATOR, len(payload) >> 8, len(payload) & 0xFF])
+        + payload
+    )
+
+
+def _send_frame(sock: socket.socket, payload: bytes) -> None:
+    sock.sendall(_frame(payload))
+
+
+def _recv_frame(sock: socket.socket) -> bytes:
+    header = _recv_exact(sock, 3)
+    assert header[0] == 0x01
+    return _recv_exact(sock, (header[1] << 8) | header[2])
+
+
+class FakeEncryptedDevice(threading.Thread):
+    """Responder side of the encrypted OTA wire protocol."""
+
+    def __init__(
+        self,
+        psk: str = PSK,
+        version: int = 2,
+        offer_noise: bool = True,
+        require_noise: bool = True,
+        prologue_features_override: int | None = None,
+    ) -> None:
+        super().__init__(daemon=True)
+        self.psk = psk
+        self.version = version
+        self.offer_noise = offer_noise
+        self.require_noise = require_noise
+        self.prologue_features_override = prologue_features_override
+        self.received: bytes | None = None
+        self.error: Exception | None = None
+        self.listener = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+        self.listener.bind(("127.0.0.1", 0))
+        self.listener.listen(1)
+        self.port = self.listener.getsockname()[1]
+
+    def run(self) -> None:
+        try:
+            sock, _ = self.listener.accept()
+            sock.settimeout(10)
+            with sock:
+                self._serve(sock)
+        except Exception as err:  # noqa: BLE001 - surfaced via join_and_check
+            self.error = err
+        finally:
+            self.listener.close()
+
+    def join_and_check(self) -> None:
+        self.join(timeout=10)
+        assert not self.is_alive(), "fake device did not finish"
+        if self.error is not None:
+            raise self.error
+
+    def _serve(self, sock: socket.socket) -> None:
+        assert _recv_exact(sock, 5) == MAGIC
+        sock.sendall(bytes([espota2.RESPONSE_OK, self.version]))
+        features = _recv_exact(sock, 1)[0]
+        noise_negotiated = bool(
+            features & espota2.CLIENT_FEATURE_SUPPORTS_NOISE
+            and features & espota2.CLIENT_FEATURE_SUPPORTS_EXTENDED_PROTOCOL
+        )
+        if self.require_noise and not noise_negotiated:
+            sock.sendall(bytes([espota2.RESPONSE_ERROR_ENCRYPTION_REQUIRED]))
+            return
+        server_flags = espota2.SERVER_FEATURE_SUPPORTS_NOISE if self.offer_noise else 0
+        sock.sendall(bytes([espota2.RESPONSE_FEATURE_FLAGS, server_flags]))
+        if not (self.offer_noise and noise_negotiated):
+            return  # the client fails closed; nothing further arrives
+
+        from cryptography.exceptions import InvalidTag
+        from noise.connection import NoiseConnection
+
+        prologue_features = (
+            features
+            if self.prologue_features_override is None
+            else self.prologue_features_override
+        )
+        prologue = (
+            espota2.NOISE_PROLOGUE_INIT
+            + MAGIC
+            + bytes([espota2.RESPONSE_OK, self.version, prologue_features])
+            + bytes([espota2.RESPONSE_FEATURE_FLAGS, server_flags])
+        )
+        proto = NoiseConnection.from_name(b"Noise_NNpsk0_25519_ChaChaPoly_SHA256")
+        proto.set_as_responder()
+        proto.set_psks(base64.b64decode(self.psk))
+        proto.set_prologue(prologue)
+        proto.start_handshake()
+
+        msg1 = _recv_frame(sock)
+        assert msg1[0] == 0x00
+        try:
+            proto.read_message(msg1[1:])
+        except InvalidTag:
+            _send_frame(sock, b"\x01" + espota2.NOISE_MAC_FAILURE_REASON.encode())
+            return
+        _send_frame(sock, b"\x00" + bytes(proto.write_message()))
+
+        def send_byte(byte: int) -> None:
+            _send_frame(sock, proto.encrypt(bytes([byte])))
+
+        def recv_unit(length: int) -> bytes:
+            plaintext = proto.decrypt(_recv_frame(sock))
+            assert len(plaintext) == length, "control units must be one per frame"
+            return plaintext
+
+        send_byte(espota2.RESPONSE_AUTH_OK)
+        recv_unit(1)  # ota type
+        size = int.from_bytes(recv_unit(4), "big")
+        send_byte(espota2.RESPONSE_UPDATE_PREPARE_OK)
+        md5_hex = recv_unit(32)
+        send_byte(espota2.RESPONSE_BIN_MD5_OK)
+
+        received = b""
+        acked = 0
+        while len(received) < size:
+            plaintext = proto.decrypt(_recv_frame(sock))
+            assert 0 < len(plaintext) <= espota2.NOISE_MAX_PLAINTEXT
+            received += plaintext
+            if self.version >= espota2.OTA_VERSION_2_0:
+                while acked + espota2.UPLOAD_BLOCK_SIZE <= len(received) or (
+                    len(received) == size and acked < size
+                ):
+                    send_byte(espota2.RESPONSE_CHUNK_OK)
+                    acked += espota2.UPLOAD_BLOCK_SIZE
+        assert hashlib.md5(received).hexdigest().encode() == md5_hex
+        send_byte(espota2.RESPONSE_RECEIVE_OK)
+        send_byte(espota2.RESPONSE_UPDATE_END_OK)
+        assert recv_unit(1) == bytes([espota2.RESPONSE_OK])
+        self.received = received
+
+
+def _upload(
+    device: FakeEncryptedDevice, firmware: bytes, noise_psk: str | None
+) -> None:
+    device.start()
+    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+    sock.settimeout(10)
+    sock.connect(("127.0.0.1", device.port))
+    try:
+        espota2.perform_ota(
+            sock, None, io.BytesIO(firmware), Path("firmware.bin"), noise_psk=noise_psk
+        )
+    finally:
+        sock.close()
+
+
+def test_encrypted_upload_success() -> None:
+    """A full encrypted v2 upload spanning several 8192-byte blocks."""
+    pytest.importorskip("aioesphomeapi.noise")
+    firmware = bytes(range(256)) * 80  # 20480 bytes, crosses chunk-ack boundaries
+    device = FakeEncryptedDevice()
+    with patch("time.sleep"):
+        _upload(device, firmware, PSK)
+    device.join_and_check()
+    assert device.received == firmware
+
+
+def test_encrypted_upload_version_1() -> None:
+    """Version 1 protocol (no chunk acks) works through the noise transport."""
+    pytest.importorskip("aioesphomeapi.noise")
+    firmware = b"v1 firmware image" * 100
+    device = FakeEncryptedDevice(version=1)
+    with patch("time.sleep"):
+        _upload(device, firmware, PSK)
+    device.join_and_check()
+    assert device.received == firmware
+
+
+def test_wrong_key_fails_with_clear_error() -> None:
+    """A key mismatch surfaces the device's handshake reject readably."""
+    pytest.importorskip("aioesphomeapi.noise")
+    device = FakeEncryptedDevice(psk=OTHER_PSK)
+    with pytest.raises(espota2.OTAError, match="encryption key correct"):
+        _upload(device, b"firmware", PSK)
+    device.join_and_check()
+
+
+def test_tampered_negotiation_breaks_handshake() -> None:
+    """A negotiation byte differing between the sides breaks the prologue MAC."""
+    pytest.importorskip("aioesphomeapi.noise")
+    device = FakeEncryptedDevice(
+        prologue_features_override=espota2.CLIENT_FEATURE_SUPPORTS_EXTENDED_PROTOCOL
+    )
+    with pytest.raises(espota2.OTAError, match="encryption key correct"):
+        _upload(device, b"firmware", PSK)
+    device.join_and_check()
+
+
+def test_client_fails_closed_when_device_lacks_encryption() -> None:
+    """With a key configured, a device not offering noise aborts the upload."""
+    device = FakeEncryptedDevice(offer_noise=False, require_noise=False)
+    with pytest.raises(espota2.OTAError, match="refusing to send the image"):
+        _upload(device, b"firmware", PSK)
+    device.join_and_check()
+
+
+def test_plaintext_client_gets_encryption_required_error() -> None:
+    """A client without a key gets the device's 0x94 error message."""
+    device = FakeEncryptedDevice()
+    with pytest.raises(espota2.OTAError, match="requires an encrypted OTA"):
+        _upload(device, b"firmware", None)
+    device.join_and_check()
+
+
+def test_missing_aioesphomeapi_noise_module_message() -> None:
+    """An aioesphomeapi without the noise module produces a clear error."""
+    with (
+        patch.dict(sys.modules, {"aioesphomeapi.noise": None}),
+        pytest.raises(espota2.OTAError, match="requires a newer aioesphomeapi"),
+    ):
+        espota2.NoiseSocketWrapper(Mock(), PSK, b"prologue")
+
+
+class ScriptedSocket:
+    """Serves scripted recv chunks; b"" means the peer closed."""
+
+    def __init__(self, *chunks: bytes | Exception) -> None:
+        self.chunks = list(chunks)
+        self.sent: list[bytes] = []
+
+    def sendall(self, data: bytes) -> None:
+        self.sent.append(data)
+
+    def settimeout(self, timeout: float) -> None:
+        pass
+
+    def recv(self, amount: int) -> bytes:
+        if not self.chunks:
+            return b""
+        chunk = self.chunks[0]
+        if isinstance(chunk, Exception):
+            self.chunks.pop(0)
+            raise chunk
+        take, rest = chunk[:amount], chunk[amount:]
+        if rest:
+            self.chunks[0] = rest
+        else:
+            self.chunks.pop(0)
+        return take
+
+
+def _wrapper(*chunks: bytes | Exception) -> espota2.NoiseSocketWrapper:
+    pytest.importorskip("aioesphomeapi.noise")
+    return espota2.NoiseSocketWrapper(ScriptedSocket(*chunks), PSK, b"prologue")
+
+
+def test_wrapper_rejects_malformed_psk() -> None:
+    pytest.importorskip("aioesphomeapi.noise")
+    with pytest.raises(espota2.OTAError, match="Invalid OTA encryption key"):
+        espota2.NoiseSocketWrapper(ScriptedSocket(), "not-base64!!!", b"prologue")
+
+
+def test_handshake_socket_error_is_network_error() -> None:
+    wrapper = _wrapper(OSError("boom"))
+    with pytest.raises(espota2.OTANetworkError, match="noise handshake"):
+        wrapper.do_handshake()
+
+
+def test_handshake_closed_at_frame_boundary() -> None:
+    wrapper = _wrapper()
+    with pytest.raises(espota2.OTANetworkError, match="closed connection during"):
+        wrapper.do_handshake()
+
+
+def test_handshake_reject_with_other_reason() -> None:
+    wrapper = _wrapper(_frame(b"\x01Handshake error"))
+    with pytest.raises(
+        espota2.OTAError, match="rejected the noise handshake: Handshake error"
+    ):
+        wrapper.do_handshake()
+
+
+def test_handshake_garbage_second_message() -> None:
+    """A valid-looking point with a garbage MAC fails cleanly."""
+    wrapper = _wrapper(_frame(b"\x00" + bytes(range(48))))
+    with pytest.raises(
+        espota2.OTAError, match="handshake failed; is the OTA encryption key"
+    ):
+        wrapper.do_handshake()
+
+
+def test_handshake_invalid_curve_point() -> None:
+    """An all-zero x25519 point is rejected as a clean error, not a crash."""
+    wrapper = _wrapper(_frame(b"\x00" + bytes(48)))
+    with pytest.raises(
+        espota2.OTAError, match="handshake failed; is the OTA encryption key"
+    ):
+        wrapper.do_handshake()
+
+
+def test_recv_closed_at_frame_boundary_returns_empty() -> None:
+    wrapper = _wrapper()
+    assert wrapper.recv(1) == b""
+
+
+def test_recv_corrupt_frame_is_retryable_network_error() -> None:
+    from cryptography.exceptions import InvalidTag
+
+    wrapper = _wrapper(_frame(b"ciphertext"))
+    wrapper._decrypt = Mock(decrypt=Mock(side_effect=InvalidTag()))
+    with pytest.raises(espota2.OTANetworkError, match="decryption failed"):
+        wrapper.recv(1)
+
+
+def test_wrapper_blocks_unencrypted_socket_methods() -> None:
+    """Byte-moving socket methods must not bypass the encrypted transport."""
+    wrapper = _wrapper()
+    # The harmless socket controls pass through to the wrapped socket
+    wrapper._sock = Mock()
+    wrapper.settimeout(1)
+    wrapper._sock.settimeout.assert_called_once_with(1)
+    wrapper.setsockopt(6, 1, 1)
+    wrapper._sock.setsockopt.assert_called_once_with(6, 1, 1)
+    wrapper.close()
+    wrapper._sock.close.assert_called_once_with()
+    with pytest.raises(AttributeError):
+        _ = wrapper.send
+    with pytest.raises(AttributeError):
+        _ = wrapper.recv_into
+
+
+def test_recv_empty_plaintext_frame_is_protocol_error() -> None:
+    """A MAC-only frame decrypts to nothing; b'' from recv must mean close."""
+    wrapper = _wrapper(_frame(bytes(16)))
+    wrapper._decrypt = Mock(decrypt=Mock(return_value=b""))
+    with pytest.raises(espota2.OTANetworkError, match="empty noise frame"):
+        wrapper.recv(1)
+
+
+def test_recv_frame_bad_indicator_is_retryable() -> None:
+    wrapper = _wrapper(b"\x02\x00\x01x")
+    with pytest.raises(espota2.OTANetworkError, match="Bad noise frame indicator"):
+        wrapper._recv_frame()
+
+
+def test_recv_frame_zero_length_is_retryable() -> None:
+    wrapper = _wrapper(bytes([espota2.NOISE_FRAME_INDICATOR, 0, 0]))
+    with pytest.raises(espota2.OTANetworkError, match="empty noise frame"):
+        wrapper._recv_frame()
+
+
+def test_perform_ota_blank_key_refuses_plaintext() -> None:
+    with pytest.raises(espota2.OTAError, match="empty OTA encryption key"):
+        espota2.perform_ota(
+            ScriptedSocket(), None, io.BytesIO(b"x"), Path("f.bin"), noise_psk=""
+        )
+
+
+def test_recv_exact_closed_mid_frame() -> None:
+    wrapper = _wrapper(_frame(b"partial")[:5])
+    with pytest.raises(OSError, match="closed inside a noise frame"):
+        wrapper._recv_frame()
+
+
+def test_recv_serves_buffered_plaintext_without_new_frame() -> None:
+    """A second recv drains the decrypted buffer without reading another frame."""
+    wrapper = _wrapper(_frame(b"ciphertext"))
+    wrapper._decrypt = Mock(decrypt=Mock(return_value=b"AB"))
+    assert wrapper.recv(1) == b"A"  # reads and decrypts one frame
+    assert wrapper.recv(1) == b"B"  # served from the buffer, no new frame
+    wrapper._decrypt.decrypt.assert_called_once()

tests/unit_tests/test_framework_helpers.py

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

tests/unit_tests/test_git.py

@@ -714,6 +714,25 @@ def test_run_git_command_without_git_dir_raises_error(
         git.run_git_command(["git", "clone", "https://invalid.url/repo.git"])
 
 
+def test_has_complete_clone(tmp_path: Path) -> None:
+    """The lock-free probe tracks the completion marker, subpath included."""
+    CORE.config_path = tmp_path / "test.yaml"
+
+    url = "https://github.com/test/repo"
+    subpath = Path("lib")
+    assert not git.has_complete_clone(url, "v1", "test_domain", subpath)
+
+    repo_dir = _compute_repo_dir(url, "v1", "test_domain") / subpath
+    (repo_dir / ".git").mkdir(parents=True)
+    # A directory without the marker is an incomplete clone
+    assert not git.has_complete_clone(url, "v1", "test_domain", subpath)
+
+    _mark_clone_complete(repo_dir)
+    assert git.has_complete_clone(url, "v1", "test_domain", subpath)
+    # The ref is part of the cache key
+    assert not git.has_complete_clone(url, "v2", "test_domain", subpath)
+
+
 def test_clone_or_update_with_never_refresh(
     tmp_path: Path, mock_run_git_command: Mock
 ) -> None:

tests/unit_tests/test_helpers.py

@@ -1,10 +1,12 @@
+import errno
 import io
 import logging
 import os
 from pathlib import Path
 import socket
 import stat
-from unittest.mock import MagicMock, patch
+import types
+from unittest.mock import MagicMock, call, patch
 
 from aioesphomeapi.host_resolver import AddrInfo, IPv4Sockaddr, IPv6Sockaddr
 from hypothesis import given, settings
@@ -965,6 +967,77 @@ def test_copy_file_if_changed_nonexistent_source(tmp_path: Path) -> None:
         helpers.copy_file_if_changed(src, dst)
 
 
+def test_rmtree_removes_tree(tmp_path: Path) -> None:
+    """Test rmtree removes a populated directory tree."""
+    target = tmp_path / "target"
+    (target / "sub").mkdir(parents=True)
+    (target / "sub" / "file.txt").write_text("content")
+
+    helpers.rmtree(target)
+    assert not target.exists()
+
+
+def test_rmtree_nonexistent_path(tmp_path: Path) -> None:
+    """Test rmtree on an already-removed path is a no-op."""
+    helpers.rmtree(tmp_path / "gone")
+
+
+def test_rmtree_retries_when_directory_repopulated(tmp_path: Path) -> None:
+    """Test rmtree retries when a file appears mid-delete (Finder .DS_Store race)."""
+    target = tmp_path / "target"
+    (target / "sub").mkdir(parents=True)
+    real_rmdir = os.rmdir
+    repopulated = False
+
+    def racy_rmdir(path, **kwargs):
+        nonlocal repopulated
+        if not repopulated and Path(path).name == "target":
+            repopulated = True
+            (target / ".DS_Store").write_text("x")  # Finder wins the race
+        real_rmdir(path, **kwargs)
+
+    with patch("os.rmdir", side_effect=racy_rmdir), patch("time.sleep"):
+        helpers.rmtree(target)
+    assert repopulated
+    assert not target.exists()
+
+
+def test_rmtree_raises_after_retries_exhausted(tmp_path: Path) -> None:
+    """Test rmtree gives up on a persistent ENOTEMPTY once attempts run out."""
+    target = tmp_path / "target"
+    target.mkdir()
+    errs = [
+        OSError(errno.ENOTEMPTY, "Directory not empty", str(target))
+        for _ in range(helpers.RMTREE_MAX_ATTEMPTS)
+    ]
+
+    with (
+        patch("shutil.rmtree", side_effect=errs) as mock_rmtree,
+        patch("time.sleep") as mock_sleep,
+        pytest.raises(OSError, match="Directory not empty") as excinfo,
+    ):
+        helpers.rmtree(target)
+    assert mock_rmtree.call_count == helpers.RMTREE_MAX_ATTEMPTS
+    assert mock_sleep.call_args_list == [call(0.05), call(0.1)]
+    # Final failure chains to the last retried race
+    assert excinfo.value is errs[-1]
+    assert excinfo.value.__cause__ is errs[-2]
+
+
+def test_rmtree_does_not_retry_other_oserror(tmp_path: Path) -> None:
+    """Test rmtree raises non-ENOTEMPTY errors immediately."""
+    target = tmp_path / "target"
+    target.mkdir()
+    err = OSError(errno.EACCES, "Permission denied", str(target))
+
+    with (
+        patch("shutil.rmtree", side_effect=err) as mock_rmtree,
+        pytest.raises(OSError, match="Permission denied"),
+    ):
+        helpers.rmtree(target)
+    assert mock_rmtree.call_count == 1
+
+
 def test_resolve_ip_address_sorting() -> None:
     """Test that results are sorted by preference."""
     # Create multiple address infos with different preferences
@@ -1154,3 +1227,26 @@ def test_progressbar_interrupt_keeps_finished_bar_done(monkeypatch) -> None:
 def test_format_duration(seconds: float, expected: str) -> None:
     """Test that durations are rendered as short human-readable strings."""
     assert helpers.format_duration(seconds) == expected
+
+
+def test_get_usable_cpu_count() -> None:
+    """Returns a positive int on the real host."""
+    count = helpers.get_usable_cpu_count()
+    assert isinstance(count, int)
+    assert count > 0
+
+
+def test_get_usable_cpu_count_sources() -> None:
+    """Prefers process_cpu_count, falls back to cpu_count, degrades to 1."""
+    mock_os = types.SimpleNamespace(process_cpu_count=lambda: 8, cpu_count=lambda: 4)
+    with patch("esphome.helpers.os", mock_os):
+        assert helpers.get_usable_cpu_count() == 8
+
+    mock_os_no_process = types.SimpleNamespace(cpu_count=lambda: 4)
+    with patch("esphome.helpers.os", mock_os_no_process):
+        assert helpers.get_usable_cpu_count() == 4
+
+    # An undeterminable count degrades to one worker, never zero
+    mock_os_unknown = types.SimpleNamespace(cpu_count=lambda: None)
+    with patch("esphome.helpers.os", mock_os_unknown):
+        assert helpers.get_usable_cpu_count() == 1

tests/unit_tests/test_main.py

@@ -86,7 +86,9 @@ from esphome.const import (
     CONF_BROKER,
     CONF_DISABLED,
     CONF_DISCOVER_IP,
+    CONF_ENCRYPTION,
     CONF_ESPHOME,
+    CONF_KEY,
     CONF_LEVEL,
     CONF_LOG,
     CONF_LOG_TOPIC,
@@ -112,6 +114,7 @@ from esphome.const import (
     PLATFORM_BK72XX,
     PLATFORM_ESP32,
     PLATFORM_ESP8266,
+    PLATFORM_HOST,
     PLATFORM_NRF52,
     PLATFORM_RP2,
     Toolchain,
@@ -2105,10 +2108,65 @@ def test_upload_program_ota_success(
         tmp_path / ".esphome" / "build" / "test" / ".pioenvs" / "test" / "firmware.bin"
     )
     mock_run_ota.assert_called_once_with(
-        ["192.168.1.100"], 3232, "secret", expected_firmware, OTA_TYPE_UPDATE_APP
+        ["192.168.1.100"], 3232, "secret", expected_firmware, OTA_TYPE_UPDATE_APP, None
     )
 
 
+def test_upload_program_ota_encryption_key(
+    mock_run_ota: Mock,
+    mock_get_port_type: Mock,
+    tmp_path: Path,
+) -> None:
+    """The resolved encryption key is passed through to run_ota."""
+    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")
+
+    key = "AAECAwQFBgcICQoLDA0ODxAREhMUFRYXGBkaGxwdHh8="
+    config = {
+        CONF_OTA: [
+            {
+                CONF_PLATFORM: CONF_ESPHOME,
+                CONF_PORT: 3232,
+                CONF_ENCRYPTION: {CONF_KEY: key},
+            }
+        ]
+    }
+    exit_code, host = upload_program(config, MockArgs(), ["192.168.1.100"])
+
+    assert exit_code == 0
+    assert host == "192.168.1.100"
+    expected_firmware = (
+        tmp_path / ".esphome" / "build" / "test" / ".pioenvs" / "test" / "firmware.bin"
+    )
+    mock_run_ota.assert_called_once_with(
+        ["192.168.1.100"], 3232, None, expected_firmware, OTA_TYPE_UPDATE_APP, key
+    )
+
+
+def test_upload_program_ota_encryption_without_key_fails_closed(
+    mock_run_ota: Mock,
+    mock_get_port_type: Mock,
+    tmp_path: Path,
+) -> None:
+    """An encryption block with no resolved key must never upload plaintext."""
+    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_ENCRYPTION: {},
+            }
+        ]
+    }
+    with pytest.raises(EsphomeError, match="no key was resolved"):
+        upload_program(config, MockArgs(), ["192.168.1.100"])
+    mock_run_ota.assert_not_called()
+
+
 def test_upload_program_ota_with_file_arg(
     mock_run_ota: Mock,
     mock_get_port_type: Mock,
@@ -2136,7 +2194,7 @@ def test_upload_program_ota_with_file_arg(
     assert exit_code == 0
     assert host == "192.168.1.100"
     mock_run_ota.assert_called_once_with(
-        ["192.168.1.100"], 3232, None, Path("custom.bin"), OTA_TYPE_UPDATE_APP
+        ["192.168.1.100"], 3232, None, Path("custom.bin"), OTA_TYPE_UPDATE_APP, None
     )
 
 
@@ -2191,6 +2249,7 @@ def test_upload_program_ota_partition_table_with_file_arg(
         None,
         partition_file,
         OTA_TYPE_UPDATE_PARTITION_TABLE,
+        None,
     )
 
 
@@ -2252,6 +2311,7 @@ def test_upload_program_ota_partition_table_mqttip(
         None,
         partition_file,
         OTA_TYPE_UPDATE_PARTITION_TABLE,
+        None,
     )
 
 
@@ -2439,6 +2499,7 @@ def test_upload_program_ota_bootloader_with_file_arg(
         None,
         bootloader_file,
         OTA_TYPE_UPDATE_BOOTLOADER,
+        None,
     )
 
 
@@ -2601,6 +2662,42 @@ def test_has_web_server_logging_respects_log_disabled() -> None:
     assert has_web_server_logging() is False
 
 
+def test_upload_program_web_server_warns_when_encryption_configured(
+    mock_run_web_server_ota: Mock,
+    mock_run_ota: Mock,
+    mock_get_port_type: Mock,
+    tmp_path: Path,
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """Explicitly picking web_server OTA on an encrypted config warns about
+    the plaintext upload path."""
+    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_ENCRYPTION: {CONF_KEY: "test_key"},
+            },
+            {CONF_PLATFORM: CONF_WEB_SERVER},
+        ],
+        CONF_WEB_SERVER: {
+            CONF_PORT: 80,
+            CONF_AUTH: {CONF_USERNAME: "admin", CONF_PASSWORD: "pw"},
+        },
+    }
+    args = MockArgs(ota_platform=CONF_WEB_SERVER)
+    with caplog.at_level(logging.WARNING):
+        exit_code, _ = upload_program(config, args, ["192.168.1.100"])
+
+    assert exit_code == 0
+    assert any("plaintext HTTP" in record.message for record in caplog.records)
+    mock_run_ota.assert_not_called()
+
+
 def test_upload_program_web_server_only_auto_dispatches(
     mock_run_web_server_ota: Mock,
     mock_run_ota: Mock,
@@ -2891,7 +2988,7 @@ def test_upload_program_ota_with_mqtt_resolution(
         tmp_path / ".esphome" / "build" / "test" / ".pioenvs" / "test" / "firmware.bin"
     )
     mock_run_ota.assert_called_once_with(
-        ["192.168.1.100"], 3232, None, expected_firmware, OTA_TYPE_UPDATE_APP
+        ["192.168.1.100"], 3232, None, expected_firmware, OTA_TYPE_UPDATE_APP, None
     )
 
 
@@ -2941,7 +3038,7 @@ def test_upload_program_ota_with_mqtt_empty_broker(
         tmp_path / ".esphome" / "build" / "test" / ".pioenvs" / "test" / "firmware.bin"
     )
     mock_run_ota.assert_called_once_with(
-        ["192.168.1.50"], 3232, None, expected_firmware, OTA_TYPE_UPDATE_APP
+        ["192.168.1.50"], 3232, None, expected_firmware, OTA_TYPE_UPDATE_APP, None
     )
     # Verify warning was logged
     assert "MQTT IP discovery failed" in caplog.text
@@ -5113,6 +5210,7 @@ def test_upload_program_ota_static_ip_with_mqttip(
         None,
         expected_firmware,
         OTA_TYPE_UPDATE_APP,
+        None,
     )
 
 
@@ -5162,6 +5260,7 @@ def test_upload_program_ota_multiple_mqttip_resolves_once(
         None,
         expected_firmware,
         OTA_TYPE_UPDATE_APP,
+        None,
     )
 
 
@@ -5339,7 +5438,7 @@ def test_upload_program_ota_mqtt_timeout_fallback(
         tmp_path / ".esphome" / "build" / "test" / ".pioenvs" / "test" / "firmware.bin"
     )
     mock_run_ota.assert_called_once_with(
-        ["192.168.1.100"], 3232, None, expected_firmware, OTA_TYPE_UPDATE_APP
+        ["192.168.1.100"], 3232, None, expected_firmware, OTA_TYPE_UPDATE_APP, None
     )
 
 
@@ -7254,3 +7353,54 @@ async def test_wrap_to_code_comment_is_insertion_order_independent() -> None:
     assert first == second
     assert second.index("alpha") < second.index("beta")
     assert second.index("a: 2") < second.index("z: 1")
+
+
+def test_host_program_path_platformio_toolchain() -> None:
+    """Host + PlatformIO toolchain reads the memoized idedata path."""
+    setup_core(platform=PLATFORM_HOST)
+    idedata = SimpleNamespace(firmware_elf_path="/build/x/.pioenvs/x/program")
+    with patch(
+        "esphome.platformio.toolchain.get_idedata", return_value=idedata
+    ) as mock_get:
+        assert main._host_program_path({}) == "/build/x/.pioenvs/x/program"
+    mock_get.assert_called_once_with({})
+
+
+def test_host_program_path_esp_idf_toolchain() -> None:
+    """Host + native ESP-IDF toolchain asks the espidf toolchain for the ELF."""
+    setup_core(platform=PLATFORM_HOST)
+    CORE.toolchain = Toolchain.ESP_IDF
+    with patch(
+        "esphome.espidf.toolchain.get_elf_path", return_value=Path("/b/app.elf")
+    ):
+        assert main._host_program_path({}) == str(Path("/b/app.elf"))
+
+
+def test_command_compile_host_logs_program_path(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """command_compile on host logs the compiled program path."""
+    setup_core(platform=PLATFORM_HOST)
+    with (
+        patch.object(main, "write_cpp", return_value=0),
+        patch.object(main, "compile_program", return_value=0),
+        patch.object(main, "_host_program_path", return_value="/b/program"),
+        caplog.at_level(logging.INFO),
+    ):
+        assert main.command_compile(SimpleNamespace(only_generate=False), {}) == 0
+    assert "Successfully compiled program to path '/b/program'" in caplog.text
+
+
+def test_command_run_host_executes_program(caplog: pytest.LogCaptureFixture) -> None:
+    """command_run on host logs and executes the compiled program directly."""
+    setup_core(platform=PLATFORM_HOST)
+    with (
+        patch.object(main, "write_cpp", return_value=0),
+        patch.object(main, "compile_program", return_value=0),
+        patch.object(main, "_host_program_path", return_value="/b/program"),
+        patch.object(main, "run_external_process", return_value=0) as mock_run,
+        caplog.at_level(logging.INFO),
+    ):
+        assert main.command_run(SimpleNamespace(), {}) == 0
+    mock_run.assert_called_with("/b/program")
+    assert "Running program from path '/b/program'" in caplog.text

tests/unit_tests/test_platformio_library.py

@@ -10,7 +10,7 @@ from pathlib import Path
 
 import pytest
 
-from esphome.core import EsphomeError, Library
+from esphome.core import CORE, EsphomeError, Library
 import esphome.platformio.library as lib
 from esphome.platformio.library import (
     SOURCE_KIND_FOR_SUFFIX,
@@ -29,9 +29,13 @@ from esphome.platformio.library import (
 )
 
 
-def _backend(emit=lambda component: None) -> LibraryBackend:
+def _backend(emit=lambda component: None, provides=None) -> LibraryBackend:
     return LibraryBackend(
-        platform="espressif32", framework="espidf", emit=emit, cache_key="idf"
+        platform="espressif32",
+        framework="espidf",
+        emit=emit,
+        cache_key="idf",
+        provides=provides,
     )
 
 
@@ -634,7 +638,7 @@ def test_prefetch_wave_downloads_registry_archives_in_parallel(
     setup_core, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
 ) -> None:
     """Registry archives in one wave download concurrently, deduped by URL;
-    git/local sources and failures are left to the sequential call."""
+    local sources and failures are left to the sequential call."""
     calls: list[str] = []
 
     def fake_download(
@@ -654,7 +658,7 @@ def test_prefetch_wave_downloads_registry_archives_in_parallel(
         # into the same cache directory)
         ("b2", ConvertedLibrary("b2", "1.0", URLSource("https://x/b.tar.gz", 1))),
         ("c", ConvertedLibrary("c", "1.0", URLSource("https://x/boom.tar.gz", 1))),
-        ("g", ConvertedLibrary("g", "*", lib.GitSource("https://x/g.git", None))),
+        ("l", ConvertedLibrary("l", "*", LocalSource("/some/lib"))),
     ]
     lib._prefetch_wave(wave, "", "idf")
     assert sorted(calls) == [
@@ -666,6 +670,83 @@ def test_prefetch_wave_downloads_registry_archives_in_parallel(
     assert "Prefetch of c failed (retrying sequentially)" in caplog.text
 
 
+def test_prefetch_wave_clones_git_sources_in_parallel(
+    setup_core, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
+) -> None:
+    """Git sources join the same prefetch batch as the archives, deduped by
+    clone target; a clone failure warns and is left to the sequential call."""
+    caplog.set_level("INFO")
+    calls: list[str] = []
+
+    def fake_clone(self, dir_suffix, force=False, salt="", namespace=""):
+        calls.append(f"{self}/{dir_suffix}")
+        if "boom" in self.url:
+            raise RuntimeError("boom")
+
+    monkeypatch.setattr(GitSource, "download", fake_clone)
+    wave = [
+        ("a", ConvertedLibrary("a", "1.0", URLSource("https://x/a.tar.gz", 1))),
+        ("g", ConvertedLibrary("g", "*", GitSource("https://x/g.git", "v1"))),
+        # Same url@ref and target dir must clone once
+        ("g2", ConvertedLibrary("g", "*", GitSource("https://x/g.git", "v1"))),
+        ("h", ConvertedLibrary("h", "*", GitSource("https://x/boom.git", None))),
+    ]
+    monkeypatch.setattr(
+        URLSource, "download", lambda self, dir_suffix, progress=None, **kw: None
+    )
+    lib._prefetch_wave(wave, "", "idf")
+    assert sorted(calls) == ["https://x/boom.git/h", "https://x/g.git#v1/g"]
+    assert "Cloning 2 library repo(s): g, h" in caplog.text
+    assert "Prefetch of h failed (retrying sequentially)" in caplog.text
+
+
+def test_source_base_prefetch_defaults() -> None:
+    """The base Source is not prefetchable and reports cached (nothing to do)."""
+    source = Source()
+    assert source.prefetch_key("x") is None
+    assert source.is_cached("x") is True
+
+
+def test_prefetch_wave_single_clone_uses_the_batch(
+    monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
+) -> None:
+    """A wave with only git sources still clones through the batch runner."""
+    caplog.set_level("INFO")
+    calls: list[str] = []
+    monkeypatch.setattr(GitSource, "is_cached", lambda self, *a, **kw: False)
+    monkeypatch.setattr(
+        GitSource,
+        "download",
+        lambda self, dir_suffix, force=False, salt="", namespace="": calls.append(
+            self.url
+        ),
+    )
+    lib._prefetch_wave(
+        [("g", ConvertedLibrary("g", "*", GitSource("https://x/g.git", None)))],
+        "",
+        "idf",
+    )
+    assert calls == ["https://x/g.git"]
+    assert "Cloning 1 library repo(s): g" in caplog.text
+    assert "Downloading" not in caplog.text
+
+
+def test_prefetch_wave_warm_git_cache_is_silent(
+    setup_core, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
+) -> None:
+    """An already-complete clone is neither re-fetched nor announced."""
+    caplog.set_level("INFO")
+    monkeypatch.setattr(
+        GitSource,
+        "download",
+        lambda self, dir_suffix, **kw: (_ for _ in ()).throw(AssertionError("cloned")),
+    )
+    monkeypatch.setattr(GitSource, "is_cached", lambda self, *a, **kw: True)
+    wave = [("g", ConvertedLibrary("g", "*", GitSource("https://x/g.git", None)))]
+    lib._prefetch_wave(wave, "", "idf")
+    assert "Cloning" not in caplog.text
+
+
 def test_prefetch_wave_unknown_size_left_to_sequential(
     setup_core, monkeypatch: pytest.MonkeyPatch
 ) -> None:
@@ -952,3 +1033,202 @@ def test_source_kind_map_shape() -> None:
     assert SOURCE_KIND_FOR_SUFFIX[".S"] == "aspp"
     assert SOURCE_KIND_FOR_SUFFIX[".c"] == "c"
     assert SOURCE_KIND_FOR_SUFFIX[".cpp"] == "cxx"
+    # SCons's case-sensitive C++ suffixes: PIO compiles .C as C++
+    assert SOURCE_KIND_FOR_SUFFIX[".C"] == "cxx"
+    assert SOURCE_KIND_FOR_SUFFIX[".C++"] == "cxx"
+
+
+def test_versionless_platform_filtered_dependency_stays_quiet(
+    tmp_path, monkeypatch, caplog: pytest.LogCaptureFixture
+) -> None:
+    """A version-less dependency the platform filter excludes is
+    deliberately absent, not a drop to warn about."""
+    _patch_download_with_manifests(
+        monkeypatch,
+        tmp_path,
+        {
+            "esphome/A": {
+                "name": "A",
+                "dependencies": [{"name": "Hash", "platforms": "espressif8266"}],
+            }
+        },
+    )
+    convert_libraries([Library("esphome/A", None, None)], _backend())
+    assert "has no version to resolve" not in caplog.text
+
+
+def test_versionless_ignored_dependency_stays_quiet(
+    tmp_path, monkeypatch, caplog: pytest.LogCaptureFixture
+) -> None:
+    """A lib_ignore'd version-less dependency is deliberately excluded, not
+    a drop; no reconciliation warning."""
+    _patch_download_with_manifests(
+        monkeypatch,
+        tmp_path,
+        {"esphome/A": {"name": "A", "dependencies": [{"name": "Hash"}]}},
+    )
+    CORE.platformio_options = {"lib_ignore": ["Hash"]}
+    convert_libraries([Library("esphome/A", None, None)], _backend())
+    assert "has no version to resolve" not in caplog.text
+
+
+def test_versionless_dependency_without_provider_warns(
+    tmp_path, monkeypatch, caplog: pytest.LogCaptureFixture
+) -> None:
+    """A backend whose tree could supply the name warns on the drop; one
+    without provides() can never act on it, so it stays at debug."""
+    _patch_download_with_manifests(
+        monkeypatch,
+        tmp_path,
+        {
+            "esphome/A": {
+                "name": "A",
+                # The duplicate entry warns once (reconciliation dedup)
+                "dependencies": [{"name": "Hash"}, {"name": "Hash"}],
+            }
+        },
+    )
+    convert_libraries(
+        [Library("esphome/A", None, None)], _backend(provides=lambda name: False)
+    )
+    assert (
+        caplog.text.count(
+            "Hash of esphome/A has no version to resolve and nothing provides it"
+        )
+        == 1
+    )
+    caplog.clear()
+    with caplog.at_level(logging.DEBUG):
+        convert_libraries([Library("esphome/A", None, None)], _backend())
+    records = [
+        r
+        for r in caplog.records
+        if "has no version to resolve and nothing provides it" in r.message
+    ]
+    assert records and all(r.levelno == logging.DEBUG for r in records)
+
+
+def test_url_version_dependency_is_not_substituted_by_provides(
+    tmp_path, monkeypatch, caplog: pytest.LogCaptureFixture
+) -> None:
+    """A URL-valued version names one specific source; the backend-provided
+    skip must not replace it with the bundled copy."""
+    _patch_download_with_manifests(
+        monkeypatch,
+        tmp_path,
+        {
+            "esphome/A": {
+                "name": "A",
+                "dependencies": [
+                    {"name": "Hash", "version": "https://github.com/o/Hash.git"}
+                ],
+            },
+            "o/Hash": {"name": "Hash"},
+        },
+    )
+    emitted: list[str] = []
+    convert_libraries(
+        [Library("esphome/A", "1.0.0", None)],
+        _backend(emit=lambda c: emitted.append(c.name), provides=lambda name: True),
+    )
+    assert "Skip backend-provided" not in caplog.text
+    assert "using the library bundled" not in caplog.text
+    assert any("o/hash" in n.lower() for n in emitted)
+
+
+def test_versionless_owner_qualified_dependency_warns_despite_provides(
+    tmp_path, monkeypatch, caplog: pytest.LogCaptureFixture
+) -> None:
+    """An owner-qualified version-less dependency is not satisfied by
+    provides(); it must still warn."""
+    _patch_download_with_manifests(
+        monkeypatch,
+        tmp_path,
+        {
+            "esphome/A": {
+                "name": "A",
+                "dependencies": [{"name": "Wire", "owner": "Foo"}],
+            }
+        },
+    )
+    convert_libraries(
+        [Library("esphome/A", None, None)],
+        _backend(provides=lambda name: name == "Wire"),
+    )
+    assert "Wire of esphome/A has no version to resolve" in caplog.text
+
+
+def test_versionless_provided_dependency_stays_quiet(
+    tmp_path, monkeypatch, caplog: pytest.LogCaptureFixture
+) -> None:
+    """An owner-less version-less dependency the backend provides is added
+    by the backend after emit; no reconciliation warning."""
+    _patch_download_with_manifests(
+        monkeypatch,
+        tmp_path,
+        {"esphome/A": {"name": "A", "dependencies": [{"name": "Wire"}]}},
+    )
+    convert_libraries(
+        [Library("esphome/A", None, None)],
+        _backend(provides=lambda name: name == "Wire"),
+    )
+    assert "has no version to resolve" not in caplog.text
+
+
+def test_versionless_dependency_requested_top_level_stays_quiet(
+    tmp_path, monkeypatch, caplog: pytest.LogCaptureFixture
+) -> None:
+    """A version-less dependency the config also requests top-level is in
+    the build; no drop warning even without a provides backend."""
+    _patch_download_with_manifests(
+        monkeypatch,
+        tmp_path,
+        {
+            "esphome/A": {"name": "A", "dependencies": [{"name": "Hash"}]},
+            "Hash": {"name": "Hash"},
+        },
+    )
+    convert_libraries(
+        [Library("esphome/A", None, None), Library("Hash", None, None)],
+        _backend(),
+    )
+    assert "has no version to resolve" not in caplog.text
+
+
+def test_versionless_url_ish_dependency_name_warns_cleanly(
+    tmp_path, monkeypatch, caplog: pytest.LogCaptureFixture
+) -> None:
+    """A malformed URL-ish dependency name falls to the drop warning, never
+    a RuntimeError out of the key parser."""
+    _patch_download_with_manifests(
+        monkeypatch,
+        tmp_path,
+        {"esphome/A": {"name": "A", "dependencies": [{"name": "file://"}]}},
+    )
+    convert_libraries(
+        [Library("esphome/A", None, None)], _backend(provides=lambda name: False)
+    )
+    assert (
+        "file:// of esphome/A has no version to resolve and nothing provides it"
+        in caplog.text
+    )
+
+
+def test_versionless_dependency_matching_resolved_manifest_name_stays_quiet(
+    tmp_path, monkeypatch, caplog: pytest.LogCaptureFixture
+) -> None:
+    """A bare name satisfied by an owner-qualified component's manifest
+    name is not a drop."""
+    _patch_download_with_manifests(
+        monkeypatch,
+        tmp_path,
+        {
+            "esphome/A": {"name": "A", "dependencies": [{"name": "B"}]},
+            "esphome/B": {"name": "B"},
+        },
+    )
+    convert_libraries(
+        [Library("esphome/A", None, None), Library("esphome/B", None, None)],
+        _backend(),
+    )
+    assert "has no version to resolve" not in caplog.text

tests/unit_tests/test_platformio_toolchain.py

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

tests/unit_tests/test_yaml_util.py

@@ -1706,6 +1706,53 @@ def test_dump_path_dotdot_reference_outside_anchor() -> None:
     assert output.strip() == "file: ../shared/font.ttf"
 
 
+@pytest.mark.parametrize(
+    "data_dir",
+    [
+        pytest.param(Path("/config/.esphome"), id="cli"),
+        pytest.param(Path("/data"), id="addon"),
+    ],
+)
+def test_dump_path_under_data_dir_uses_default_location(data_dir: Path) -> None:
+    """Test that Path values under data_dir dump as .esphome/<rest> for any layout."""
+    anchor = Path("/config").absolute()
+    path = data_dir.absolute() / "image" / "c44630d6"
+    output = yaml_util.dump(
+        {"file": path}, relative_to=anchor, data_dir=data_dir.absolute()
+    )
+    assert output.strip() == "file: .esphome/image/c44630d6"
+
+
+def test_dump_path_equal_to_data_dir() -> None:
+    """Test that the data dir itself dumps as .esphome, matching the default layout."""
+    anchor = Path("/config").absolute()
+    data_dir = Path("/data").absolute()
+    output = yaml_util.dump({"dir": data_dir}, relative_to=anchor, data_dir=data_dir)
+    assert output.strip() == "dir: .esphome"
+    default = yaml_util.dump(
+        {"dir": anchor / ".esphome"}, relative_to=anchor, data_dir=anchor / ".esphome"
+    )
+    assert default == output
+
+
+def test_dump_path_outside_data_dir_still_relative_to_anchor() -> None:
+    """Test that data_dir does not affect paths that are not under it."""
+    anchor = Path("/config").absolute()
+    path = anchor / "fonts" / "arial.ttf"
+    output = yaml_util.dump(
+        {"file": path}, relative_to=anchor, data_dir=Path("/data").absolute()
+    )
+    assert output.strip() == "file: fonts/arial.ttf"
+
+
+def test_dump_path_data_dir_without_relative_to_is_unchanged() -> None:
+    """Test that data_dir alone does not change the output."""
+    data_dir = Path("/data").absolute()
+    path = data_dir / "image" / "c44630d6"
+    output = yaml_util.dump({"file": path}, data_dir=data_dir)
+    assert output.strip() == f"file: {path}"
+
+
 def test_dump_relative_to_does_not_leak_between_calls() -> None:
     """Test that the relative_to flag is scoped to a single dump call."""
     anchor = Path("/config/esphome").absolute()