From 387df1ca771232a421f8142a2f12fea2b61976f2 Mon Sep 17 00:00:00 2001 From: "J. Nick Koston" Date: Tue, 29 Sep 2026 21:12:51 +0200 Subject: [PATCH] [espidf] Require the partition table offset and complete cache entries --- esphome/espidf/bootloader.py | 22 +++++---- tests/unit_tests/test_espidf_bootloader.py | 54 ++++++++++++++++------ 2 files changed, 54 insertions(+), 22 deletions(-) diff --git a/esphome/espidf/bootloader.py b/esphome/espidf/bootloader.py index ae947ee7ca8..b1959b778a5 100644 --- a/esphome/espidf/bootloader.py +++ b/esphome/espidf/bootloader.py @@ -248,9 +248,16 @@ def _work_dir(prefix: str) -> Path: return Path(tempfile.mkdtemp(prefix=prefix, dir=root)) +def _entry_complete(entry: Path) -> bool: + return all((entry / name).is_file() for name in _OUTPUTS) + + 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_complete(entry): + # A partial entry (manual cleanup, killed publish) must not win. + _remove_dir(entry) stage = _work_dir(f".stage-{key}-") for name in _OUTPUTS: shutil.copy2(build_dir / name, stage / name) @@ -263,7 +270,7 @@ def _publish(build_dir: Path, key: str, payload: dict) -> Path: except OSError: # Another process published the same key first. _remove_dir(stage) - if not (entry / "bootloader.bin").is_file(): + if not _entry_complete(entry): raise return entry @@ -282,8 +289,7 @@ def _install_into_build(entry: Path) -> Path: rmtree(dest) dest.mkdir(parents=True, exist_ok=True) for name in _OUTPUTS: - if (src := entry / name).is_file(): - shutil.copy2(src, dest / name) + shutil.copy2(entry / name, dest / name) return dest @@ -294,9 +300,8 @@ def _bootloader_offset(app_config: dict | None) -> int | None: return int(app_config["BOOTLOADER_OFFSET_IN_FLASH"]) -def _check_bootloader_size(bin_path: Path, offset: int, app_config: dict) -> None: +def _check_bootloader_size(bin_path: Path, offset: int, table_offset: int) -> None: """The check_sizes.py bootloader check, without spawning a process.""" - table_offset = int(app_config.get("PARTITION_TABLE_OFFSET", 0x8000)) size = bin_path.stat().st_size free = table_offset - offset - size if free < 0: @@ -330,8 +335,9 @@ def ensure_cached_bootloader(verbose: bool = False) -> bool: """ app_config = toolchain._load_sdkconfig_json(toolchain._build_dir()) offset = _bootloader_offset(app_config) + table_offset = (app_config or {}).get("PARTITION_TABLE_OFFSET") compiler = toolchain._resolved_c_compiler() - if offset is None or compiler is None: + if offset is None or table_offset is None or compiler is None: _LOGGER.debug("Bootloader cache unusable: app configure outputs missing") return False try: @@ -342,7 +348,7 @@ def ensure_cached_bootloader(verbose: bool = False) -> bool: if dest is None: return False # Outside the fail-safe net: the in-tree build would overflow the same way. - _check_bootloader_size(dest / "bootloader.bin", offset, app_config) + _check_bootloader_size(dest / "bootloader.bin", offset, int(table_offset)) return True @@ -356,7 +362,7 @@ def _install_cached(app_config: dict, compiler: str, verbose: bool) -> Path | No entry = _cache_root() / _compute_key( _key_payload(names, app_config, compiler, stamp) ) - if (entry / "bootloader.bin").is_file(): + if _entry_complete(entry): return _install_into_build(entry) tmp = _work_dir(".build-") probe = _work_dir(".build-") diff --git a/tests/unit_tests/test_espidf_bootloader.py b/tests/unit_tests/test_espidf_bootloader.py index e50746de6ad..b8839fa371c 100644 --- a/tests/unit_tests/test_espidf_bootloader.py +++ b/tests/unit_tests/test_espidf_bootloader.py @@ -379,14 +379,16 @@ def test_publish_lost_race_reuses_winner(tmp_path: Path) -> None: _make_built_tree(build) entry = root / ("k" * 16) - def rename_raises(src: Path, dst: Path) -> None: - entry.mkdir(parents=True) - (entry / "bootloader.bin").write_bytes(b"winner") - raise OSError("directory not empty") + # The winner finished first; the pre-clean must not remove it. + entry.mkdir(parents=True) + for name in bootloader._OUTPUTS: + (entry / name).write_bytes(b"winner") with ( patch.object(bootloader, "_cache_root", return_value=root), - patch.object(bootloader, "_rename_with_retry", rename_raises), + patch.object( + bootloader, "_rename_with_retry", side_effect=OSError("not empty") + ), ): assert bootloader._publish(build, "k" * 16, {}) == entry assert (entry / "bootloader.bin").read_bytes() == b"winner" @@ -411,18 +413,17 @@ def test_publish_rename_failure_without_winner_raises(tmp_path: Path) -> None: def test_install_into_build_replaces_stale_subbuild(tmp_path: Path) -> None: entry = tmp_path / "entry" entry.mkdir() - # No .map: an optional output missing from the entry is skipped. - (entry / "bootloader.bin").write_bytes(b"bin") - (entry / "bootloader.elf").write_bytes(b"elf") + for name in bootloader._OUTPUTS: + (entry / name).write_bytes(name.encode()) dest = CORE.relative_build_path("build", "bootloader") dest.mkdir(parents=True) (dest / "CMakeCache.txt").write_text("stale in-tree sub-build") (dest / "old.obj").write_bytes(b"x") assert bootloader._install_into_build(entry) == dest - assert (dest / "bootloader.bin").read_bytes() == b"bin" + for name in bootloader._OUTPUTS: + assert (dest / name).read_bytes() == name.encode() assert not (dest / "CMakeCache.txt").exists() assert not (dest / "old.obj").exists() - assert not (dest / "bootloader.map").exists() # -------------------------------------------------------------- size check @@ -433,18 +434,16 @@ def test_size_check_passes_and_logs( ) -> None: bin_path = tmp_path / "bootloader.bin" bin_path.write_bytes(b"\xe9" * 0x6620) - config = {"PARTITION_TABLE_OFFSET": 0x8000} with caplog.at_level("INFO"): - bootloader._check_bootloader_size(bin_path, 0x1000, config) + bootloader._check_bootloader_size(bin_path, 0x1000, 0x8000) assert "0x9e0 bytes (9%) free" in caplog.text def test_size_check_overflow_raises(tmp_path: Path) -> None: bin_path = tmp_path / "bootloader.bin" bin_path.write_bytes(b"\xe9" * 0x7100) - config = {"PARTITION_TABLE_OFFSET": 0x8000} with pytest.raises(EsphomeError, match="overflows by 0x100 bytes"): - bootloader._check_bootloader_size(bin_path, 0x1000, config) + bootloader._check_bootloader_size(bin_path, 0x1000, 0x8000) # ------------------------------------------------------------ orchestration @@ -539,6 +538,33 @@ def test_ensure_soft_fails_when_build_yields_no_config(tmp_path: Path) -> None: assert bootloader.ensure_cached_bootloader() is False +def test_ensure_rebuilds_an_incomplete_entry(tmp_path: Path) -> None: + """An entry that lost its elf or map is a miss and gets replaced.""" + with ( + _orchestration_env(tmp_path) as root, + patch.object( + bootloader, "_build_standalone", side_effect=_make_built_tree_rc + ) as mock_build, + ): + partial = root / "deadbeefdeadbeef" + partial.mkdir(parents=True) + (partial / "bootloader.bin").write_bytes(b"old") + (root / "config_names.json").write_text('["A"]') + 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: + assert (root / "deadbeefdeadbeef" / name).read_bytes() == name.encode() + + +def test_ensure_soft_fails_without_a_partition_table_offset(tmp_path: Path) -> None: + with _orchestration_env(tmp_path): + build = CORE.relative_build_path("build") + (build / "config" / "sdkconfig.json").write_text( + '{"BOOTLOADER_OFFSET_IN_FLASH": 4096}' + ) + assert bootloader.ensure_cached_bootloader() is False + + def test_ensure_rejects_an_unreproducible_bootloader( tmp_path: Path, caplog: pytest.LogCaptureFixture ) -> None: