[espidf] Require the partition table offset and complete cache entries

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-")

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: