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

tests/unit_tests/build_gen/test_espidf.py

@@ -11,10 +11,12 @@ import pytest
 from esphome.components.esp32 import (
     KEY_COMPONENTS,
     KEY_ESP32,
+    KEY_EXCLUDE_COMPONENTS,
     KEY_IDF_VERSION,
     KEY_PATH,
     KEY_REF,
     KEY_REPO,
+    register_exclude_components_cmake_arg,
 )
 import esphome.config_validation as cv
 from esphome.const import KEY_CORE
@@ -28,6 +30,7 @@ def _reset_core(tmp_path: Path) -> None:
     CORE.data.setdefault(KEY_CORE, {})
     CORE.data[KEY_ESP32] = {
         KEY_COMPONENTS: {},
+        KEY_EXCLUDE_COMPONENTS: set(),
         KEY_IDF_VERSION: cv.Version(5, 5, 4),
     }
 
@@ -47,6 +50,17 @@ def _write_project_description(tmp_path: Path, components: dict[str, str]) -> No
     )
 
 
+def _render(minimal: bool = False) -> str:
+    """Render the top-level CMakeLists with the standard variant/name patches."""
+    with (
+        patch("esphome.build_gen.espidf.get_esp32_variant", return_value="ESP32"),
+        patch.object(CORE, "name", "test"),
+    ):
+        from esphome.build_gen.espidf import get_project_cmakelists
+
+        return get_project_cmakelists(minimal=minimal)
+
+
 def test_get_available_components_returns_none_without_build_path() -> None:
     """No build_path set yet: must not raise on Path(None)."""
     CORE.build_path = None
@@ -88,13 +102,7 @@ def test_get_project_cmakelists_minimal_omits_builtin_components_property(
     first write before the discovery pass refreshes it)."""
     _write_project_description(tmp_path, {"esp_lcd": "/idf/components/esp_lcd"})
 
-    with (
-        patch("esphome.build_gen.espidf.get_esp32_variant", return_value="ESP32"),
-        patch.object(CORE, "name", "test"),
-    ):
-        from esphome.build_gen.espidf import get_project_cmakelists
-
-        content = get_project_cmakelists(minimal=True)
+    content = _render(minimal=True)
 
     assert "ESPHOME_PROJECT_BUILTIN_COMPONENTS" not in content
 
@@ -115,13 +123,7 @@ def test_get_project_cmakelists_full_emits_builtin_components_property(
         },
     )
 
-    with (
-        patch("esphome.build_gen.espidf.get_esp32_variant", return_value="ESP32"),
-        patch.object(CORE, "name", "test"),
-    ):
-        from esphome.build_gen.espidf import get_project_cmakelists
-
-        content = get_project_cmakelists(minimal=False)
+    content = _render()
 
     assert (
         "idf_build_set_property(ESPHOME_PROJECT_BUILTIN_COMPONENTS esp_lcd APPEND)"
@@ -136,6 +138,118 @@ def test_get_project_cmakelists_full_emits_builtin_components_property(
     assert "JPEGDEC APPEND" not in content
 
 
+def test_get_project_cmakelists_emits_cmake_args() -> None:
+    """Args registered via CORE.add_cmake_arg() are emitted as set() lines,
+    on minimal writes too."""
+    CORE.add_cmake_arg("EXECUTABLE_COMPONENT_NAME", "src")
+
+    content = _render(minimal=True)
+
+    assert 'set(EXECUTABLE_COMPONENT_NAME "src")' in content
+
+
+def test_get_project_cmakelists_escapes_backslashes_in_cmake_args() -> None:
+    """Backslashes (the only character escaping applies to; the rest are
+    rejected at registration) are doubled so CMake reads the value back
+    verbatim."""
+    CORE.add_cmake_arg("MY_PATH", r"C:\esp\idf")
+
+    content = _render(minimal=True)
+
+    assert r'set(MY_PATH "C:\\esp\\idf")' in content
+
+
+def test_get_project_cmakelists_emits_exclude_components(tmp_path: Path) -> None:
+    """Excluded components are passed to IDF via EXCLUDE_COMPONENTS and are
+    dropped from ESPHOME_PROJECT_BUILTIN_COMPONENTS even when a stale
+    project_description.json still lists them (requiring an excluded
+    component would pull it back into the build)."""
+    _write_project_description(
+        tmp_path,
+        {
+            "esp_lcd": "/idf/components/esp_lcd",
+            "freertos": "/idf/components/freertos",
+            "unity": "/idf/components/unity",
+        },
+    )
+    CORE.data[KEY_ESP32][KEY_EXCLUDE_COMPONENTS] = {"unity", "esp_lcd"}
+    register_exclude_components_cmake_arg()
+
+    content = _render()
+
+    assert 'set(EXCLUDE_COMPONENTS "esp_lcd;unity")' in content
+    # Must be set before project() so project.cmake sees it.
+    assert content.index("set(EXCLUDE_COMPONENTS") < content.index("project(test)")
+    assert (
+        "idf_build_set_property(ESPHOME_PROJECT_BUILTIN_COMPONENTS freertos APPEND)"
+        in content
+    )
+    assert "ESPHOME_PROJECT_BUILTIN_COMPONENTS unity" not in content
+    assert "ESPHOME_PROJECT_BUILTIN_COMPONENTS esp_lcd" not in content
+
+
+def test_get_project_cmakelists_minimal_emits_exclude_components() -> None:
+    """The discovery (minimal) write also excludes components so they never
+    register in project_description.json."""
+    CORE.data[KEY_ESP32][KEY_EXCLUDE_COMPONENTS] = {"unity"}
+    register_exclude_components_cmake_arg()
+
+    content = _render(minimal=True)
+
+    assert 'set(EXCLUDE_COMPONENTS "unity")' in content
+
+
+def test_get_project_cmakelists_no_exclude_components_line_when_empty() -> None:
+    """No EXCLUDE_COMPONENTS line at all when nothing is excluded."""
+    register_exclude_components_cmake_arg()
+
+    content = _render()
+
+    assert "EXCLUDE_COMPONENTS" not in content
+
+
+def test_include_builtin_idf_component_removes_exclusion() -> None:
+    """include_builtin_idf_component() drops a name from the exclusion set so
+    a component a config actually uses is not passed to EXCLUDE_COMPONENTS."""
+    from esphome.components.esp32 import (
+        exclude_builtin_idf_component,
+        get_excluded_builtin_components,
+        include_builtin_idf_component,
+    )
+
+    exclude_builtin_idf_component("esp_eth")
+    exclude_builtin_idf_component("unity")
+    include_builtin_idf_component("esp_eth")
+
+    assert get_excluded_builtin_components() == ["unity"]
+
+    register_exclude_components_cmake_arg()
+    content = _render()
+
+    assert 'set(EXCLUDE_COMPONENTS "unity")' in content
+    assert "esp_eth" not in content
+
+
+def test_write_project_writes_exclude_components_stamp(tmp_path: Path) -> None:
+    """write_project() snapshots the exclusion set; the toolchain watches the
+    stamp to trigger a discovery reconfigure when the set changes (excluded
+    components never register in project_description.json)."""
+    CORE.build_flags = set()
+    CORE.build_path = tmp_path
+    CORE.data[KEY_ESP32][KEY_EXCLUDE_COMPONENTS] = {"unity", "esp_lcd"}
+
+    with (
+        patch("esphome.build_gen.espidf.get_esp32_variant", return_value="ESP32"),
+        patch.object(CORE, "name", "test"),
+    ):
+        from esphome.build_gen.espidf import write_project
+
+        write_project()
+
+    stamp = tmp_path / "exclude_components.esphomeinternal"
+    assert stamp.read_text() == "esp_lcd;unity"
+
+
 def test_get_component_cmakelists_no_link_flags() -> None:
     """With no -Wl, flags the target_link_options block is emitted with an empty body."""
     CORE.build_flags = set()

tests/unit_tests/build_gen/test_platformio.py

@@ -169,6 +169,7 @@ def clean_core(monkeypatch: pytest.MonkeyPatch) -> None:
     monkeypatch.setattr(CORE, "platformio_libraries", {})
     monkeypatch.setattr(CORE, "build_flags", set())
     monkeypatch.setattr(CORE, "build_unflags", set())
+    monkeypatch.setattr(CORE, "cmake_args", {})
 
 
 def test_get_ini_content_pins_cpp_standard(
@@ -202,6 +203,49 @@ def test_get_ini_content_no_cpp_standard(
     assert "-std=" not in content
 
 
+def test_get_ini_content_emits_cmake_args(
+    clean_core: None, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """Registered args are space-joined into one option, sorted by name."""
+    monkeypatch.setattr(
+        CORE,
+        "cmake_args",
+        {"EXECUTABLE_COMPONENT_NAME": "src", "EXCLUDE_COMPONENTS": "unity"},
+    )
+
+    content = platformio.get_ini_content()
+
+    assert (
+        "board_build.cmake_extra_args = "
+        "-DEXCLUDE_COMPONENTS=unity -DEXECUTABLE_COMPONENT_NAME=src" in content
+    )
+
+
+def test_get_ini_content_no_cmake_option_when_no_args(clean_core: None) -> None:
+    """No board_build.cmake_extra_args line at all when nothing registered
+    (ESP8266/RP2040/LibreTiny builds must not get a blank option)."""
+    content = platformio.get_ini_content()
+
+    assert "board_build.cmake_extra_args" not in content
+
+
+def test_get_ini_content_overwrites_list_valued_user_cmake_option(
+    clean_core: None, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """A user-supplied board_build.cmake_extra_args may be a list; the
+    registered args must replace it without tripping add_platformio_option's
+    list-append assert."""
+    monkeypatch.setattr(
+        CORE, "platformio_options", {"board_build.cmake_extra_args": ["-DFOO=1"]}
+    )
+    monkeypatch.setattr(CORE, "cmake_args", {"EXECUTABLE_COMPONENT_NAME": "src"})
+
+    content = platformio.get_ini_content()
+
+    assert "board_build.cmake_extra_args = -DEXECUTABLE_COMPONENT_NAME=src" in content
+    assert "-DFOO=1" not in content
+
+
 def test_write_cxx_flags_script_emits_registered_flags(
     tmp_path: Path, monkeypatch: pytest.MonkeyPatch
 ) -> None:

tests/unit_tests/test_bundle.py

@@ -29,8 +29,9 @@ from esphome.bundle import (
     read_bundle_manifest,
     remap_bundle_path,
 )
+from esphome.components.substitutions import do_substitution_pass
 from esphome.core import CORE, EsphomeError
-from esphome.yaml_util import force_load_include_files
+from esphome.yaml_util import force_load_include_files, load_yaml
 
 # ---------------------------------------------------------------------------
 # Helpers
@@ -1277,6 +1278,59 @@ def test_discover_files_bundles_all_include_candidates(tmp_path: Path) -> None:
     assert "includes/empty.yaml" in paths
 
 
+@pytest.mark.parametrize("enable_proxy", [True, False])
+def test_bundle_roundtrip_templated_include_with_path_separator(
+    tmp_path: Path, enable_proxy: bool
+) -> None:
+    r"""The issue-18545 flow: a Jinja !include whose branches contain "/" still
+    resolves after the bundle is extracted on the build server.
+
+    Windows is the leg that regresses: the raw expression text must survive
+    verbatim, or its separators get rewritten to "\" and Jinja decodes
+    sequences like "\b" as string escapes.
+    """
+    config_dir = _setup_config_dir(
+        tmp_path,
+        files={
+            "includes/boards/board.yaml": (
+                "packages:\n"
+                '  - !include ${ "bluetooth/bluetooth_proxy_single_core.yaml"'
+                ' if enable_bluetooth_proxy else "../empty.yaml" }\n'
+            ),
+            "includes/boards/bluetooth/bluetooth_proxy_single_core.yaml": (
+                "bluetooth_proxy:\n  active: true\n"
+            ),
+            "includes/empty.yaml": "{}\n",
+        },
+    )
+    (config_dir / "test.yaml").write_text(
+        "substitutions:\n"
+        f"  enable_bluetooth_proxy: {str(enable_proxy).lower()}\n"
+        "esphome:\n  name: test\n"
+        "packages:\n  - !include includes/boards/board.yaml\n"
+    )
+
+    result = ConfigBundleCreator({}).create_bundle()
+    bundle_path = tmp_path / "device.esphomebundle.tar.gz"
+    bundle_path.write_bytes(result.data)
+
+    # Both conditional branches must ship in the bundle.
+    paths = [f.path for f in result.files]
+    assert "includes/boards/bluetooth/bluetooth_proxy_single_core.yaml" in paths
+    assert "includes/empty.yaml" in paths
+
+    # Extract to a fresh directory and resolve the config from there, as a
+    # remote build server would.
+    extracted_config = extract_bundle(bundle_path, tmp_path / "remote")
+    config = do_substitution_pass(load_yaml(extracted_config))
+
+    board_pkg = config["packages"][0]["packages"][0]
+    if enable_proxy:
+        assert board_pkg == {"bluetooth_proxy": {"active": True}}
+    else:
+        assert board_pkg == {}
+
+
 def test_discover_files_candidate_outside_config_dir_skipped(
     tmp_path: Path, caplog: pytest.LogCaptureFixture
 ) -> None:

tests/unit_tests/test_config_helpers.py

@@ -6,6 +6,7 @@ from unittest.mock import patch
 import pytest
 
 from esphome.config_helpers import (
+    filter_source_files_from_defines,
     filter_source_files_from_platform,
     frameworks_for_platforms,
     get_logger_level,
@@ -18,6 +19,7 @@ from esphome.const import (
     KEY_TARGET_PLATFORM,
     PlatformFramework,
 )
+from esphome.core import Define
 
 
 def test_filter_source_files_from_platform_esp32() -> None:
@@ -148,3 +150,25 @@ def test_frameworks_for_platforms_derives_and_rejects_unknown() -> None:
     }
     with pytest.raises(ValueError, match="unknown platform"):
         frameworks_for_platforms(["esp32", "not_a_platform"])
+
+
+def test_filter_source_files_from_defines() -> None:
+    """Files are excluded unless one of their defines is set."""
+    files_map: dict[str, str | tuple[str, ...]] = {
+        "filter.cpp": "USE_SENSOR_FILTER",
+        "automation.cpp": ("USE_CLICK", "USE_MULTI_CLICK"),
+    }
+    filter_func: Callable[[], list[str]] = filter_source_files_from_defines(files_map)
+
+    with patch("esphome.config_helpers.CORE") as mock_core:
+        mock_core.defines = {Define("USE_SENSOR_FILTER")}
+        assert filter_func() == ["automation.cpp"]
+
+        mock_core.defines = {Define("USE_MULTI_CLICK")}
+        assert filter_func() == ["filter.cpp"]
+
+        mock_core.defines = {Define("USE_SENSOR_FILTER"), Define("USE_CLICK")}
+        assert filter_func() == []
+
+        mock_core.defines = set()
+        assert sorted(filter_func()) == ["automation.cpp", "filter.cpp"]

tests/unit_tests/test_core.py

@@ -990,3 +990,35 @@ class TestEsphomeCore:
         )
         # The unflag is still recorded either way.
         assert target.build_unflags == {"-fno-rtti", "-fno-exceptions"}
+
+    def test_add_cmake_arg(self, target) -> None:
+        target.add_cmake_arg("EXCLUDE_COMPONENTS", "unity;esp_lcd")
+        assert target.cmake_args == {"EXCLUDE_COMPONENTS": "unity;esp_lcd"}
+
+    @pytest.mark.parametrize("name", ["", "BAD NAME", 'A"B', "A(B)", "1ABC"])
+    def test_add_cmake_arg__rejects_invalid_name(self, target, name: str) -> None:
+        with pytest.raises(ValueError, match="Invalid CMake arg name"):
+            target.add_cmake_arg(name, "value")
+
+    @pytest.mark.parametrize("value", ["a b", "a\tb", 'a"b', "a'b", "a${FOO}b"])
+    def test_add_cmake_arg__rejects_invalid_value(self, target, value: str) -> None:
+        """Whitespace and quotes are rejected (the PlatformIO backend passes
+        args as one space-joined string, which would split such a value), and
+        so is '$' (expanded differently by CMake and PlatformIO)."""
+        with pytest.raises(ValueError, match="must not contain"):
+            target.add_cmake_arg("MY_ARG", value)
+
+    def test_add_cmake_arg__warns_on_overwrite(
+        self, target, caplog: pytest.LogCaptureFixture
+    ) -> None:
+        """Re-registering with a different value is last-writer-wins; warn so
+        the silently dropped value is diagnosable."""
+        target.add_cmake_arg("MY_ARG", "one")
+        target.add_cmake_arg("MY_ARG", "one")
+        assert "overwriting" not in caplog.text
+
+        target.add_cmake_arg("MY_ARG", "two")
+        assert (
+            "CMake arg MY_ARG already set to one; overwriting with two" in caplog.text
+        )
+        assert target.cmake_args == {"MY_ARG": "two"}

tests/unit_tests/test_espidf_toolchain.py

@@ -100,6 +100,33 @@ def _setup_build(setup_core: Path) -> tuple[Path, Path]:
     return compile_commands, cache
 
 
+def test_has_outdated_files_detects_exclusion_change(setup_core: Path) -> None:
+    """A newer exclude_components.esphomeinternal stamp forces a reconfigure
+    so components that leave the exclusion set get rediscovered."""
+    CORE.build_path = setup_core
+    build = setup_core / "build"
+    (build / "config").mkdir(parents=True)
+    (build / "config" / "sdkconfig.h").write_text("")
+    cmakecache = build / "CMakeCache.txt"
+    cmakecache.write_text("")
+    (build / "build.ninja").write_text("")
+
+    with patch.object(CORE, "name", "test"):
+        assert not toolchain.has_outdated_files()
+
+        stamp = setup_core / "exclude_components.esphomeinternal"
+        stamp.write_text("unity")
+        os.utime(stamp, (cmakecache.stat().st_mtime + 10,) * 2)
+
+        assert toolchain.has_outdated_files()
+
+        # The flag must clear once the reference file is restamped (as
+        # run_compile does after a successful discovery reconfigure);
+        # otherwise every later build would repeat the discovery pass.
+        os.utime(cmakecache, (stamp.stat().st_mtime + 10,) * 2)
+        assert not toolchain.has_outdated_files()
+
+
 def test_get_idedata_returns_none_without_compile_commands(setup_core: Path) -> None:
     """No compile DB yet -> None (rather than an error)."""
     _setup_build(setup_core)
@@ -373,6 +400,48 @@ def test_run_idf_py_jobs_sets_build_jobs_env(setup_core: Path) -> None:
         assert "IDF_PY_BUILD_JOBS" not in env
 
 
+def test_run_compile_restamps_cmakecache_after_discovery(setup_core: Path) -> None:
+    """After a successful discovery reconfigure the reference CMakeCache.txt
+    is restamped; cmake does not rewrite it when only properties or plain
+    variables change, so the staleness flag would otherwise never clear."""
+    _setup_build(setup_core)
+    config = {CONF_ESPHOME: {}}
+    cmakecache = CORE.relative_build_path("build/CMakeCache.txt")
+    cmakecache.parent.mkdir(parents=True, exist_ok=True)
+    cmakecache.write_text("")
+    old = cmakecache.stat().st_mtime - 100
+    os.utime(cmakecache, (old, old))
+
+    with (
+        patch.object(toolchain, "need_reconfigure", return_value=True),
+        patch("esphome.build_gen.espidf.write_project"),
+        patch.object(toolchain, "run_reconfigure", return_value=0),
+        patch.object(toolchain, "run_idf_py", return_value=0),
+        patch.object(toolchain, "print_summary"),
+    ):
+        assert toolchain.run_compile(config, verbose=False) == 0
+
+    assert cmakecache.stat().st_mtime > old
+
+
+def test_run_compile_discovery_without_cmakecache(setup_core: Path) -> None:
+    """A discovery pass that produced no CMakeCache.txt (nothing to restamp)
+    still completes normally."""
+    _setup_build(setup_core)
+    config = {CONF_ESPHOME: {}}
+
+    with (
+        patch.object(toolchain, "need_reconfigure", return_value=True),
+        patch("esphome.build_gen.espidf.write_project"),
+        patch.object(toolchain, "run_reconfigure", return_value=0),
+        patch.object(toolchain, "run_idf_py", return_value=0),
+        patch.object(toolchain, "print_summary"),
+    ):
+        assert toolchain.run_compile(config, verbose=False) == 0
+
+    assert not CORE.relative_build_path("build/CMakeCache.txt").exists()
+
+
 def test_run_compile_passes_compile_process_limit(setup_core: Path) -> None:
     """compile_process_limit is forwarded to run_idf_py as the job limit."""
     _setup_build(setup_core)

tests/unit_tests/test_espota2.py

@@ -6,6 +6,8 @@ from collections.abc import Generator
 import gzip
 import hashlib
 import io
+import itertools
+import logging
 from pathlib import Path
 import socket
 import struct
@@ -53,8 +55,9 @@ def mock_sleep() -> Generator[Mock]:
 @pytest.fixture
 def mock_time(mock_sleep: Mock) -> Generator[None]:
     """Mock time-related functions for consistent testing."""
-    # Provide enough values for multiple calls (tests may call perform_ota multiple times)
-    with patch("time.perf_counter", side_effect=[0, 1, 0, 1, 0, 1]):
+    # Monotonically increasing, never exhausted regardless of how many timing
+    # windows perform_ota measures or how many times a test calls it
+    with patch("time.perf_counter", side_effect=itertools.count()):
         yield
 
 
@@ -372,7 +375,9 @@ def test_perform_ota_successful_md5_auth(
 
 
 @pytest.mark.usefixtures("mock_time")
-def test_perform_ota_no_auth(mock_socket: Mock, mock_file: io.BytesIO) -> None:
+def test_perform_ota_no_auth(
+    mock_socket: Mock, mock_file: io.BytesIO, caplog: pytest.LogCaptureFixture
+) -> None:
     """Test OTA without authentication."""
     recv_responses = [
         bytes([espota2.RESPONSE_OK]),  # First byte of version response
@@ -387,7 +392,14 @@ def test_perform_ota_no_auth(mock_socket: Mock, mock_file: io.BytesIO) -> None:
 
     mock_socket.recv.side_effect = recv_responses
 
-    espota2.perform_ota(mock_socket, None, mock_file, "test.bin")
+    # Distinct window lengths pin each duration to its label; exactly the 6
+    # expected perf_counter calls, so an unaccounted timing window raises
+    timings = [0.0, 2.0, 10.0, 15.0, 20.0, 27.0]
+    with (
+        patch("time.perf_counter", side_effect=timings),
+        caplog.at_level(logging.INFO),
+    ):
+        espota2.perform_ota(mock_socket, None, mock_file, "test.bin")
 
     # Should not send any auth-related data
     auth_calls = [
@@ -397,6 +409,14 @@ def test_perform_ota_no_auth(mock_socket: Mock, mock_file: io.BytesIO) -> None:
     ]
     assert len(auth_calls) == 0
 
+    # The timing summary is the observable output of the upload; exact strings
+    # pin each duration to its label
+    assert "Preparing for upload took 2.00 seconds" in caplog.text
+    assert (
+        "Update took 14.00 seconds (prepare 2.00, upload 5.00, commit 7.00)"
+        in caplog.text
+    )
+
 
 @pytest.mark.usefixtures("mock_time")
 def test_perform_ota_with_compression(mock_socket: Mock) -> None:

tests/unit_tests/test_external_files.py

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

tests/unit_tests/test_framework_helpers.py

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

tests/unit_tests/test_helpers.py

@@ -171,6 +171,13 @@ def test_is_ip_address__valid(value):
         ("FOO", "fAlSe", True, False),
         ("FOO", "Yes", False, True),
         ("FOO", "123", False, True),
+        # cv.boolean's spellings; falsy rows use default=True on purpose
+        ("FOO", "on", False, True),
+        ("FOO", "enable", False, True),
+        ("FOO", "no", True, False),
+        ("FOO", "off", True, False),
+        ("FOO", "OFF", True, False),
+        ("FOO", "Disable", True, False),
     ),
 )
 def test_get_bool_env(monkeypatch, var, value, default, expected):
@@ -253,6 +260,31 @@ class Test_write_file_if_changed:
 
         assert dst.read_text() == text
 
+    def test_damaged_existing_file_is_replaced(
+        self, tmp_path: Path, caplog: pytest.LogCaptureFixture
+    ):
+        """A non-UTF-8 existing file is logged and overwritten."""
+        dst = tmp_path / "generated.txt"
+        dst.write_bytes(b"\xff\xfe")
+
+        assert helpers.write_file_if_changed(dst, "fresh content") is True
+
+        assert dst.read_text(encoding="utf-8") == "fresh content"
+        assert "Replacing damaged file" in caplog.text
+
+    def test_unreadable_existing_file_still_raises(self, tmp_path: Path):
+        """An OSError on the comparison read still raises EsphomeError."""
+        dst = tmp_path / "generated.txt"
+        dst.write_text("intact")
+
+        with (
+            patch.object(Path, "read_text", side_effect=OSError("permission denied")),
+            pytest.raises(EsphomeError, match="Error reading file"),
+        ):
+            helpers.write_file_if_changed(dst, "fresh content")
+
+        assert dst.exists()
+
     def test_dst_does_not_exist(self, tmp_path: Path):
         text = "A files are unique.\n"
         dst = tmp_path / "file-a.txt"

tests/unit_tests/test_log.py

@@ -1,5 +1,4 @@
 from collections.abc import Generator
-import errno
 import io
 import logging
 import os
@@ -178,37 +177,34 @@ def _run_probe_on_pty(
     output = b""
     deadline = time.monotonic() + 60
     try:
-        try:
-            proc = subprocess.Popen(
-                _probe_command(fixture_path),
-                stdout=follower,
-                stderr=follower if stderr_to_pty else subprocess.PIPE,
-                stdin=follower,
-                env=probe_env,
-            )
-        finally:
-            os.close(follower)
-        while True:
-            timeout = deadline - time.monotonic()
-            if timeout <= 0 or not select.select([controller], [], [], timeout)[0]:
-                pytest.fail(f"pty probe produced no EOF in time; got {output!r}")
-            try:
-                chunk = os.read(controller, 1024)
-            except OSError as err:
-                # macOS raises EIO once the child closes its end of the pty;
-                # anything else is a real failure, not end-of-stream.
-                if err.errno != errno.EIO:
-                    raise
-                break
-            if not chunk:
-                break
+        proc = subprocess.Popen(
+            _probe_command(fixture_path),
+            stdout=follower,
+            stderr=follower if stderr_to_pty else subprocess.PIPE,
+            stdin=follower,
+            env=probe_env,
+        )
+        # The parent keeps the follower open until the child has exited and
+        # the controller is drained: macOS discards buffered pty output once
+        # the last follower closes, so closing it early loses the probe's
+        # output whenever the child finishes before the first read.
+        while proc.poll() is None:
+            if time.monotonic() > deadline:
+                pytest.fail(f"pty probe did not exit in time; got {output!r}")
+            if select.select([controller], [], [], 0.01)[0]:
+                output += os.read(controller, 4096)
+        # Everything the child wrote is already buffered, so drain without waiting.
+        while select.select([controller], [], [], 0)[0] and (
+            chunk := os.read(controller, 4096)
+        ):
             output += chunk
         stderr_text = ""
         if proc.stderr is not None:
             stderr_text = proc.stderr.read().decode(errors="replace")
             proc.stderr.close()
-        assert proc.wait(60) == 0, stderr_text
+        assert proc.returncode == 0, stderr_text
     finally:
+        os.close(follower)
         os.close(controller)
         if proc is not None and proc.poll() is None:
             proc.kill()

tests/unit_tests/test_main.py

@@ -11,6 +11,7 @@ from pathlib import Path
 import re
 import sys
 import time
+from types import SimpleNamespace
 from typing import Any, Self
 from unittest.mock import AsyncMock, MagicMock, Mock, patch
 
@@ -18,7 +19,7 @@ import pytest
 from pytest import CaptureFixture
 from zeroconf import ServiceStateChange
 
-from esphome import __main__ as main
+from esphome import __main__ as main, yaml_util
 from esphome.__main__ import (
     Purpose,
     _get_configured_xtal_freq,
@@ -29,6 +30,7 @@ from esphome.__main__ import (
     _unresolved_default_error,
     _validate_bootloader_binary,
     _validate_partition_table_binary,
+    _wrap_to_code,
     check_permissions,
     choose_upload_log_host,
     command_analyze_memory,
@@ -116,6 +118,7 @@ from esphome.espota2 import (
     OTA_TYPE_UPDATE_PARTITION_TABLE,
 )
 from esphome.platformio import toolchain
+from esphome.types import ConfigType
 from esphome.util import BootselResult, FlashImage
 from esphome.zeroconf import _await_discovery, discover_mdns_devices
 
@@ -7130,3 +7133,28 @@ def test_warn_source_tree_mismatch_falls_back_when_stat_fails(
 
     # Same tree, so the path comparison still finds them equal and stays silent
     assert not caplog.text
+
+
+@pytest.mark.asyncio
+async def test_wrap_to_code_comment_is_insertion_order_independent() -> None:
+    """The config comment dumps with sorted keys: voluptuous fills schema
+    defaults in set-iteration order, so an unsorted dump would churn
+    main.cpp and relink the firmware on every run."""
+    comments: list[str] = []
+
+    async def to_code(conf: ConfigType) -> None:
+        """Accept any config; only the wrapper's comment output matters."""
+
+    comp = SimpleNamespace(to_code=to_code, config_schema=object())
+    wrapped = _wrap_to_code("demo", comp, yaml_util)
+    with patch("esphome.codegen.add", side_effect=lambda st: comments.append(str(st))):
+        # Nested on purpose: the real churn lives in nested action configs,
+        # so sorting must apply at every mapping level
+        await wrapped({"beta": 1, "alpha": {"z": 1, "a": 2}})
+        first = "\n".join(comments)
+        comments.clear()
+        await wrapped({"alpha": {"a": 2, "z": 1}, "beta": 1})
+    second = "\n".join(comments)
+    assert first == second
+    assert second.index("alpha") < second.index("beta")
+    assert second.index("a: 2") < second.index("z: 1")

