diff --git a/esphome/build_gen/espidf.py b/esphome/build_gen/espidf.py index 7d8f5f0b270..41634886eda 100644 --- a/esphome/build_gen/espidf.py +++ b/esphome/build_gen/espidf.py @@ -291,6 +291,13 @@ include($ENV{{IDF_PATH}}/tools/cmake/project.cmake) project({CORE.name}) +# Record the bootloader customization channels the cache cannot replay; +# esphome.espidf.bootloader bypasses the cache when any is non-empty. +idf_build_get_property(esphome_bl_dirs BOOTLOADER_EXTRA_COMPONENT_DIRS) +idf_build_get_property(esphome_bl_args EXTRA_CMAKE_ARGS) +file(WRITE ${{CMAKE_BINARY_DIR}}/esphome_bootloader_hooks.txt + "${{esphome_bl_dirs}}${{BOOTLOADER_IGNORE_EXTRA_COMPONENT}}${{esphome_bl_args}}") + # Emit per-memory-type JSON size data for ESPHome to read post-build. # json2 stays small; raw dumps every symbol (~2s on a large map) and # this command runs inside the link edge, blocking everything downstream. diff --git a/esphome/espidf/__init__.py b/esphome/espidf/__init__.py index 77cad45111f..1e9f4e225d0 100644 --- a/esphome/espidf/__init__.py +++ b/esphome/espidf/__init__.py @@ -12,6 +12,10 @@ from typing import TYPE_CHECKING if TYPE_CHECKING: from pathlib import Path +# Written by the generated CMakeLists after project(); holds the bootloader +# customization channels the cache cannot replay (see espidf.bootloader). +BOOTLOADER_HOOKS_FILE = "esphome_bootloader_hooks.txt" + def variant_to_idf_target(variant: str) -> str: """Map an esp32 variant name (e.g. "ESP32S3") to its ESP-IDF target name.""" diff --git a/esphome/espidf/bootloader.py b/esphome/espidf/bootloader.py index 09507f2f2ae..762ffcb402d 100644 --- a/esphome/espidf/bootloader.py +++ b/esphome/espidf/bootloader.py @@ -11,7 +11,6 @@ from __future__ import annotations from collections.abc import Iterable import contextlib -from dataclasses import dataclass import hashlib import json import logging @@ -25,15 +24,9 @@ import time from esphome.build_helpers.ccache import parse_enable_env from esphome.build_helpers.tool_runner import run_build_tool from esphome.core import CORE, EsphomeError -from esphome.espidf import framework, parse_sdkconfig, toolchain +from esphome.espidf import BOOTLOADER_HOOKS_FILE, framework, parse_sdkconfig, toolchain from esphome.framework_helpers import _rename_with_retry -from esphome.helpers import ( - copy_file_if_changed, - file_compare, - read_json_file, - rmtree, - write_file, -) +from esphome.helpers import copy_file_if_changed, file_compare, rmtree, write_file _LOGGER = logging.getLogger(__name__) @@ -41,9 +34,6 @@ BOOTLOADER_CACHE_ENV = "ESPHOME_BOOTLOADER_CACHE" # Bump when the build or storage recipe changes; old entries then miss. _CACHE_SCHEMA = "1" _SECURE_OPTION = re.compile(r"CONFIG_(SECURE_|FLASH_ENCRYPTION)") -_BOOTLOADER_PROPERTY = re.compile( - r"BOOTLOADER_(EXTRA_COMPONENT_DIRS|IGNORE_EXTRA_COMPONENT)" -) _DISABLED_VALUES = frozenset(("n", "0", "")) # Marks a config name the app's sdkconfig does not define. _ABSENT = "\x00absent" @@ -79,7 +69,9 @@ def _compute_enabled() -> bool: if CORE.relative_build_path("bootloader_components").exists(): # Project-local bootloader overrides are inputs the key can't see. return False - if _managed_bootloader_hook() is not None: + if _recorded_bootloader_hooks(): + # The last configure saw a component customizing the bootloader. + _LOGGER.info("Bootloader cache off: build customizes the bootloader") return False if not os.access(tools := framework.get_idf_tools_path(), os.W_OK): # A read-only shared prefix would fail the cache on every build. @@ -100,19 +92,19 @@ def _compute_enabled() -> bool: return False -def _managed_bootloader_hook() -> str | None: - """The managed component wiring the bootloader build, or None. +def _recorded_bootloader_hooks() -> str | None: + """The bootloader customization channels the last configure recorded. - Checked again after configure: a cold tree has no managed_components - yet when the enable check first runs. + The generated CMakeLists dumps them after project(), so every source + counts: managed and local components, main, and EXTRA_CMAKE_ARGS. + None means no configure has recorded them yet; rechecked after + configure, since the pre-configure read can be stale or missing. """ - managed = CORE.relative_build_path("managed_components") - hooks = managed.glob("*/project_include.cmake") if managed.is_dir() else () - for hook in hooks: - if _BOOTLOADER_PROPERTY.search(hook.read_text(encoding="utf-8")): - _LOGGER.info("Bootloader cache off: %s customizes it", hook.parent.name) - return hook.parent.name - return None + path = toolchain._build_dir() / BOOTLOADER_HOOKS_FILE + try: + return path.read_text(encoding="utf-8").strip() + except FileNotFoundError: + return None def _snapshot_blocks_cache(sdkconfig_path: Path) -> bool: @@ -148,69 +140,50 @@ def _cache_root() -> Path: ) -def _load_config_names() -> list[str] | None: - """The known union of config option names the bootloader consumes.""" - path = _cache_root() / "config_names.json" - names = read_json_file(path) - if isinstance(names, list) and names and all(isinstance(n, str) for n in names): - return names - if names is not None: - # The next merge rewrites the full union; this only costs a miss. - _LOGGER.info("Ignoring corrupt %s", path) - return None +def _load_config_names() -> list[str]: + """The known union of config option names the bootloader consumes. + + Each harvest is an immutable file named by its digest; the union over + them is monotonic without any cross-process locking. + """ + union: set[str] = set() + for path in _cache_root().glob("config_names/*.json"): + try: + names = json.loads(path.read_text(encoding="utf-8")) + except (OSError, ValueError) as err: + _LOGGER.info("Ignoring unreadable %s: %s", path, err) + continue + if isinstance(names, list) and all(isinstance(n, str) for n in names): + union.update(names) + else: + _LOGGER.info("Ignoring corrupt %s", path) + return sorted(union) def _merge_config_names(names: Iterable[str]) -> list[str]: - """Union-merge harvested names; a corrupt file is replaced outright. + """Record this build's harvested names and return the resulting union. - write_file renames into place, and re-merging until the file covers our - names keeps the union monotonic when two builds publish at once. + A set that was already recorded is not rewritten; concurrent publishes + can only add files, never lose another's discovered names. """ - path = _cache_root() / "config_names.json" - merged = sorted(names) - for _ in range(3): - merged = sorted(set(_load_config_names() or ()) | set(merged)) + merged = sorted(set(names)) + path = _cache_root() / "config_names" / f"{_compute_key(merged)}.json" + if not path.is_file(): write_file(path, json.dumps(merged, separators=(",", ":"))) - if set(_load_config_names() or ()) >= set(merged): - break - return merged - - -@dataclass(frozen=True, kw_only=True) -class _SubprojectContract: - """How the pinned IDF's bootloader ExternalProject configures the subproject.""" - - self_append_bootloader_dir: bool # EXTRA_COMPONENT_DIRS gets the bootloader dir - forward_idf_build_v2: bool # IDF_BUILD_V2 forwarded (empty: never opted in) - - -_SUBPROJECT_5 = _SubprojectContract( - self_append_bootloader_dir=True, - forward_idf_build_v2=False, -) -_SUBPROJECT_6 = _SubprojectContract( - self_append_bootloader_dir=False, - forward_idf_build_v2=True, -) - - -def _subproject() -> _SubprojectContract: - from esphome.components.esp32 import idf_version - import esphome.config_validation as cv - - return _SUBPROJECT_6 if idf_version() >= cv.Version(6, 0, 0) else _SUBPROJECT_5 + # The union guards against our own file being unreadable right back. + return sorted(set(merged) | set(_load_config_names())) def _subproject_cmake_args( cmake: str, python: str, sdkconfig: str, idf_path: str, project_dir: str ) -> list[str]: """The configure argv, mirroring IDF's bootloader ExternalProject args.""" - contract = _subproject() - extra_dirs = ( - f"{idf_path}/components/bootloader" - if contract.self_append_bootloader_dir - else "" - ) + from esphome.components.esp32 import idf_version + import esphome.config_validation as cv + + # 5.x self-appends the bootloader dir to EXTRA_COMPONENT_DIRS; 6.x + # instead forwards IDF_BUILD_V2 (empty: the subproject never opts in). + idf6 = idf_version() >= cv.Version(6, 0, 0) args = [ cmake, "-G", @@ -220,9 +193,9 @@ def _subproject_cmake_args( f"-DIDF_TARGET={toolchain._idf_target()}", "-DPYTHON_DEPS_CHECKED=1", f"-DPYTHON={python}", - f"-DEXTRA_COMPONENT_DIRS={extra_dirs}", + f"-DEXTRA_COMPONENT_DIRS={'' if idf6 else f'{idf_path}/components/bootloader'}", ] - if contract.forward_idf_build_v2: + if idf6: args.append("-DIDF_BUILD_V2=") args += [ f"-DPROJECT_SOURCE_DIR={project_dir}", @@ -244,13 +217,13 @@ def _key_payload(names: list[str], app_config: dict, compiler: str, stamp: str) "args": _subproject_cmake_args( "cmake", "@PYTHON@", "@SDKCONFIG@", "@IDF@", "@PROJECT@" ), - "config": {n: app_config.get(n, _ABSENT) for n in sorted(names)}, + "config": {n: app_config.get(n, _ABSENT) for n in names}, "env": {n: os.environ.get(n, _ABSENT) for n in _FLAG_ENV_VARS}, } -def _compute_key(payload: dict) -> str: - """The cache entry name for this build's bootloader inputs.""" +def _compute_key(payload: dict | list) -> str: + """The digest naming a cache entry or a recorded config name set.""" text = json.dumps(payload, sort_keys=True, separators=(",", ":")) return hashlib.sha256(text.encode()).hexdigest()[:16] @@ -294,10 +267,17 @@ def _bin_sha256(path: Path) -> str: def _entry_intact(entry: Path) -> bool: - """Complete with the published bin bytes; anything else is a miss.""" + """Complete with the published bin bytes; anything else is a miss. + + False only for provable damage; a transient read error raises so a + valid shared entry is never mistaken for a corrupt one and deleted. + """ if not all((entry / name).is_file() for name in _OUTPUTS): return False - meta = read_json_file(entry / "meta.json") + try: + meta = json.loads((entry / "meta.json").read_text(encoding="utf-8")) + except (FileNotFoundError, ValueError): + return False expected = meta.get("bootloader_bin_sha256") if isinstance(meta, dict) else None return expected == _bin_sha256(entry / "bootloader.bin") @@ -305,8 +285,9 @@ def _entry_intact(entry: Path) -> bool: def _publish(build_dir: Path, key: str, payload: dict) -> Path: """Move the distilled outputs into the cache; losing a race is fine.""" entry = _cache_root() / key - if not _entry_intact(entry): + if entry.is_dir() and not _entry_intact(entry): # A partial or corrupted entry (manual cleanup) must not win. + _LOGGER.debug("Discarding damaged entry %s", entry) _remove_dir(entry) stage = _work_dir(f".stage-{key}-") try: @@ -347,6 +328,10 @@ def _install_into_build(entry: Path) -> Path: if (dest / "CMakeCache.txt").is_file(): # A leftover in-tree sub-build; its ninja state must not linger. rmtree(dest) + if (stamps := dest.parent / "bootloader-prefix").is_dir(): + # Stale ExternalProject stamps would make a later flip back to + # in-tree skip the sub-configure and fail on the missing cache. + rmtree(stamps) for name in _OUTPUTS: copy_file_if_changed(entry / name, dest / name) return dest @@ -398,8 +383,10 @@ def ensure_cached_bootloader(verbose: bool = False) -> bool: bootloader raises instead, since in-tree would fail the same way. """ try: - if _managed_bootloader_hook() is not None: - # Configure just downloaded it; the pre-configure scan ran too early. + if _recorded_bootloader_hooks() != "": + # Non-empty: configure just recorded a bootloader customization. + # None: this configure did not write the record at all. + _LOGGER.info("Bootloader customized by the build; building in-tree") return False app_config = toolchain._load_sdkconfig_json(toolchain._build_dir()) offset = _bootloader_offset(app_config) @@ -414,7 +401,7 @@ def ensure_cached_bootloader(verbose: bool = False) -> bool: ) return False dest = _install_cached(app_config, compiler, verbose) - except (OSError, EsphomeError) as err: + except (OSError, ValueError, EsphomeError) as err: _LOGGER.warning("Bootloader cache failed: %s", err) return False if dest is None: @@ -422,7 +409,7 @@ def ensure_cached_bootloader(verbose: bool = False) -> bool: try: # The size overflow raises out: the in-tree build would fail the same. _check_bootloader_size(dest / "bootloader.bin", offset, int(table_offset)) - except OSError as err: + except (OSError, ValueError) as err: _LOGGER.warning("Bootloader cache failed: %s", err) return False return True @@ -434,7 +421,7 @@ def _install_cached(app_config: dict, compiler: str, verbose: bool) -> Path | No if not (stamp := framework.read_idf_version_txt(toolchain._get_idf_path())): _LOGGER.info("Framework version.txt missing; cannot key the bootloader") return None - if (names := _load_config_names()) is not None: + if names := _load_config_names(): entry = _cache_root() / _compute_key( _key_payload(names, app_config, compiler, stamp) ) diff --git a/script/check_idf_py_equivalence.py b/script/check_idf_py_equivalence.py index 218df842e49..3c0fcdeb7b5 100755 --- a/script/check_idf_py_equivalence.py +++ b/script/check_idf_py_equivalence.py @@ -184,10 +184,10 @@ def _bootloader_parity(build_path: Path) -> list[str]: # pylint: disable=protected-access from esphome.espidf import bootloader, toolchain - bin_path = build_path / "build" / "bootloader" / "bootloader.bin" if not bootloader.tree_uses_cached_bootloader(build_path / "build"): print("note: tree not in cached mode; bootloader parity not exercised") return [] + bin_path = build_path / "build" / "bootloader" / "bootloader.bin" cached = bin_path.read_bytes() bin_path.unlink() # the rebuild must produce it, never compare to itself os.environ[bootloader.BOOTLOADER_CACHE_ENV] = "0" diff --git a/tests/unit_tests/build_gen/test_espidf.py b/tests/unit_tests/build_gen/test_espidf.py index 28f3dc3d873..b20ae06dd8d 100644 --- a/tests/unit_tests/build_gen/test_espidf.py +++ b/tests/unit_tests/build_gen/test_espidf.py @@ -175,6 +175,20 @@ def test_get_project_cmakelists_size_command_uses_json2() -> None: assert "--format=json2" in content +def test_get_project_cmakelists_records_bootloader_hooks() -> None: + """The post-project() dump feeds espidf.bootloader's bypass check.""" + from esphome.espidf import BOOTLOADER_HOOKS_FILE + + content = _render() + assert f"file(WRITE ${{CMAKE_BINARY_DIR}}/{BOOTLOADER_HOOKS_FILE}" in content + for source in ( + "esphome_bl_dirs BOOTLOADER_EXTRA_COMPONENT_DIRS", + "${BOOTLOADER_IGNORE_EXTRA_COMPONENT}", + "esphome_bl_args EXTRA_CMAKE_ARGS", + ): + assert source in content + + def test_get_project_cmakelists_uses_supplied_builtin_components() -> None: """A cached list replaces project_description.json and is still filtered by EXCLUDE_COMPONENTS.""" diff --git a/tests/unit_tests/test_espidf_bootloader.py b/tests/unit_tests/test_espidf_bootloader.py index 08f5763d1a5..2efff1215d1 100644 --- a/tests/unit_tests/test_espidf_bootloader.py +++ b/tests/unit_tests/test_espidf_bootloader.py @@ -33,6 +33,14 @@ def _fresh_cache(setup_core: Path, monkeypatch: pytest.MonkeyPatch) -> None: CORE.data.pop(toolchain.DOMAIN, None) +def _write_hooks_record(text: str) -> Path: + """The hooks dump the generated CMakeLists writes on every configure.""" + path = toolchain._build_dir() / "esphome_bootloader_hooks.txt" + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(text) + return path + + def _write_snapshot(text: str = "") -> Path: path = CORE.relative_build_path("sdkconfig.dev.esphomeinternal") path.parent.mkdir(parents=True, exist_ok=True) @@ -117,22 +125,16 @@ def test_enabled_project_bootloader_components_disable(tmp_path: Path) -> None: assert bootloader._compute_enabled() is False -def test_enabled_managed_component_bootloader_hook_disables(tmp_path: Path) -> None: - """A managed component wiring the bootloader build is an unseen input.""" +def test_enabled_recorded_bootloader_hooks_disable(tmp_path: Path) -> None: + """A customization recorded by the last configure is an unseen input.""" _write_snapshot("CONFIG_APP_REPRODUCIBLE_BUILD=y\n") - hook = CORE.relative_build_path( - "managed_components", "vendor__boot", "project_include.cmake" - ) - hook.parent.mkdir(parents=True) - hook.write_text( - 'idf_build_set_property(BOOTLOADER_EXTRA_COMPONENT_DIRS "x" APPEND)' - ) + _write_hooks_record("/some/component/dir") with ( _tools_prefix(tmp_path), patch.object(build_gen, "idf_macro_matches", return_value=True), ): assert bootloader._compute_enabled() is False - hook.write_text("# nothing bootloader related") + _write_hooks_record("") assert bootloader._compute_enabled() is True @@ -292,45 +294,43 @@ def test_merge_config_names_unions_and_sorts(tmp_path: Path) -> None: assert bootloader._load_config_names() == ["A", "B", "C"] -def test_merge_config_names_remerges_after_a_racing_writer(tmp_path: Path) -> None: - """A name landed by a concurrent build must survive the merge.""" - # merge-read, verify-read (racer landed), merge-read, verify-read - loads = iter([None, ["OTHER"], ["OTHER"], ["A", "B", "OTHER"]]) - with ( - patch.object(bootloader, "_load_config_names", side_effect=loads), - patch.object(bootloader, "write_file") as mock_write, - ): - assert bootloader._merge_config_names(["B", "A"]) == ["A", "B", "OTHER"] - assert mock_write.call_count == 2 # rewrote once with the racer's name - - -def test_merge_config_names_gives_up_after_bounded_retries(tmp_path: Path) -> None: - """A pathological racer costs at most a miss, never a spin.""" - with ( - patch.object(bootloader, "_load_config_names", return_value=["X"]), - patch.object(bootloader, "write_file") as mock_write, - ): - assert bootloader._merge_config_names(["A"]) == ["A", "X"] - assert mock_write.call_count == 3 +def test_merge_config_names_skips_rewriting_a_recorded_set(tmp_path: Path) -> None: + """The same harvest recorded twice writes one immutable file.""" + root = tmp_path / "cache" + with patch.object(bootloader, "_cache_root", return_value=root): + bootloader._merge_config_names(["A"]) + with patch.object(bootloader, "write_file") as mock_write: + assert bootloader._merge_config_names(["A"]) == ["A"] + mock_write.assert_not_called() @pytest.mark.parametrize( - "content", - [None, "not json", "[]", '["a", 1]', '"str"'], - ids=["missing", "corrupt", "empty", "mixed-types", "not-a-list"], + ("content", "log"), + [ + (None, ""), + ("not json", "Ignoring unreadable"), + ("[]", ""), + ('["a", 1]', "Ignoring corrupt"), + ('"str"', "Ignoring corrupt"), + ], + ids=["missing", "unreadable", "empty", "mixed-types", "not-a-list"], ) def test_load_config_names_rejects_bad_files( - tmp_path: Path, content: str | None, caplog: pytest.LogCaptureFixture + tmp_path: Path, content: str | None, log: str, caplog: pytest.LogCaptureFixture ) -> None: if content is not None: - (tmp_path / "config_names.json").write_text(content) + (tmp_path / "config_names").mkdir() + (tmp_path / "config_names" / "x.json").write_text(content) with ( patch.object(bootloader, "_cache_root", return_value=tmp_path), caplog.at_level("INFO"), ): - assert bootloader._load_config_names() is None - # A present-but-unusable file says so; a missing one is normal. - assert ("Ignoring corrupt" in caplog.text) is (content not in (None, "not json")) + assert bootloader._load_config_names() == [] + # A present-but-unusable file says so; missing or empty is normal. + if log: + assert log in caplog.text + else: + assert "Ignoring" not in caplog.text # ------------------------------------------------------------- cmake args @@ -508,9 +508,7 @@ def test_publish_keeps_the_original_error_when_the_intact_check_breaks( (build / "bootloader.map").unlink() # the real failure with ( patch.object(bootloader, "_cache_root", return_value=root), - patch.object( - bootloader, "_entry_intact", side_effect=[False, PermissionError()] - ), + patch.object(bootloader, "_entry_intact", side_effect=PermissionError()), # FileNotFoundError can only be the staging error, not the check's. pytest.raises(FileNotFoundError), ): @@ -529,11 +527,15 @@ def test_install_into_build_replaces_stale_subbuild(tmp_path: Path) -> None: dest.mkdir(parents=True) (dest / "CMakeCache.txt").write_text("stale in-tree sub-build") (dest / "old.obj").write_bytes(b"x") + stamps = CORE.relative_build_path("build", "bootloader-prefix", "src") + stamps.mkdir(parents=True) assert bootloader._install_into_build(entry) == dest for name in bootloader._OUTPUTS: assert (dest / name).read_bytes() == name.encode() assert not (dest / "CMakeCache.txt").exists() assert not (dest / "old.obj").exists() + # Stale stamps would make a later in-tree flip skip the sub-configure. + assert not stamps.parent.exists() # -------------------------------------------------------------- size check @@ -559,26 +561,35 @@ def test_size_check_overflow_raises(tmp_path: Path) -> None: # ------------------------------------------------------------ orchestration -def test_ensure_rechecks_managed_hooks_after_configure(tmp_path: Path) -> None: - """A hook that configure just downloaded must stop the first build too.""" - hook = CORE.relative_build_path( - "managed_components", "vendor__boot", "project_include.cmake" - ) - hook.parent.mkdir(parents=True) - hook.write_text("idf_build_set_property(BOOTLOADER_IGNORE_EXTRA_COMPONENT x)") +def test_ensure_rechecks_recorded_hooks_after_configure(tmp_path: Path) -> None: + """A customization configure just recorded must stop the first build too.""" with ( _orchestration_env(tmp_path), patch.object(bootloader, "_build_standalone") as mock_build, ): + _write_hooks_record("vendor__boot") assert bootloader.ensure_cached_bootloader() is False mock_build.assert_not_called() -def test_ensure_fails_soft_when_hook_scan_errors(tmp_path: Path) -> None: - """An unreadable managed hook falls back instead of aborting the build.""" +def test_ensure_fails_soft_without_a_hooks_record(tmp_path: Path) -> None: + """A configure that wrote no record cannot prove the bootloader is stock.""" with ( _orchestration_env(tmp_path), - patch.object(bootloader, "_managed_bootloader_hook", side_effect=OSError("io")), + patch.object(bootloader, "_build_standalone") as mock_build, + ): + (toolchain._build_dir() / "esphome_bootloader_hooks.txt").unlink() + assert bootloader.ensure_cached_bootloader() is False + mock_build.assert_not_called() + + +def test_ensure_fails_soft_when_hooks_record_is_unreadable(tmp_path: Path) -> None: + """An unreadable hooks record falls back instead of aborting the build.""" + with ( + _orchestration_env(tmp_path), + patch.object( + bootloader, "_recorded_bootloader_hooks", side_effect=OSError("io") + ), ): assert bootloader.ensure_cached_bootloader() is False @@ -592,7 +603,7 @@ def test_ensure_fails_soft_when_size_check_cannot_stat( patch.object(bootloader, "_check_bootloader_size", side_effect=OSError("gone")), ): _make_entry(root / "deadbeefdeadbeef", b"\xe9" * 64) - (root / "config_names.json").write_text('["A"]') + _seed_names(root) assert bootloader.ensure_cached_bootloader() is False assert "Bootloader cache failed" in caplog.text @@ -601,11 +612,18 @@ def test_ensure_fails_soft_without_configure_outputs( caplog: pytest.LogCaptureFixture, ) -> None: """No app config or compiler id means fall back, not crash.""" + _write_hooks_record("") with caplog.at_level("INFO"): assert bootloader.ensure_cached_bootloader() is False assert "cache unusable" in caplog.text # says which input is missing +def _seed_names(root: Path) -> None: + """Record a harvested name set the way a previous publish would have.""" + (root / "config_names").mkdir(parents=True, exist_ok=True) + (root / "config_names" / "seed.json").write_text('["A"]') + + @contextmanager def _orchestration_env(tmp_path: Path) -> Iterator[Path]: """A configured app tree plus the patches every ensure_* test shares; @@ -613,6 +631,7 @@ def _orchestration_env(tmp_path: Path) -> Iterator[Path]: root = tmp_path / "cache-root" build = CORE.relative_build_path("build") build.mkdir(parents=True, exist_ok=True) + (build / "esphome_bootloader_hooks.txt").write_text("") (build / "config").mkdir(exist_ok=True) (build / "config" / "sdkconfig.json").write_text( json.dumps( @@ -639,7 +658,7 @@ def test_ensure_cache_hit_installs_without_building(tmp_path: Path) -> None: patch.object(bootloader, "_build_standalone") as mock_build, ): _make_entry(root / "deadbeefdeadbeef", b"\xe9" * 64) - (root / "config_names.json").write_text('["A"]') + _seed_names(root) assert bootloader.ensure_cached_bootloader() is True mock_build.assert_not_called() installed = CORE.relative_build_path("build", "bootloader", "bootloader.bin") @@ -656,12 +675,12 @@ def test_ensure_cache_miss_builds_and_publishes(tmp_path: Path) -> None: ) as mock_build, ): root.mkdir(parents=True) - (root / "config_names.json").write_text('["A"]') + _seed_names(root) assert bootloader.ensure_cached_bootloader() is True + assert "A" in bootloader._load_config_names() # Both probe builds ran; entry published, names harvested, work dirs gone. assert len(mock_build.call_args_list) == 2 assert (root / "deadbeefdeadbeef" / "bootloader.bin").is_file() - assert "A" in json.loads((root / "config_names.json").read_text()) assert not list(root.glob(".build-*")) installed = CORE.relative_build_path("build", "bootloader", "bootloader.bin") assert installed.is_file() @@ -692,6 +711,37 @@ def test_ensure_soft_fails_when_build_yields_no_config( assert "produced no" in caplog.text # names the expected path +def test_entry_intact_false_on_corrupt_meta(tmp_path: Path) -> None: + """Provably damaged metadata is a miss and may be discarded.""" + entry = tmp_path / "k" + _make_entry(entry) + (entry / "meta.json").write_text("not json") + assert bootloader._entry_intact(entry) is False + + +def test_entry_intact_raises_on_an_unreadable_meta(tmp_path: Path) -> None: + """Transient read trouble must not look like corruption.""" + entry = tmp_path / "k" + _make_entry(entry) + (entry / "meta.json").unlink() + (entry / "meta.json").mkdir() # read_text raises OSError, not ValueError + with pytest.raises(OSError): + bootloader._entry_intact(entry) + + +def test_publish_keeps_an_entry_it_cannot_verify(tmp_path: Path) -> None: + """A transient read error must never delete a valid shared entry.""" + with patch.object(bootloader, "_cache_root", return_value=tmp_path): + entry = tmp_path / "k" + _make_entry(entry) + with ( + patch.object(bootloader, "_entry_intact", side_effect=OSError("locked")), + pytest.raises(OSError), + ): + bootloader._publish(tmp_path / "build", "k", {}) + assert (entry / "bootloader.bin").is_file() + + @pytest.mark.parametrize("damage", ["missing-elf", "corrupt-bin"]) def test_ensure_rebuilds_a_damaged_entry(tmp_path: Path, damage: str) -> None: """A partial or corrupted entry is a miss and gets replaced.""" @@ -707,7 +757,7 @@ def test_ensure_rebuilds_a_damaged_entry(tmp_path: Path, damage: str) -> None: (entry / "bootloader.elf").unlink() else: (entry / "bootloader.bin").write_bytes(b"truncated") - (root / "config_names.json").write_text('["A"]') + _seed_names(root) assert bootloader.ensure_cached_bootloader() is True assert mock_build.call_count == 2 # built and probed, not served as a hit for name in bootloader._OUTPUTS: