[core] Harden the shared library converter and download registry archives in parallel

tests/unit_tests/test_espidf_component.py

@@ -1,12 +1,12 @@
 import glob
 import hashlib
 import json
-import os
 from pathlib import Path
 from unittest.mock import MagicMock
 
 import pytest
 
+from esphome.components import esp32 as esp32_module
 from esphome.const import (
     KEY_CORE,
     KEY_TARGET_FRAMEWORK,
@@ -16,6 +16,7 @@ from esphome.const import (
 )
 from esphome.core import CORE, Library
 from esphome.espidf.component import (
+    _emit_idf_component,
     generate_cmakelists_txt,
     generate_idf_component_yml,
     generate_idf_components,
@@ -26,11 +27,11 @@ from esphome.platformio.library import (
     GitSource,
     URLSource,
     _node_key,
-    _normalize_dependencies,
-    _parse_library_json,
-    _parse_library_properties,
     _resolve_registry_version,
     collect_filtered_files,
+    normalize_dependencies,
+    parse_library_json,
+    parse_library_properties,
     split_list_by_condition,
 )
 
@@ -369,133 +370,11 @@ def test_generate_idf_component_yml_missing_path_raises(tmp_component):
         generate_idf_component_yml(tmp_component)
 
 
-def test_extra_script_captures_libpath_libs_and_defines(tmp_path):
-    from esphome.espidf.extra_script import captured_as_build_flags, run_extra_script
-
-    (tmp_path / "src" / "esp32").mkdir(parents=True)
-    script = tmp_path / "extra_script.py"
-    script.write_text(
-        "Import('env')\n"
-        "mcu = env.get('BOARD_MCU')\n"
-        "env.Append(\n"
-        "    LIBPATH=[join('src', mcu)],\n"
-        "    LIBS=['algobsec'],\n"
-        "    CPPDEFINES=['FOO', ('BAR', '1')],\n"
-        "    LINKFLAGS=['-Wl,--gc-sections'],\n"
-        ")\n"
-    )
-    # The script uses bare ``join`` (PIO's extra-scripts run inside SCons
-    # where this is in scope). Inject it via the script header so the
-    # shim's exec namespace can resolve it.
-    script.write_text("from os.path import join\n" + script.read_text())
-
-    result = run_extra_script(script, library_dir=tmp_path, idf_target="esp32")
-
-    assert result.libpath == [str(Path("src") / "esp32")]
-    assert result.libs == ["algobsec"]
-    assert ("BAR", "1") in result.cppdefines
-    assert "FOO" in result.cppdefines
-    assert result.linkflags == ["-Wl,--gc-sections"]
-
-    flags = captured_as_build_flags(result, library_dir=tmp_path)
-    sep = os.sep
-    assert f"-Lsrc{sep}esp32" in flags
-    assert "-lalgobsec" in flags
-    assert "-DFOO" in flags
-    assert "-DBAR=1" in flags
-    assert "-Wl,--gc-sections" in flags
-
-
-def test_extra_script_libpath_relative_resolves_against_library_dir(
-    tmp_path, monkeypatch
-):
-    """Relative LIBPATH entries must resolve against ``library_dir``, not the
-    caller's CWD (the shim restores CWD before ``captured_as_build_flags``
-    runs)."""
-    from esphome.espidf.extra_script import ExtraScriptResult, captured_as_build_flags
-
-    (tmp_path / "lib" / "esp32").mkdir(parents=True)
-    elsewhere = tmp_path.parent / "not_the_library_dir"
-    elsewhere.mkdir(exist_ok=True)
-    monkeypatch.chdir(elsewhere)
-
-    result = ExtraScriptResult(libpath=["lib/esp32"])
-    flags = captured_as_build_flags(result, library_dir=tmp_path)
-
-    sep = os.sep
-    assert flags == [f"-Llib{sep}esp32"]
-
-
-def test_extra_script_libpath_absolute_outside_library_dir(tmp_path):
-    from esphome.espidf.extra_script import ExtraScriptResult, captured_as_build_flags
-
-    outside = tmp_path.parent / "system_lib"
-    outside.mkdir(exist_ok=True)
-    result = ExtraScriptResult(libpath=[str(outside)])
-
-    flags = captured_as_build_flags(result, library_dir=tmp_path)
-    assert flags == [f"-L{outside.resolve()}"]
-
-
-def test_extra_script_failure_returns_empty_result(tmp_path, caplog):
-    from esphome.espidf.extra_script import run_extra_script
-
-    script = tmp_path / "broken.py"
-    script.write_text("raise RuntimeError('boom')\n")
-
-    with caplog.at_level("WARNING"):
-        result = run_extra_script(script, library_dir=tmp_path, idf_target="esp32")
-
-    assert result.libpath == []
-    assert result.libs == []
-    assert "broken.py" in caplog.text
-
-
-def test_apply_extra_script_path_traversal_is_rejected(tmp_path):
-    from esphome.espidf.component import _apply_extra_script
-
-    library_dir = tmp_path / "lib"
-    library_dir.mkdir()
-    outside = tmp_path / "evil.py"
-    outside.write_text("env.Append(LIBS=['pwned'])\n")
-
-    c = IDFComponent("owner/name", "1.0", source=URLSource("http://dummy"))
-    c.path = library_dir
-    c.data = {"build": {"extraScript": "../evil.py"}}
-
-    _apply_extra_script(c)
-
-    # Nothing was folded into flags: the traversal was rejected before
-    # the script could run.
-    assert "flags" not in c.data["build"]
-
-
-def test_apply_extra_script_merges_into_existing_flags(tmp_path, monkeypatch):
-    from esphome.components import esp32 as esp32_module
-
-    monkeypatch.setattr(esp32_module, "get_esp32_variant", lambda: "ESP32")
-
-    from esphome.espidf.component import _apply_extra_script
-
-    (tmp_path / "src").mkdir()
-    script = tmp_path / "extra.py"
-    script.write_text("env.Append(LIBS=['algobsec'])\n")
-
-    c = IDFComponent("owner/name", "1.0", source=URLSource("http://dummy"))
-    c.path = tmp_path
-    c.data = {"build": {"extraScript": "extra.py", "flags": ["-DEXISTING"]}}
-
-    _apply_extra_script(c)
-
-    assert "-DEXISTING" in c.data["build"]["flags"]
-    assert "-lalgobsec" in c.data["build"]["flags"]
-
-
 def test_parse_library_json(tmp_path):
     f = tmp_path / "library.json"
     f.write_text(json.dumps({"name": "test"}))
 
-    result = _parse_library_json(f)
+    result = parse_library_json(f)
     assert result["name"] == "test"
 
 
@@ -510,7 +389,7 @@ empty=
 """
     )
 
-    result = _parse_library_properties(f)
+    result = parse_library_properties(f)
 
     assert result["name"] == "Test"
     assert result["version"] == "1.0"
@@ -680,22 +559,22 @@ def test_node_key_registry_bare_name():
 
 
 def test_normalize_dependencies_none():
-    assert _normalize_dependencies(None) == []
+    assert normalize_dependencies(None) == []
 
 
 def test_normalize_dependencies_list_form():
     deps = [{"name": "foo", "version": "1.0"}]
-    assert _normalize_dependencies(deps) == [{"name": "foo", "version": "1.0"}]
+    assert normalize_dependencies(deps) == [{"name": "foo", "version": "1.0"}]
 
 
 def test_normalize_dependencies_dict_form():
-    out = _normalize_dependencies({"nanopb/Nanopb": "^0.4.91", "BareName": "1.2.3"})
+    out = normalize_dependencies({"nanopb/Nanopb": "^0.4.91", "BareName": "1.2.3"})
     assert {"name": "Nanopb", "owner": "nanopb", "version": "^0.4.91"} in out
     assert {"name": "BareName", "owner": None, "version": "1.2.3"} in out
 
 
 def test_normalize_dependencies_dict_form_nested_spec():
-    out = _normalize_dependencies(
+    out = normalize_dependencies(
         {"nanopb/Nanopb": {"version": "^0.4.91", "platforms": "espidf"}}
     )
     assert out == [
@@ -1187,6 +1066,36 @@ def test_idf_component_download_passes_salt() -> None:
     c.download(force=True, salt="abcd1234", namespace="idf")
 
     source.download.assert_called_once_with(
-        "owner/name", force=True, salt="abcd1234", namespace="idf"
+        "owner/name", force=True, salt="abcd1234", namespace="idf", progress=None
     )
     assert c.path == Path("/converted/owner/name")
+
+
+def test_emit_idf_component_wires_esp32_target(tmp_path, monkeypatch):
+    """Emitting a component resolves the esp32 variant into the shared
+    extraScript helper."""
+
+    monkeypatch.setattr(esp32_module, "get_esp32_variant", lambda: "ESP32")
+    (tmp_path / "src").mkdir()
+    script = tmp_path / "extra.py"
+    script.write_text("env.Append(LIBS=[env.get('BOARD_MCU')])\n")
+    c = IDFComponent("owner/name", "1.0", source=URLSource("http://dummy"))
+    c.path = tmp_path
+    c.data = {"build": {"extraScript": "extra.py"}}
+    _emit_idf_component(c)
+    assert c.data["build"]["flags"] == ["-lesp32"]
+
+
+def test_build_flags_dangling_flag_does_not_cross_entries(
+    tmp_path, caplog: pytest.LogCaptureFixture
+) -> None:
+    """Each entry is lexed independently, as ParseFlags does: a dangling -I ending one
+    entry warns instead of absorbing the next entry's first token."""
+    (tmp_path / "src").mkdir()
+    c = IDFComponent("owner/name", "1.0", source=URLSource("http://dummy"))
+    c.path = tmp_path
+    c.data = {"build": {"flags": ["-Wall -I", "-DFOO=1"]}}
+    content = generate_cmakelists_txt(c)
+    assert "FOO=1" in content
+    assert "-I-DFOO" not in content
+    assert "Ignoring trailing '-I'" in caplog.text

tests/unit_tests/test_espidf_framework.py

@@ -2,6 +2,7 @@
 
 # pylint: disable=protected-access
 
+from concurrent.futures import ThreadPoolExecutor
 from contextlib import contextmanager
 import importlib.util
 import io
@@ -14,7 +15,7 @@ import subprocess
 import sys
 import tarfile
 from types import SimpleNamespace
-from unittest.mock import patch
+from unittest.mock import MagicMock, patch
 
 import pytest
 
@@ -887,6 +888,58 @@ _PREFETCH_JSON = json.dumps(
 )
 
 
+def test_prefetch_leaves_unverifiable_entries_to_the_installer(
+    tmp_path: Path, caplog: pytest.LogCaptureFixture
+) -> None:
+    """An entry missing sha256 or size must not download unverified; the
+    installer handles it and fails loudly on a bad archive."""
+    entries = json.loads(_PREFETCH_JSON)
+    del entries[0]["sha256"]
+    del entries[1]["size"]
+    entries.append(
+        {
+            "name": "gcc@14.2.0",
+            "url": "https://example.com/gcc.tar.gz",
+            "size": 67,
+            "sha256": "ef" * 32,
+            "dest": "gcc.tar.gz",
+        }
+    )
+    with (
+        patch(
+            "esphome.espidf.framework.run_command",
+            return_value=(True, json.dumps(entries), ""),
+        ),
+        patch("esphome.espidf.framework.download_with_resume") as download,
+        patch("esphome.espidf.framework.get_system_python_path", return_value="python"),
+        patch("esphome.espidf.framework.BatchDownloadProgress") as progress_cls,
+    ):
+        _prefetch_idf_tool_archives(tmp_path, "esp32", ["required"], None)
+    assert [call[0][0] for call in download.call_args_list] == [
+        "https://example.com/gcc.tar.gz"
+    ]
+    assert download.call_args[1]["sha256"] == "ef" * 32
+    progress_cls.assert_called_once_with("Downloading ESP-IDF tools", 67)
+    assert "cmake@3.30.2 has no sha256/size" in caplog.text
+    assert "ninja@1.12.1 has no sha256/size" in caplog.text
+
+
+def test_prefetch_all_entries_unverifiable_is_a_noop(tmp_path: Path) -> None:
+    entries = json.loads(_PREFETCH_JSON)
+    for entry in entries:
+        del entry["sha256"]
+    with (
+        patch(
+            "esphome.espidf.framework.run_command",
+            return_value=(True, json.dumps(entries), ""),
+        ),
+        patch("esphome.espidf.framework.download_with_resume") as download,
+        patch("esphome.espidf.framework.get_system_python_path", return_value="python"),
+    ):
+        _prefetch_idf_tool_archives(tmp_path, "esp32", ["required"], None)
+    download.assert_not_called()
+
+
 def test_prefetch_downloads_each_archive_with_resume(tmp_path: Path) -> None:
     with (
         patch(
@@ -895,16 +948,73 @@ def test_prefetch_downloads_each_archive_with_resume(tmp_path: Path) -> None:
         ),
         patch("esphome.espidf.framework.download_with_resume") as download,
         patch("esphome.espidf.framework.get_system_python_path", return_value="python"),
+        patch("esphome.espidf.framework.BatchDownloadProgress") as progress_cls,
     ):
+        # Materialize the lazy mock before threads race its first creation
+        tracker = progress_cls.return_value.tracker.return_value
         _prefetch_idf_tool_archives(tmp_path, "esp32", ["required"], None)
 
     dist = get_idf_tools_path() / "dist"
-    assert download.call_count == 2
-    assert download.call_args_list[0][0] == (
-        "https://example.com/cmake.tar.gz",
-        dist / "cmake-3.30.2.tar.gz",
-    )
-    assert download.call_args_list[0][1] == {"sha256": "ab" * 32, "size": 123}
+    # Archives download concurrently, so the call order is not fixed.
+    calls = {call[0]: call[1] for call in download.call_args_list}
+    assert set(calls) == {
+        ("https://example.com/cmake.tar.gz", dist / "cmake-3.30.2.tar.gz"),
+        ("https://example.com/ninja.zip", dist / "ninja.zip"),
+    }
+    kwargs = calls[("https://example.com/cmake.tar.gz", dist / "cmake-3.30.2.tar.gz")]
+    assert kwargs["sha256"] == "ab" * 32
+    assert kwargs["size"] == 123
+    # every archive reports into the one combined progress bar
+    progress_cls.assert_called_once_with("Downloading ESP-IDF tools", 123 + 45)
+    assert all(kw["progress"] is tracker for kw in calls.values())
+
+
+def test_prefetch_downloads_archives_concurrently(tmp_path: Path) -> None:
+    """More than one archive fans out over a bounded thread pool."""
+    entries = [
+        {
+            "name": f"tool{i}@1",
+            "url": f"https://example.com/tool{i}.tar.gz",
+            "size": 10,
+            "sha256": "ab" * 32,
+            "dest": f"tool{i}.tar.gz",
+        }
+        for i in range(6)
+    ]
+    with (
+        patch(
+            "esphome.espidf.framework.run_command",
+            return_value=(True, json.dumps(entries), ""),
+        ),
+        patch("esphome.espidf.framework.download_with_resume") as download,
+        patch("esphome.espidf.framework.get_system_python_path", return_value="python"),
+        patch(
+            "esphome.framework_helpers.ThreadPoolExecutor", wraps=ThreadPoolExecutor
+        ) as pool,
+    ):
+        _prefetch_idf_tool_archives(tmp_path, "esp32", ["required"], None)
+
+    pool.assert_called_once_with(max_workers=4)
+    assert download.call_count == 6
+
+
+def test_prefetch_single_archive_uses_one_worker(tmp_path: Path) -> None:
+    entries = json.loads(_PREFETCH_JSON)[:1]
+    with (
+        patch(
+            "esphome.espidf.framework.run_command",
+            return_value=(True, json.dumps(entries), ""),
+        ),
+        patch("esphome.espidf.framework.download_with_resume") as download,
+        patch("esphome.espidf.framework.get_system_python_path", return_value="python"),
+        patch(
+            "esphome.framework_helpers.ThreadPoolExecutor", wraps=ThreadPoolExecutor
+        ) as pool,
+    ):
+        _prefetch_idf_tool_archives(tmp_path, "esp32", ["required"], None)
+
+    pool.assert_called_once_with(max_workers=1)
+    assert download.call_count == 1
 
 
 def test_prefetch_skips_already_downloaded_archives(tmp_path: Path) -> None:
@@ -964,6 +1074,11 @@ def test_prefetch_one_failed_archive_does_not_stop_the_rest(
 ) -> None:
     """A single archive failing its download must not abort the prefetch of
     the remaining archives."""
+
+    def _fail_cmake_download(url: str, *args, **kwargs) -> None:
+        if "cmake" in url:
+            raise OSError("network down")
+
     with (
         patch(
             "esphome.espidf.framework.run_command",
@@ -971,7 +1086,7 @@ def test_prefetch_one_failed_archive_does_not_stop_the_rest(
         ),
         patch(
             "esphome.espidf.framework.download_with_resume",
-            side_effect=[OSError("network down"), None],
+            side_effect=_fail_cmake_download,
         ) as download,
         patch("esphome.espidf.framework.get_system_python_path", return_value="python"),
     ):
@@ -981,6 +1096,29 @@ def test_prefetch_one_failed_archive_does_not_stop_the_rest(
     assert "Could not prefetch cmake@3.30.2" in caplog.text
 
 
+def test_prefetch_finishes_progress_bar_and_cancels_queue(tmp_path: Path) -> None:
+    """The batch bar is closed out after the pool, and the pool is shut down
+    with cancel_futures so Ctrl-C does not drain every queued archive."""
+    with (
+        patch(
+            "esphome.espidf.framework.run_command",
+            return_value=(True, _PREFETCH_JSON, ""),
+        ),
+        patch("esphome.espidf.framework.download_with_resume"),
+        patch("esphome.espidf.framework.get_system_python_path", return_value="python"),
+        patch("esphome.espidf.framework.BatchDownloadProgress") as progress_cls,
+        patch(
+            "esphome.framework_helpers.ThreadPoolExecutor", wraps=ThreadPoolExecutor
+        ) as pool_cls,
+    ):
+        pool = MagicMock(wraps=ThreadPoolExecutor(max_workers=2))
+        pool_cls.return_value = pool
+        _prefetch_idf_tool_archives(tmp_path, "esp32", ["required"], None)
+
+    pool.shutdown.assert_called_once_with(wait=True, cancel_futures=True)
+    progress_cls.return_value.done.assert_called_once_with()
+
+
 def test_prefetch_passes_targets_and_tools_to_script(tmp_path: Path) -> None:
     with (
         patch(

tests/unit_tests/test_framework_helpers.py

@@ -21,6 +21,7 @@ import requests as req
 from esphome import framework_helpers
 from esphome.core import EsphomeError
 from esphome.framework_helpers import (
+    BatchDownloadProgress,
     _7z_extract_all,
     _detect_archive_root,
     _is_transient_download_error,
@@ -1112,6 +1113,108 @@ class TestDownloadWithResume:
         assert mock_get.call_args[1]["headers"] == {}
         assert dest.read_bytes() == b"data"
 
+    def test_progress_callback_reports_absolute_bytes(self, tmp_path: Path) -> None:
+        """With a callback no bar is drawn; the callback sees the running
+        byte count of this file, then its final verified size."""
+        dest = tmp_path / "tool.tar.gz"
+        resp = _mock_response(b"")
+        resp.headers = {"content-length": "7"}
+        resp.iter_content.return_value = [b"1234", b"567"]
+        seen: list[int] = []
+        with (
+            patch("requests.get", return_value=resp),
+            patch("esphome.framework_helpers.ProgressBar") as bar,
+        ):
+            download_with_resume(
+                "https://example.com/t", dest, size=7, progress=seen.append
+            )
+        assert seen == [0, 4, 7, 7]
+        bar.assert_not_called()
+
+    def test_progress_callback_seeds_with_resume_offset(self, tmp_path: Path) -> None:
+        dest = tmp_path / "tool.tar.gz"
+        (tmp_path / "tool.tar.gz.part").write_bytes(b"12345")
+        good = hashlib.sha256(b"12345678").hexdigest()
+        seen: list[int] = []
+        with patch("requests.get", return_value=_resumed_response(b"678")):
+            download_with_resume(
+                "https://example.com/t", dest, sha256=good, size=8, progress=seen.append
+            )
+        assert seen[0] == 5
+        assert seen[-1] == 8
+
+    def test_progress_callback_credits_already_complete_download(
+        self, tmp_path: Path
+    ) -> None:
+        """A verified dest from an earlier run still counts toward the batch."""
+        dest = tmp_path / "tool.tar.gz"
+        dest.write_bytes(b"12345678")
+        seen: list[int] = []
+        with patch("requests.get") as mock_get:
+            download_with_resume(
+                "https://example.com/t", dest, size=8, progress=seen.append
+            )
+        mock_get.assert_not_called()
+        assert seen == [8]
+
+
+class TestBatchDownloadProgress:
+    def test_sums_trackers_into_one_bar(self) -> None:
+        with patch("esphome.framework_helpers.ProgressBar") as bar_cls:
+            progress = BatchDownloadProgress("Downloading", 100)
+            a = progress.tracker()
+            b = progress.tracker()
+            a(10)
+            b(20)
+            a(30)
+            a(0)  # a restart from zero takes that file's bytes back out
+        bar_cls.assert_called_once_with("Downloading")
+        updates = [c[0][0] for c in bar_cls.return_value.update.call_args_list]
+        assert updates == [0.1, 0.3, 0.5, 0.2]
+
+    def test_clamps_at_one(self) -> None:
+        """Sizes are advisory; an over-delivering server never pushes past 100%."""
+        with patch("esphome.framework_helpers.ProgressBar") as bar_cls:
+            progress = BatchDownloadProgress("Downloading", 10)
+            progress.tracker()(25)
+        assert bar_cls.return_value.update.call_args[0][0] == 1
+
+    def test_unknown_total_draws_nothing(self) -> None:
+        with patch("esphome.framework_helpers.ProgressBar") as bar_cls:
+            progress = BatchDownloadProgress("Downloading", 0)
+            progress.tracker()(5)
+            progress.done()
+        bar_cls.assert_not_called()
+
+    def test_done_ends_an_unfinished_bar(self) -> None:
+        """A batch that stops short of 100% (a failed archive) still ends its
+        line so the next log message starts on a fresh row."""
+        stream = io.StringIO()
+        stream.isatty = lambda: True  # type: ignore[method-assign]
+        with patch("esphome.helpers.sys.stderr", stream):
+            progress = BatchDownloadProgress("Downloading", 10)
+            progress.tracker()(5)
+            progress.done()
+        assert stream.getvalue().endswith("50% \n")
+
+    def test_done_before_any_frame_writes_nothing(self) -> None:
+        """A batch aborted before any tracker fired must not emit a stray
+        newline for a bar that was never drawn."""
+        stream = io.StringIO()
+        stream.isatty = lambda: True  # type: ignore[method-assign]
+        with patch("esphome.helpers.sys.stderr", stream):
+            BatchDownloadProgress("Downloading", 10).done()
+        assert stream.getvalue() == ""
+
+    def test_done_after_full_bar_adds_nothing(self) -> None:
+        stream = io.StringIO()
+        stream.isatty = lambda: True  # type: ignore[method-assign]
+        with patch("esphome.helpers.sys.stderr", stream):
+            progress = BatchDownloadProgress("Downloading", 10)
+            progress.tracker()(10)
+            progress.done()
+        assert stream.getvalue().endswith("100% Done...\r\n")
+
 
 class TestDownloadFromMirrors:
     def test_success_returns_url_and_writes_content(self, tmp_path: Path) -> None:

tests/unit_tests/test_platformio_extra_script.py

@@ -0,0 +1,377 @@
+"""Tests for the shared extraScript machinery (platformio.extra_script)."""
+
+from __future__ import annotations
+
+import logging
+import os
+from pathlib import Path
+from unittest.mock import patch
+
+import pytest
+
+from esphome.core import EsphomeError
+from esphome.platformio.extra_script import (
+    ExtraScriptResult,
+    _FakeSConsEnv,
+    apply_extra_script,
+    captured_as_build_flags,
+    run_extra_script,
+)
+from esphome.platformio.library import ConvertedLibrary as IDFComponent, URLSource
+
+
+def test_extra_script_captures_libpath_libs_and_defines(tmp_path):
+
+    (tmp_path / "src" / "esp32").mkdir(parents=True)
+    script = tmp_path / "extra_script.py"
+    script.write_text(
+        "Import('env')\n"
+        "mcu = env.get('BOARD_MCU')\n"
+        "env.Append(\n"
+        "    LIBPATH=[join('src', mcu)],\n"
+        "    LIBS=['algobsec'],\n"
+        "    CPPDEFINES=['FOO', ('BAR', '1')],\n"
+        "    LINKFLAGS=['-Wl,--gc-sections'],\n"
+        ")\n"
+    )
+    # The script uses bare ``join`` (PIO's extra-scripts run inside SCons
+    # where this is in scope). Inject it via the script header so the
+    # shim's exec namespace can resolve it.
+    script.write_text("from os.path import join\n" + script.read_text())
+
+    result = run_extra_script(
+        script, library_dir=tmp_path, board_mcu="esp32", pio_platform="espressif32"
+    )
+
+    assert result.libpath == [str(Path("src") / "esp32")]
+    assert result.libs == ["algobsec"]
+    assert ("BAR", "1") in result.cppdefines
+    assert "FOO" in result.cppdefines
+    assert result.linkflags == ["-Wl,--gc-sections"]
+
+    flags = captured_as_build_flags(result, library_dir=tmp_path)
+    sep = os.sep
+    assert f"-Lsrc{sep}esp32" in flags
+    assert "-lalgobsec" in flags
+    assert "-DFOO" in flags
+    assert "-DBAR=1" in flags
+    assert "-Wl,--gc-sections" in flags
+
+
+def test_extra_script_libpath_relative_resolves_against_library_dir(
+    tmp_path, monkeypatch
+):
+    """Relative LIBPATH entries must resolve against ``library_dir``, not the
+    caller's CWD (the shim restores CWD before ``captured_as_build_flags``
+    runs)."""
+
+    (tmp_path / "lib" / "esp32").mkdir(parents=True)
+    elsewhere = tmp_path.parent / "not_the_library_dir"
+    elsewhere.mkdir(exist_ok=True)
+    monkeypatch.chdir(elsewhere)
+
+    result = ExtraScriptResult(libpath=["lib/esp32"])
+    flags = captured_as_build_flags(result, library_dir=tmp_path)
+
+    sep = os.sep
+    assert flags == [f"-Llib{sep}esp32"]
+
+
+def test_extra_script_libpath_absolute_outside_library_dir(tmp_path):
+
+    outside = tmp_path.parent / "system_lib"
+    outside.mkdir(exist_ok=True)
+    result = ExtraScriptResult(libpath=[str(outside)])
+
+    flags = captured_as_build_flags(result, library_dir=tmp_path)
+    assert flags == [f"-L{outside.resolve()}"]
+
+
+def test_extra_script_failure_returns_empty_result(tmp_path, caplog):
+
+    script = tmp_path / "broken.py"
+    script.write_text("raise RuntimeError('boom')\n")
+
+    with caplog.at_level("WARNING"):
+        result = run_extra_script(
+            script, library_dir=tmp_path, board_mcu="esp32", pio_platform="espressif32"
+        )
+
+    assert result.libpath == []
+    assert result.libs == []
+    assert "broken.py" in caplog.text
+
+
+def test_apply_extra_script_path_traversal_is_rejected(tmp_path):
+
+    library_dir = tmp_path / "lib"
+    library_dir.mkdir()
+    outside = tmp_path / "evil.py"
+    outside.write_text("env.Append(LIBS=['pwned'])\n")
+
+    c = IDFComponent("owner/name", "1.0", source=URLSource("http://dummy"))
+    c.path = library_dir
+    c.data = {"build": {"extraScript": "../evil.py"}}
+
+    with pytest.raises(EsphomeError, match="escapes the library directory"):
+        apply_extra_script(c, board_mcu=lambda: "esp32", pio_platform="espressif32")
+    # Nothing was folded into flags: the traversal was rejected before
+    # the script could run.
+    assert "flags" not in c.data["build"]
+
+
+def test_apply_extra_script_merges_into_existing_flags(tmp_path):
+
+    (tmp_path / "src").mkdir()
+    script = tmp_path / "extra.py"
+    script.write_text("env.Append(LIBS=['algobsec'])\n")
+
+    c = IDFComponent("owner/name", "1.0", source=URLSource("http://dummy"))
+    c.path = tmp_path
+    c.data = {"build": {"extraScript": "extra.py", "flags": ["-DEXISTING"]}}
+
+    apply_extra_script(c, board_mcu=lambda: "esp32", pio_platform="espressif32")
+
+    assert "-DEXISTING" in c.data["build"]["flags"]
+    assert "-lalgobsec" in c.data["build"]["flags"]
+
+
+def test_apply_extra_script_malformed_flags_raises(tmp_path) -> None:
+    """A null/dict build.flags fails naming the library instead of injecting
+    a non-string into the compiler command line."""
+
+    (tmp_path / "src").mkdir()
+    script = tmp_path / "extra.py"
+    script.write_text("env.Append(LIBS=['algobsec'])\n")
+
+    c = IDFComponent("owner/name", "1.0", source=URLSource("http://dummy"))
+    c.path = tmp_path
+    c.data = {"build": {"extraScript": "extra.py", "flags": None}}
+
+    with pytest.raises(EsphomeError, match="malformed build.flags"):
+        apply_extra_script(c, board_mcu=lambda: "esp32", pio_platform="espressif32")
+
+
+def test_apply_extra_script_callable_target_and_str_flags(tmp_path) -> None:
+    """The shared helper resolves the board_mcu callable lazily and normalizes
+    a string ``build.flags`` value into a list before extending it."""
+
+    (tmp_path / "src").mkdir()
+    script = tmp_path / "extra.py"
+    script.write_text("env.Append(LIBS=[env.get('BOARD_MCU')])\n")
+
+    c = IDFComponent("owner/name", "1.0", source=URLSource("http://dummy"))
+    c.path = tmp_path
+    c.data = {"build": {"extraScript": "extra.py", "flags": "-DBASE=1"}}
+
+    apply_extra_script(c, board_mcu=lambda: "esp8266", pio_platform="espressif8266")
+
+    assert c.data["build"]["flags"] == ["-DBASE=1", "-lesp8266"]
+
+
+def test_captured_dict_cppdefines_warn_and_skip(tmp_path, caplog) -> None:
+    """A dict CPPDEFINES entry (legal SCons) must warn and skip; formatting
+    it blind would hand the compiler -D{'FOO': '1'} garbage."""
+    (tmp_path / "src").mkdir()
+    script = tmp_path / "extra.py"
+    script.write_text(
+        "env.Append(CPPDEFINES=[{'FOO': '1'}, ('BAR', 2), ['BAZ', 3], 'PLAIN'])\n"
+    )
+
+    c = IDFComponent("owner/name", "1.0", source=URLSource("http://dummy"))
+    c.path = tmp_path
+    c.data = {"build": {"extraScript": "extra.py"}}
+
+    apply_extra_script(c, board_mcu=lambda: "esp8266", pio_platform="espressif8266")
+
+    assert c.data["build"]["flags"] == ["-DBAR=2", "-DBAZ=3", "-DPLAIN"]
+    assert "Ignoring unsupported CPPDEFINES entry" in caplog.text
+
+
+def test_apply_extra_script_subscript_env_read(tmp_path) -> None:
+    """Scripts also read env["BOARD_MCU"]; the subscript form must work or
+    the broad handler discards every flag the script captured."""
+    (tmp_path / "src").mkdir()
+    script = tmp_path / "extra.py"
+    script.write_text("env.Append(LIBS=[env['BOARD_MCU']])\n")
+
+    c = IDFComponent("owner/name", "1.0", source=URLSource("http://dummy"))
+    c.path = tmp_path
+    c.data = {"build": {"extraScript": "extra.py"}}
+
+    apply_extra_script(c, board_mcu=lambda: "esp8266", pio_platform="espressif8266")
+
+    assert c.data["build"]["flags"] == ["-lesp8266"]
+
+
+def test_apply_extra_script_no_script_and_no_flags(tmp_path) -> None:
+
+    # No extraScript declared: nothing happens, the target is never resolved
+    c = IDFComponent("owner/name", "1.0", source=URLSource("http://dummy"))
+    c.path = tmp_path
+    c.data = {"build": {}}
+    apply_extra_script(
+        c,
+        board_mcu=lambda: pytest.fail("target resolved without a script"),
+        pio_platform="espressif8266",
+    )
+
+    # A script that captures nothing leaves the flags untouched
+    script = tmp_path / "noop.py"
+    script.write_text("pass\n")
+    c.data = {"build": {"extraScript": "noop.py"}}
+    apply_extra_script(c, board_mcu=lambda: "esp8266", pio_platform="espressif8266")
+    assert "flags" not in c.data["build"]
+
+
+def test_apply_extra_script_ignores_uncaptured_env_calls(tmp_path, caplog) -> None:
+    """Un-captured env vars and unsupported env methods are skipped but
+    diagnosable from the build log."""
+
+    caplog.set_level(logging.DEBUG)
+    script = tmp_path / "extra.py"
+    script.write_text(
+        "env.Replace(CC='clang')\nenv.Append(UNCAPTURED=['x'], LIBS='single')\n"
+    )
+    c = IDFComponent("owner/name", "1.0", source=URLSource("http://dummy"))
+    c.path = tmp_path
+    c.data = {"build": {"extraScript": "extra.py"}}
+    apply_extra_script(c, board_mcu=lambda: "esp8266", pio_platform="espressif8266")
+    assert c.data["build"]["flags"] == ["-lsingle"]
+    assert "env.Append(UNCAPTURED=...) is not captured" in caplog.text
+    assert "env.Replace(...) is not supported" in caplog.text
+
+
+def test_apply_extra_script_swallows_script_errors(tmp_path, caplog) -> None:
+    """A raising extra-script is best-effort: logged and skipped."""
+
+    script = tmp_path / "extra.py"
+    script.write_text("raise RuntimeError('boom')\n")
+    c = IDFComponent("owner/name", "1.0", source=URLSource("http://dummy"))
+    c.path = tmp_path
+    c.data = {"build": {"extraScript": "extra.py"}}
+    apply_extra_script(c, board_mcu=lambda: "esp8266", pio_platform="espressif8266")
+    assert "flags" not in c.data["build"]
+    assert "ignoring its output" in caplog.text
+
+
+def test_apply_extra_script_pio_platform(tmp_path) -> None:
+    """The backend's platform token is exposed to the script as PIOPLATFORM."""
+
+    script = tmp_path / "extra.py"
+    script.write_text("env.Append(LIBS=[env.get('PIOPLATFORM')])\n")
+    c = IDFComponent("owner/name", "1.0", source=URLSource("http://dummy"))
+    c.path = tmp_path
+    c.data = {"build": {"extraScript": "extra.py"}}
+    apply_extra_script(c, board_mcu=lambda: "esp8266", pio_platform="espressif8266")
+    assert c.data["build"]["flags"] == ["-lespressif8266"]
+
+
+def test_apply_extra_script_missing_script_raises(tmp_path) -> None:
+    """A declared but absent extraScript is a broken package and fails by
+    name, as it would under PlatformIO."""
+
+    c = IDFComponent("owner/name", "1.0", source=URLSource("http://dummy"))
+    c.path = tmp_path
+    c.data = {"build": {"extraScript": "nope.py"}}
+    with pytest.raises(EsphomeError, match="nope.py of library owner/name not found"):
+        apply_extra_script(c, board_mcu=lambda: "esp8266", pio_platform="espressif8266")
+
+
+def test_run_extra_script_failure_discards_partial_capture(tmp_path, caplog) -> None:
+    """A crashed script yields an empty result: half-applied flags could
+    build wrong-output firmware that links cleanly."""
+
+    script = tmp_path / "extra.py"
+    script.write_text("env.Append(LIBS=['algobsec'])\nraise RuntimeError('boom')\n")
+    result = run_extra_script(
+        script, library_dir=tmp_path, board_mcu="esp32", pio_platform="espressif32"
+    )
+    assert result.libs == []
+    assert "ignoring its output" in caplog.text
+
+
+def test_run_extra_script_syntax_error_is_best_effort(tmp_path, caplog) -> None:
+    """A vendored script that does not even compile warns and skips instead
+    of aborting the build."""
+
+    script = tmp_path / "extra.py"
+    script.write_text("def broken(:\n")
+    result = run_extra_script(
+        script, library_dir=tmp_path, board_mcu="esp32", pio_platform="espressif32"
+    )
+    assert result.libs == []
+    assert "ignoring its output" in caplog.text
+
+
+def test_unsupported_env_method_warns_once(caplog) -> None:
+    """Repeated calls to the same unsupported method warn only once."""
+
+    env = _FakeSConsEnv(
+        board_mcu="esp8266", pio_env="esphome_esp8266", pio_platform="espressif8266"
+    )
+    env.Replace(CC="clang")
+    env.Replace(CC="gcc")
+    assert caplog.text.count("env.Replace(...) is not supported") == 1
+
+
+def test_run_extra_script_sys_exit_is_best_effort(tmp_path, caplog) -> None:
+    """A nonzero sys.exit() in a vendored script must not kill the esphome
+    run, and its output is discarded."""
+
+    script = tmp_path / "extra.py"
+    script.write_text("import sys\nenv.Append(LIBS=['x'])\nsys.exit(3)\n")
+    result = run_extra_script(
+        script, library_dir=tmp_path, board_mcu="esp32", pio_platform="espressif32"
+    )
+    assert result.libs == []
+    assert "exited with status 3" in caplog.text
+
+
+def test_run_extra_script_sys_exit_zero_is_success(tmp_path, caplog) -> None:
+    """sys.exit(0) is a normal PlatformIO script ending: the capture is kept."""
+
+    script = tmp_path / "extra.py"
+    script.write_text("import sys\nenv.Append(LIBS=['algobsec'])\nsys.exit(0)\n")
+    result = run_extra_script(
+        script, library_dir=tmp_path, board_mcu="esp32", pio_platform="espressif32"
+    )
+    assert result.libs == ["algobsec"]
+    assert "ignoring its output" not in caplog.text
+
+
+def test_run_extra_script_unreadable_raises(tmp_path) -> None:
+    """An unreadable declared script is a broken package, like a missing one."""
+
+    script = tmp_path / "extra.py"
+    script.write_text("")
+    with (
+        patch("pathlib.Path.read_text", side_effect=OSError("denied")),
+        pytest.raises(EsphomeError, match="is unreadable"),
+    ):
+        run_extra_script(
+            script, library_dir=tmp_path, board_mcu="esp32", pio_platform="espressif32"
+        )
+
+
+def test_run_extra_script_bad_encoding_is_best_effort(tmp_path, caplog) -> None:
+    """Undecodable content warns and skips, like a SyntaxError."""
+
+    script = tmp_path / "extra.py"
+    script.write_bytes(b"\xff\xfe\x00bad")
+    result = run_extra_script(
+        script, library_dir=tmp_path, board_mcu="esp32", pio_platform="espressif32"
+    )
+    assert result.libs == []
+    assert "is not UTF-8" in caplog.text
+
+
+def test_uncaptured_append_key_warns_once(caplog) -> None:
+    """A loop of Appends to the same uncaptured key warns once."""
+
+    env = _FakeSConsEnv(
+        board_mcu="esp8266", pio_env="esphome_esp8266", pio_platform="espressif8266"
+    )
+    env.Append(CPPPATH=["a"])
+    env.Append(CPPPATH=["b"])
+    assert caplog.text.count("env.Append(CPPPATH=...) is not captured") == 1

tests/unit_tests/test_platformio_library.py

@@ -4,15 +4,18 @@ Covers the shared download/parse/resolve/dependency-walk paths in
 ``esphome.platformio.library`` directly (the ESP-IDF and Zephyr backends are
 exercised in their own test modules)."""
 
+from contextlib import contextmanager
 import json
 import logging
 from pathlib import Path
+from types import SimpleNamespace
 
 import pytest
 
 from esphome.core import EsphomeError, Library
 import esphome.platformio.library as lib
 from esphome.platformio.library import (
+    SOURCE_KIND_FOR_SUFFIX,
     ConvertedLibrary,
     GitSource,
     InvalidLibrary,
@@ -23,6 +26,8 @@ from esphome.platformio.library import (
     _resolve_registry_version,
     check_library_data,
     convert_libraries,
+    join_flag_args,
+    split_flag_entry,
 )
 
 
@@ -150,11 +155,30 @@ def test_localsource_download_returns_empty_build_dir(setup_core: Path) -> None:
     assert plain != out
 
 
+@contextmanager
+def caplog_at_info():
+    records: list[logging.LogRecord] = []
+    handler = logging.Handler()
+    handler.emit = records.append
+    logger = logging.getLogger("esphome.platformio.library")
+    logger.addHandler(handler)
+    # The level must actually admit INFO or the no-INFO assertions are vacuous
+    old_level = logger.level
+    logger.setLevel(logging.INFO)
+    try:
+        yield records
+    finally:
+        logger.setLevel(old_level)
+        logger.removeHandler(handler)
+
+
 def test_urlsource_download_extracts_then_reuses_marker(setup_core, monkeypatch):
     monkeypatch.setattr(lib, "rmdir", lambda path, msg="": None)
     dl_calls: list[list[str]] = []
     monkeypatch.setattr(
-        lib, "download_from_mirrors", lambda urls, headers, f: dl_calls.append(urls)
+        lib,
+        "download_from_mirrors",
+        lambda urls, headers, f, progress=None: dl_calls.append(urls),
     )
 
     def fake_extract(fileobj, path):
@@ -173,6 +197,12 @@ def test_urlsource_download_extracts_then_reuses_marker(setup_core, monkeypatch)
     assert out2 == out
     assert len(dl_calls) == 1
 
+    # A batch caller passes a tracker and owns the messaging; no per-file INFO
+    with caplog_at_info() as records:
+        src.download("mylib-batch", progress=lambda done: None)
+    assert len(dl_calls) == 2
+    assert not [r for r in records if "Downloading" in r.message]
+
 
 def test_resolve_registry_version_raises_without_pkg_file(monkeypatch):
     registry = lib._make_registry_client()
@@ -213,7 +243,7 @@ def _patch_registry_resolve(monkeypatch: pytest.MonkeyPatch) -> None:
 def _patch_download_with_manifests(monkeypatch, tmp_path, manifests, *, properties=()):
     """Fake ConvertedLibrary.download to materialize canned manifests on disk."""
 
-    def fake_download(self, force=False, salt="", namespace=""):
+    def fake_download(self, force=False, salt="", namespace="", progress=None):
         self.path = tmp_path / self.get_require_name()
         self.path.mkdir(parents=True, exist_ok=True)
         if self.name in properties:
@@ -292,7 +322,11 @@ def _patch_download_without_manifest(
     calls: list[bool] = []
 
     def fake_download(
-        self: ConvertedLibrary, force: bool = False, salt: str = "", namespace: str = ""
+        self: ConvertedLibrary,
+        force: bool = False,
+        salt: str = "",
+        namespace: str = "",
+        progress=None,
     ) -> None:
         calls.append(force)
         self.path = tmp_path / self.get_require_name()
@@ -531,3 +565,322 @@ def test_convert_libraries_skips_incompatible_dependency(tmp_path, monkeypatch):
     top = convert_libraries([Library("esphome/A", "1.0.0", None)], _backend())
 
     assert top[0].dependencies == []
+
+
+def test_split_flag_entry_unbalanced_quote_is_clean() -> None:
+    """A malformed flags entry raises EsphomeError, not a raw ValueError."""
+
+    assert split_flag_entry('-DX="a b"', "library x") == ["-DX=a b"]
+    with pytest.raises(EsphomeError, match=r"Malformed build flag.*library x"):
+        split_flag_entry('-DX="unclosed', "library x")
+
+
+def test_join_flag_args_reglues_spaced_define() -> None:
+    """A spaced -D re-glues to its argument, as ParseFlags does."""
+
+    assert join_flag_args(["-D", "FOO=1", "-Os"], "x") == ["-DFOO=1", "-Os"]
+
+
+def test_join_flag_args_trailing_bare_flag_warns(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+
+    assert join_flag_args(["-Os", "-l"], "library x") == ["-Os"]
+    assert "Ignoring trailing '-l'" in caplog.text
+
+
+def test_lex_build_flags_dangling_flag_does_not_cross_entries(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """Each entry is lexed independently, as ParseFlags does: a dangling -I
+    ending one entry warns instead of absorbing the next entry's first token."""
+    from esphome.platformio.library import lex_build_flags
+
+    assert lex_build_flags(["-Wall -I", "-DFOO=1"], "lib x") == ["-Wall", "-DFOO=1"]
+    assert "Ignoring trailing '-I'" in caplog.text
+
+
+def test_prefetch_wave_downloads_registry_archives_in_parallel(
+    setup_core, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
+) -> None:
+    """Registry archives in one wave download concurrently, deduped by URL;
+    git/local sources and failures are left to the sequential call."""
+    calls: list[str] = []
+
+    def fake_download(self, force=False, salt="", namespace="", progress=None):
+        calls.append(self.source.url)
+        if progress is not None:
+            progress(0)
+        if "boom" in self.source.url:
+            raise RuntimeError("boom")
+
+    monkeypatch.setattr(ConvertedLibrary, "download", fake_download)
+    monkeypatch.setattr(lib, "_content_lengths", lambda urls: [1] * len(urls))
+    wave = [
+        ("a", ConvertedLibrary("a", "1.0", URLSource("https://x/a.tar.gz"))),
+        ("b", ConvertedLibrary("b", "1.0", URLSource("https://x/b.tar.gz"))),
+        # Duplicate URL must prefetch once (two threads must never extract
+        # into the same cache directory)
+        ("b2", ConvertedLibrary("b2", "1.0", URLSource("https://x/b.tar.gz"))),
+        ("c", ConvertedLibrary("c", "1.0", URLSource("https://x/boom.tar.gz"))),
+        ("g", ConvertedLibrary("g", "*", lib.GitSource("https://x/g.git", None))),
+    ]
+    lib._prefetch_wave(wave, "", "idf")
+    assert sorted(calls) == [
+        "https://x/a.tar.gz",
+        "https://x/b.tar.gz",
+        "https://x/boom.tar.gz",
+    ]
+    # The failure surfaces at default verbosity, after the bar
+    assert "Prefetch of c failed (retrying sequentially)" in caplog.text
+
+
+def test_prefetch_wave_unknown_size_falls_back_to_sequential(
+    setup_core, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
+) -> None:
+    """Any unknown HEAD size skips the parallel prefetch entirely so the
+    sequential downloads keep their per-file bars."""
+
+    def fail_download(self, force=False, salt="", namespace="", progress=None):
+        raise AssertionError("prefetched despite unknown size")
+
+    monkeypatch.setattr(ConvertedLibrary, "download", fail_download)
+    monkeypatch.setattr(lib, "_content_lengths", lambda urls: [1, None])
+    wave = [
+        ("a", ConvertedLibrary("a", "1.0", URLSource("https://x/a.tar.gz"))),
+        ("b", ConvertedLibrary("b", "1.0", URLSource("https://x/b.tar.gz"))),
+    ]
+    caplog.set_level("DEBUG")
+    lib._prefetch_wave(wave, "", "idf")
+    # The culprit URL is named so the fallback is traceable
+    assert "No Content-Length for https://x/b.tar.gz" in caplog.text
+
+
+def test_content_lengths_head_requests(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Sizes come from HEAD Content-Length; a failing HEAD reads as 0 so
+    the combined bar is skipped rather than wrong."""
+    import requests
+
+    def fake_head(url, timeout, allow_redirects):
+        if "bad" in url:
+            raise requests.ConnectionError("down")
+        if "gone" in url:
+            return SimpleNamespace(ok=False, status_code=404, headers={})
+        if "garbage" in url:
+            # A proxy/CDN doubling the header ("123, 123") or emitting junk
+            # must degrade to unknown, not ValueError the build
+            return SimpleNamespace(ok=True, headers={"content-length": "123, 123"})
+        return SimpleNamespace(ok=True, headers={"content-length": "123"})
+
+    monkeypatch.setattr(
+        lib.requests if hasattr(lib, "requests") else requests, "head", fake_head
+    )
+    # None marks an unknown size (probe failure or non-2xx), distinct
+    # from a genuine zero
+    assert lib._content_lengths(
+        ["https://x/a", "https://x/bad", "https://x/gone", "https://x/garbage"]
+    ) == [
+        123,
+        None,
+        None,
+        None,
+    ]
+
+
+def test_prefetch_wave_cache_probe_failure_still_prefetches(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """The cache probe is best-effort; a failing probe prefetches anyway."""
+    calls: list[str] = []
+    monkeypatch.setattr(
+        ConvertedLibrary,
+        "download",
+        lambda self, **kw: calls.append(self.source.url),
+    )
+    monkeypatch.setattr(
+        URLSource,
+        "is_cached",
+        lambda self, *a, **kw: (_ for _ in ()).throw(RuntimeError("no core")),
+    )
+    monkeypatch.setattr(lib, "_content_lengths", lambda urls: [1] * len(urls))
+    wave = [
+        ("a", ConvertedLibrary("a", "1.0", URLSource("https://x/a.tar.gz"))),
+        ("b", ConvertedLibrary("b", "1.0", URLSource("https://x/b.tar.gz"))),
+    ]
+    lib._prefetch_wave(wave, "", "idf")
+    assert sorted(calls) == ["https://x/a.tar.gz", "https://x/b.tar.gz"]
+
+
+def test_prefetch_wave_warm_cache_is_silent(
+    setup_core, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
+) -> None:
+    """Already-extracted archives download nothing; a warm build must not
+    print a Downloading line or draw a bar."""
+    monkeypatch.setattr(
+        ConvertedLibrary,
+        "download",
+        lambda self, **kw: (_ for _ in ()).throw(AssertionError("downloaded")),
+    )
+    wave = []
+    for name in ("a", "b", "c"):
+        comp = ConvertedLibrary(name, "1.0", URLSource(f"https://x/{name}.tar.gz"))
+        marker_dir = comp.source._cache_dir(comp.get_sanitized_name(), "", "idf")
+        marker_dir.mkdir(parents=True)
+        (marker_dir / ".esphome_extracted").touch()
+        wave.append((name, comp))
+    lib._prefetch_wave(wave, "", "idf")
+    assert "Downloading" not in caplog.text
+
+
+def test_prefetch_wave_single_archive_skips_the_pool(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """One archive gains nothing from a pool; the sequential call keeps its
+    progress bar."""
+    monkeypatch.setattr(
+        ConvertedLibrary,
+        "download",
+        lambda self, **kw: (_ for _ in ()).throw(AssertionError("prefetched")),
+    )
+    lib._prefetch_wave(
+        [("a", ConvertedLibrary("a", "1.0", URLSource("https://x/a.tar.gz")))],
+        "",
+        "idf",
+    )
+
+
+def test_normalize_dependencies_forms(caplog) -> None:
+    """Every PIO-legal spelling normalizes; unrecognizable entries warn."""
+    from esphome.platformio.library import normalize_dependencies
+
+    assert normalize_dependencies(
+        ["Wire", {"name": "SPI"}, 5, "", {"version": "1.0"}], "libx"
+    ) == [
+        {"name": "Wire"},
+        {"name": "SPI"},
+    ]
+    # The int, the empty string, and the nameless dict all warn
+    assert caplog.text.count("unrecognized dependency entry") == 3
+    # A plain string is names, never iterated into characters
+    assert normalize_dependencies("Wire, SPI") == [
+        {"name": "Wire"},
+        {"name": "SPI"},
+    ]
+    assert normalize_dependencies("Wire") == [{"name": "Wire"}]
+    # A non-iterable value fails by manifest name, never a bare TypeError
+    assert normalize_dependencies(5, "libx") == []
+    assert "Ignoring unrecognized dependencies 5 of libx" in caplog.text
+    # The dict-shorthand form validates names like the list form: an empty
+    # key and a spec overriding name with a non-string both warn and drop
+    assert normalize_dependencies(
+        {"": "1.0", "Wire": {"name": 123, "version": "1.0"}, "SPI": "*"}, "libx"
+    ) == [{"name": "SPI", "owner": None, "version": "*"}]
+    assert caplog.text.count("unrecognized dependency entry") == 5
+    # A container or numeric version would raise from set.add() or fail
+    # opaquely in the registry; both spellings warn and drop
+    assert normalize_dependencies({"Foo": ["1.0", "2.0"]}, "libx") == []
+    assert normalize_dependencies([{"name": "Foo", "version": 1}], "libx") == []
+    assert caplog.text.count("unrecognized dependency entry") == 7
+
+
+@pytest.mark.parametrize(
+    "manifest", [["not", "a", "manifest"], {"name": "A", "build": "src"}]
+)
+def test_convert_libraries_malformed_manifest_raises(
+    tmp_path, monkeypatch, manifest
+) -> None:
+    """A manifest without the expected dict shape fails by library name
+    before any backend dereferences data/build."""
+    _patch_download_with_manifests(monkeypatch, tmp_path, {"esphome/A": manifest})
+    with pytest.raises(EsphomeError, match="has a malformed manifest"):
+        convert_libraries([Library("esphome/A", None, None)], _backend())
+
+
+def test_walk_warns_for_properties_only_depends(
+    tmp_path, monkeypatch, caplog: pytest.LogCaptureFixture
+) -> None:
+    """A manifest declaring dependencies only as library.properties depends=
+    warns in the shared walk, so every backend reports the drop."""
+    _patch_download_with_manifests(
+        monkeypatch,
+        tmp_path,
+        {"esphome/A": "name=A\nversion=1.0\ndepends=Wire, SPI\n"},
+        properties=("esphome/A",),
+    )
+    convert_libraries([Library("esphome/A", "1.0.0", None)], _backend())
+    assert "declares dependencies via library.properties" in caplog.text
+
+
+def test_walk_warns_for_nonplatform_invalid_library(
+    tmp_path, monkeypatch, caplog: pytest.LogCaptureFixture
+) -> None:
+    """A dependency dropped for any cause other than the routine platform
+    filter is visible in every backend."""
+    _patch_download_with_manifests(
+        monkeypatch,
+        tmp_path,
+        {"esphome/A": {"name": "A", "dependencies": [{"name": "B", "version": "1.0"}]}},
+    )
+    calls = {"n": 0}
+    real = lib.check_library_data
+
+    def flaky(data, platform, framework):
+        calls["n"] += 1
+        if calls["n"] > 1:
+            raise InvalidLibrary("manifest is corrupt")
+        return real(data, platform, framework)
+
+    monkeypatch.setattr(lib, "check_library_data", flaky)
+    convert_libraries([Library("esphome/A", None, None)], _backend())
+    assert "Skipping dependency B of esphome/A: manifest is corrupt" in caplog.text
+
+
+def test_convert_libraries_warns_for_nonplatform_invalid_dependency_component(
+    tmp_path, monkeypatch, caplog: pytest.LogCaptureFixture
+) -> None:
+    """A dependency component dropped for any cause other than the platform
+    filter warns; only the routine cross-platform skip stays at debug."""
+    _patch_download_with_manifests(
+        monkeypatch,
+        tmp_path,
+        {
+            "esphome/A": {
+                "name": "A",
+                "dependencies": [{"name": "C", "owner": "esphome", "version": "1.0"}],
+            },
+            "esphome/C": {"name": "C"},
+        },
+    )
+    real = lib.check_library_data
+
+    def flaky(data, platform, framework):
+        # Fail only on C's resolved manifest, not on A's dependency entry
+        if data.get("name") == "C" and "version" not in data:
+            raise InvalidLibrary("manifest is corrupt")
+        return real(data, platform, framework)
+
+    monkeypatch.setattr(lib, "check_library_data", flaky)
+    convert_libraries([Library("esphome/A", "1.0.0", None)], _backend())
+    assert "manifest is corrupt" in caplog.text
+    assert "Skipping dependency" in caplog.text
+
+
+def test_split_flag_entry_non_string_is_clean() -> None:
+    """A dict or number from a third-party manifest fails naming the entry,
+    not with an opaque shlex traceback."""
+
+    with pytest.raises(EsphomeError, match="Malformed build flag"):
+        split_flag_entry({"esp32": ["-DX"]}, "lib x")
+    with pytest.raises(EsphomeError, match="Malformed build flag 5"):
+        split_flag_entry(5, "lib x")
+
+
+def test_source_kind_map_shape() -> None:
+    """The kind values the native compile rules key on, and the deliberate
+    AS/ASPP merge (.s and .S both map to asm)."""
+
+    assert set(SOURCE_KIND_FOR_SUFFIX.values()) == {"c", "cxx", "asm"}
+    assert SOURCE_KIND_FOR_SUFFIX[".s"] == "asm"
+    assert SOURCE_KIND_FOR_SUFFIX[".S"] == "asm"
+    assert SOURCE_KIND_FOR_SUFFIX[".c"] == "c"
+    assert SOURCE_KIND_FOR_SUFFIX[".cpp"] == "cxx"

tests/unit_tests/test_platformio_toolchain.py

@@ -1977,3 +1977,10 @@ def test_run_platformio_cli_invokes_heal(
     with patch.object(toolchain, "heal_platformio_python_env") as mock_heal:
         toolchain.run_platformio_cli("test")
     mock_heal.assert_called_once()
+
+
+def test_ccache_probe_spawns_with_close_fds_false() -> None:
+    """The probe follows the repo-wide posix_spawn convention."""
+    with patch("subprocess.run") as mock_run:
+        assert toolchain._ccache_runs("/usr/bin/ccache") is True
+    assert mock_run.call_args.kwargs["close_fds"] is False