tests/unit_tests/test_net_retry.py

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

tests/unit_tests/test_nrf52_framework.py

@@ -16,6 +16,7 @@ from esphome.components.nrf52.framework import (
     _get_penv_site_packages,
     _get_platformio_penv_path,
     _get_toolchain_platform_info,
+    _needs_venv_rebuild,
     check_and_install,
     get_build_env,
     get_sdk_nrf_tools_path,
@@ -123,10 +124,19 @@ def mock_nrf52_ops():
 # ---------------------------------------------------------------------------
 
 
+def _touch_penv_python(penv: Path) -> None:
+    """Create the interpreter file so the rebuild gate sees a live venv."""
+    python = get_python_env_executable_path(penv, "python")
+    python.parent.mkdir(parents=True, exist_ok=True)
+    python.touch()
+
+
 def _mark_venv_ready(python_env: Path) -> None:
-    """Write the venv sentinel with the current requirements hash."""
+    """Write the venv sentinel with the current requirements hash and a
+    present interpreter so the rebuild gate passes."""
     requirements_hash = hashlib.sha256(_REQUIREMENTS.read_bytes()).hexdigest()
     (python_env / ".ready").write_text(requirements_hash, encoding="utf-8")
+    _touch_penv_python(python_env)
 
 
 class TestCheckAndInstall:
@@ -148,6 +158,23 @@ class TestCheckAndInstall:
         mock_nrf52_ops.download_from_mirrors.assert_not_called()
         mock_nrf52_ops.archive_extract_all.assert_not_called()
 
+    def test_missing_interpreter_rebuilds_venv(
+        self,
+        nrf52_dirs: SimpleNamespace,
+        mock_nrf52_ops: SimpleNamespace,
+    ) -> None:
+        """A valid sentinel must not mask a missing interpreter (a cached venv
+        restored after a host interpreter upgrade)."""
+        requirements_hash = hashlib.sha256(_REQUIREMENTS.read_bytes()).hexdigest()
+        (nrf52_dirs.python_env / ".ready").write_text(
+            requirements_hash, encoding="utf-8"
+        )
+        # no interpreter on disk
+
+        check_and_install()
+
+        mock_nrf52_ops.create_venv.assert_called_once()
+
     def test_fresh_install_runs_all_steps(
         self,
         nrf52_dirs: SimpleNamespace,
@@ -348,6 +375,7 @@ class TestSetupPlatformioPythonEnv:
         (platformio_penv_dir / ".ready").write_text(
             _platformio_requirements_hash(), encoding="utf-8"
         )
+        _touch_penv_python(platformio_penv_dir)
 
         with patch.dict(os.environ):
             setup_platformio_python_env()
@@ -392,6 +420,22 @@ class TestSetupPlatformioPythonEnv:
 
         assert not (platformio_penv_dir / ".ready").exists()
 
+    def test_missing_interpreter_reinstalls(
+        self,
+        platformio_penv_dir: Path,
+        mock_nrf52_ops: SimpleNamespace,
+    ) -> None:
+        """A valid sentinel must not mask a missing interpreter."""
+        (platformio_penv_dir / ".ready").write_text(
+            _platformio_requirements_hash(), encoding="utf-8"
+        )
+        # no interpreter on disk
+
+        with patch.dict(os.environ):
+            setup_platformio_python_env()
+
+        mock_nrf52_ops.create_venv.assert_called_once()
+
     def test_repeated_calls_do_not_duplicate_env_entries(
         self,
         platformio_penv_dir: Path,
@@ -401,6 +445,7 @@ class TestSetupPlatformioPythonEnv:
         (platformio_penv_dir / ".ready").write_text(
             _platformio_requirements_hash(), encoding="utf-8"
         )
+        _touch_penv_python(platformio_penv_dir)
         site_packages = str(_get_penv_site_packages(platformio_penv_dir))
         bin_dir = str(
             get_python_env_executable_path(platformio_penv_dir, "python").parent
@@ -422,6 +467,7 @@ class TestSetupPlatformioPythonEnv:
         (platformio_penv_dir / ".ready").write_text(
             _platformio_requirements_hash(), encoding="utf-8"
         )
+        _touch_penv_python(platformio_penv_dir)
         site_packages = str(_get_penv_site_packages(platformio_penv_dir))
 
         with patch.dict(os.environ, {"PYTHONPATH": "/existing/path"}):
@@ -531,3 +577,45 @@ def testget_tools_path_default_is_global_cache(
         Path(platformdirs.user_cache_dir("esphome", appauthor=False)) / "sdk-nrf"
     ).resolve()
     assert get_sdk_nrf_tools_path() == expected
+
+
+def test_needs_venv_rebuild_gates(tmp_path: Path) -> None:
+    """The shared penv gate rebuilds on any missing or stale piece."""
+    penv = tmp_path / "penv"
+    penv.mkdir()
+    python = penv / "python"
+    sentinel = penv / ".ready"
+    good_hash = "abc123"
+
+    # Nothing in place yet
+    assert _needs_venv_rebuild(python, sentinel, good_hash)
+
+    python.write_text("")
+    # Interpreter present but no sentinel
+    assert _needs_venv_rebuild(python, sentinel, good_hash)
+
+    sentinel.write_text(good_hash, encoding="utf-8")
+    # Everything in place
+    assert not _needs_venv_rebuild(python, sentinel, good_hash)
+
+    # Stale requirements hash
+    assert _needs_venv_rebuild(python, sentinel, "otherhash")
+
+
+@pytest.mark.skipif(
+    sys.platform == "win32", reason="symlink creation needs privileges on Windows"
+)
+def test_needs_venv_rebuild_on_dangling_interpreter_symlink(tmp_path: Path) -> None:
+    """A cached venv restored after a host interpreter upgrade has a
+    bin/python symlink whose target is gone; the valid sentinel must not
+    mask it."""
+    penv = tmp_path / "penv"
+    penv.mkdir()
+    python = penv / "python"
+    sentinel = penv / ".ready"
+    sentinel.write_text("abc123", encoding="utf-8")
+    python.symlink_to(tmp_path / "hostedtoolcache" / "3.12.14" / "python3")
+    assert python.is_symlink()
+    assert not python.exists()
+
+    assert _needs_venv_rebuild(python, sentinel, "abc123")

tests/unit_tests/test_substitutions.py

@@ -744,6 +744,25 @@ def test_include_filename_substitution_undefined_var(tmp_path: Path) -> None:
         substitutions.do_substitution_pass(config)
 
 
+def test_include_filename_jinja_expression_with_path_separator(
+    tmp_path: Path,
+) -> None:
+    """A jinja !include whose string literals contain "/" resolves correctly (issue #18545)."""
+    main_file = tmp_path / "main.yaml"
+    main_file.write_text(
+        "substitutions:\n"
+        "  enable_bluetooth_proxy: true\n"
+        "result: !include "
+        '${ "bluetooth/proxy.yaml" if enable_bluetooth_proxy else "../empty.yaml" }\n'
+    )
+    (tmp_path / "bluetooth").mkdir()
+    (tmp_path / "bluetooth" / "proxy.yaml").write_text("value: 42\n")
+
+    config = yaml_util.load_yaml(main_file)
+    config = substitutions.do_substitution_pass(config)
+    assert config["result"] == {"value": 42}
+
+
 def test_raise_first_undefined_logs_extras_at_debug(
     caplog: pytest.LogCaptureFixture,
 ) -> None:

tests/unit_tests/test_writer.py

@@ -2041,6 +2041,38 @@ def test_copy_src_tree_writes_build_info_files(
     assert build_info_json["esphome_version"] == "2025.1.0-dev"
 
 
+def _setup_build_info_mocks(
+    mock_core: MagicMock,
+    mock_iter_components: MagicMock,
+    mock_walk_files: MagicMock,
+    tmp_path: Path,
+) -> Path:
+    """Point CORE at tmp_path and return the build_info.json path."""
+    src_path = tmp_path / "src"
+    (src_path / "esphome" / "core").mkdir(parents=True)
+    build_path = tmp_path / "build"
+    build_path.mkdir()
+    mock_core.relative_src_path.side_effect = src_path.joinpath
+    mock_core.relative_build_path.side_effect = build_path.joinpath
+    mock_core.defines = []
+    mock_core.config_hash = 0xDEADBEEF
+    mock_core.comment = ""
+    mock_core.target_platform = "test_platform"
+    mock_core.config = {}
+    mock_iter_components.return_value = []
+    mock_walk_files.return_value = []
+    return build_path / "build_info.json"
+
+
+def _run_copy_src_tree(version: str = "2025.1.0-dev") -> None:
+    with (
+        patch("esphome.writer.__version__", version),
+        patch("esphome.writer.importlib.import_module") as mock_import,
+    ):
+        mock_import.side_effect = AttributeError
+        copy_src_tree()
+
+
 @patch("esphome.writer.CORE")
 @patch("esphome.writer.iter_components")
 @patch("esphome.writer.walk_files")
@@ -2050,59 +2082,17 @@ def test_copy_src_tree_detects_config_hash_change(
     mock_core: MagicMock,
     tmp_path: Path,
 ) -> None:
-    """Test copy_src_tree detects when config_hash changes."""
-    # Setup directory structure
-    src_path = tmp_path / "src"
-    src_path.mkdir()
-    esphome_core_path = src_path / "esphome" / "core"
-    esphome_core_path.mkdir(parents=True)
-    build_path = tmp_path / "build"
-    build_path.mkdir()
-
-    # Create existing build_info.json with different config_hash
-    build_info_json_path = build_path / "build_info.json"
-    build_info_json_path.write_text(
-        json.dumps(
-            {
-                "config_hash": 0x12345678,  # Different from current
-                "build_time": 1700000000,
-                "build_time_str": "2023-11-14 22:13:20 +0000",
-                "esphome_version": "2025.1.0-dev",
-            }
-        )
+    """A changed config_hash regenerates build_info after a steady-state run."""
+    build_info_json_path = _setup_build_info_mocks(
+        mock_core, mock_iter_components, mock_walk_files, tmp_path
     )
+    _run_copy_src_tree()
+    assert json.loads(build_info_json_path.read_text())["config_hash"] == 0xDEADBEEF
 
-    # Create existing build_info_data.h
-    build_info_h_path = esphome_core_path / "build_info_data.h"
-    build_info_h_path.write_text("// old build_info_data.h")
-
-    # Setup mocks
-    mock_core.relative_src_path.side_effect = src_path.joinpath
-    mock_core.relative_build_path.side_effect = build_path.joinpath
-    mock_core.defines = []
-    mock_core.config_hash = 0xDEADBEEF  # Different from existing
-    mock_core.comment = ""
-    mock_core.target_platform = "test_platform"
-    mock_core.config = {}
-    mock_iter_components.return_value = []
-    mock_walk_files.return_value = []
-
-    with (
-        patch("esphome.writer.__version__", "2025.1.0-dev"),
-        patch("esphome.writer.importlib.import_module") as mock_import,
-    ):
-        mock_import.side_effect = AttributeError
-        copy_src_tree()
-
-    # Verify build_info files were updated due to config_hash change
-    assert build_info_h_path.exists()
-    build_info_cpp_path = esphome_core_path / "build_info_data.cpp"
-    assert build_info_cpp_path.exists()
-    new_content = build_info_cpp_path.read_text()
-    assert "0xdeadbeef" in new_content.lower()
-
-    new_json = json.loads(build_info_json_path.read_text())
-    assert new_json["config_hash"] == 0xDEADBEEF
+    # Second run only regenerates if the hash comparison detects the change
+    mock_core.config_hash = 0xC0FFEE
+    _run_copy_src_tree()
+    assert json.loads(build_info_json_path.read_text())["config_hash"] == 0xC0FFEE
 
 
 @patch("esphome.writer.CORE")
@@ -2114,104 +2104,93 @@ def test_copy_src_tree_detects_version_change(
     mock_core: MagicMock,
     tmp_path: Path,
 ) -> None:
-    """Test copy_src_tree detects when esphome_version changes."""
-    # Setup directory structure
-    src_path = tmp_path / "src"
-    src_path.mkdir()
-    esphome_core_path = src_path / "esphome" / "core"
-    esphome_core_path.mkdir(parents=True)
-    build_path = tmp_path / "build"
-    build_path.mkdir()
-
-    # Create existing build_info.json with different version
-    build_info_json_path = build_path / "build_info.json"
-    build_info_json_path.write_text(
-        json.dumps(
-            {
-                "config_hash": 0xDEADBEEF,
-                "build_time": 1700000000,
-                "build_time_str": "2023-11-14 22:13:20 +0000",
-                "esphome_version": "2024.12.0",  # Old version
-            }
-        )
+    """A changed esphome_version regenerates build_info after a steady-state run."""
+    build_info_json_path = _setup_build_info_mocks(
+        mock_core, mock_iter_components, mock_walk_files, tmp_path
     )
-
-    # Create existing build_info_data.h
-    build_info_h_path = esphome_core_path / "build_info_data.h"
-    build_info_h_path.write_text("// old build_info_data.h")
-
-    # Setup mocks
-    mock_core.relative_src_path.side_effect = src_path.joinpath
-    mock_core.relative_build_path.side_effect = build_path.joinpath
-    mock_core.defines = []
-    mock_core.config_hash = 0xDEADBEEF
-    mock_core.comment = ""
-    mock_core.target_platform = "test_platform"
-    mock_core.config = {}
-    mock_iter_components.return_value = []
-    mock_walk_files.return_value = []
-
-    with (
-        patch("esphome.writer.__version__", "2025.1.0-dev"),  # New version
-        patch("esphome.writer.importlib.import_module") as mock_import,
-    ):
-        mock_import.side_effect = AttributeError
-        copy_src_tree()
-
-    # Verify build_info files were updated due to version change
-    assert build_info_h_path.exists()
+    # Pin version.h so only the build_info comparison can see the bump
+    with patch("esphome.writer.generate_version_h", return_value="// version.h\n"):
+        _run_copy_src_tree(version="2024.12.0")
+        _run_copy_src_tree(version="2025.1.0-dev")
     new_json = json.loads(build_info_json_path.read_text())
     assert new_json["esphome_version"] == "2025.1.0-dev"
 
 
+@pytest.mark.parametrize(
+    "damage",
+    (b"invalid json {{{", b"[]", b'\xff{"config_hash": 1}'),
+    ids=("invalid-json", "non-object", "non-utf8"),
+)
+@patch("esphome.writer.CORE")
+@patch("esphome.writer.iter_components")
+@patch("esphome.writer.walk_files")
+def test_copy_src_tree_regenerates_damaged_build_info(
+    mock_walk_files: MagicMock,
+    mock_iter_components: MagicMock,
+    mock_core: MagicMock,
+    tmp_path: Path,
+    damage: bytes,
+) -> None:
+    """A damaged build_info.json reads as stale and is regenerated, not left in place."""
+    build_info_json_path = _setup_build_info_mocks(
+        mock_core, mock_iter_components, mock_walk_files, tmp_path
+    )
+    _run_copy_src_tree()
+    build_info_json_path.write_bytes(damage)
+    # Second run only rewrites the file if the damage branch fires
+    _run_copy_src_tree()
+    new_json = json.loads(build_info_json_path.read_text())
+    assert new_json["config_hash"] == 0xDEADBEEF
+
+
 @patch("esphome.writer.CORE")
 @patch("esphome.writer.iter_components")
 @patch("esphome.writer.walk_files")
-def test_copy_src_tree_handles_invalid_build_info_json(
+def test_copy_src_tree_missing_build_info_rebuilds_quietly(
     mock_walk_files: MagicMock,
     mock_iter_components: MagicMock,
     mock_core: MagicMock,
     tmp_path: Path,
+    caplog: pytest.LogCaptureFixture,
 ) -> None:
-    """Test copy_src_tree handles invalid build_info.json gracefully."""
-    # Setup directory structure
-    src_path = tmp_path / "src"
-    src_path.mkdir()
-    esphome_core_path = src_path / "esphome" / "core"
-    esphome_core_path.mkdir(parents=True)
-    build_path = tmp_path / "build"
-    build_path.mkdir()
+    """An absent build_info.json regenerates without claiming damage."""
+    build_info_json_path = _setup_build_info_mocks(
+        mock_core, mock_iter_components, mock_walk_files, tmp_path
+    )
+    _run_copy_src_tree()
+    build_info_json_path.unlink()
+    _run_copy_src_tree()
+    assert json.loads(build_info_json_path.read_text())["config_hash"] == 0xDEADBEEF
+    assert "unreadable" not in caplog.text
 
-    # Create invalid build_info.json
-    build_info_json_path = build_path / "build_info.json"
+
+@patch("esphome.writer.CORE")
+@patch("esphome.writer.iter_components")
+@patch("esphome.writer.walk_files")
+def test_copy_src_tree_unremovable_damaged_build_info_is_logged(
+    mock_walk_files: MagicMock,
+    mock_iter_components: MagicMock,
+    mock_core: MagicMock,
+    tmp_path: Path,
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """A failed unlink of the damaged file names the real cause."""
+    build_info_json_path = _setup_build_info_mocks(
+        mock_core, mock_iter_components, mock_walk_files, tmp_path
+    )
+    _run_copy_src_tree()
     build_info_json_path.write_text("invalid json {{{")
+    real_unlink = Path.unlink
 
-    # Create existing build_info_data.h
-    build_info_h_path = esphome_core_path / "build_info_data.h"
-    build_info_h_path.write_text("// old build_info_data.h")
+    def fail_on_build_info(self: Path, missing_ok: bool = False) -> None:
+        if self.name == "build_info.json":
+            raise OSError("simulated EACCES")
+        real_unlink(self, missing_ok=missing_ok)
 
-    # Setup mocks
-    mock_core.relative_src_path.side_effect = src_path.joinpath
-    mock_core.relative_build_path.side_effect = build_path.joinpath
-    mock_core.defines = []
-    mock_core.config_hash = 0xDEADBEEF
-    mock_core.comment = ""
-    mock_core.target_platform = "test_platform"
-    mock_core.config = {}
-    mock_iter_components.return_value = []
-    mock_walk_files.return_value = []
-
-    with (
-        patch("esphome.writer.__version__", "2025.1.0-dev"),
-        patch("esphome.writer.importlib.import_module") as mock_import,
-    ):
-        mock_import.side_effect = AttributeError
-        copy_src_tree()
-
-    # Verify build_info files were created despite invalid JSON
-    assert build_info_h_path.exists()
-    new_json = json.loads(build_info_json_path.read_text())
-    assert new_json["config_hash"] == 0xDEADBEEF
+    with patch.object(Path, "unlink", fail_on_build_info):
+        _run_copy_src_tree()
+    assert "Could not remove unreadable build_info.json" in caplog.text
+    assert json.loads(build_info_json_path.read_text())["config_hash"] == 0xDEADBEEF
 
 
 @patch("esphome.writer.CORE")

tests/unit_tests/test_yaml_util.py

@@ -701,6 +701,31 @@ def test_include_file_has_unresolved_expressions(
     assert include.has_unresolved_expressions() == expected
 
 
+def test_mapping_include_non_string_file_rejected(tmp_path: Path) -> None:
+    """The mapping !include form rejects a non-string 'file' with a clear error."""
+    entry = tmp_path / "entry.yaml"
+    entry.write_text("wifi: !include\n  file: [not, a, string]\n")
+    with pytest.raises(EsphomeError, match="Include 'file' must be a string"):
+        yaml_util.load_yaml(entry)
+
+
+def test_include_file_templated_filename_stays_raw_string(tmp_path: Path) -> None:
+    """A templated filename keeps its verbatim text (issue #18545)."""
+    parent = tmp_path / "main.yaml"
+    expr = '${ "bluetooth/proxy.yaml" if enable_bluetooth_proxy else "../empty.yaml" }'
+    include = yaml_util.IncludeFile(parent, expr, None, lambda _: {})
+    assert include.file == expr
+    assert include.has_unresolved_expressions()
+    assert repr(include) == f"IncludeFile({expr})"
+
+
+def test_represent_include_file_templated() -> None:
+    """Dumping a templated IncludeFile emits the raw expression unchanged."""
+    expr = '${ "a/b.yaml" if flag else "../c.yaml" }'
+    include = yaml_util.IncludeFile(Path("/fake/main.yaml"), expr, None, lambda _: {})
+    assert yaml_util.dump({"key": include}) == f"key: !include '{expr}'\n"
+
+
 def test_include_in_list_context() -> None:
     """!include of a file returning a list is handled correctly,
     including when that list itself contains a nested IncludeFile."""
@@ -1051,7 +1076,7 @@ class _StubInclude:
     ) -> None:
         # Default parent lives in a nonexistent directory so unresolved
         # stubs never glob real files during candidate expansion.
-        self.file = Path(file)
+        self.file = file
         self.parent_file = parent_file or Path("/nonexistent/parent.yaml")
         self._unresolved = unresolved
         self._load_result = load_result if load_result is not None else {}