Add the ESP8266 Arduino framework and toolchain installer

esphome/arduino8266/__init__.py

@@ -0,0 +1,9 @@
+"""Native (PlatformIO-free) build support for the ESP8266 Arduino core.
+
+This package downloads the Arduino ESP8266 core and the xtensa-lx106
+toolchain, generates a ninja build for them plus the ESPHome sources, and
+drives the build directly — the ESP8266 equivalent of ``esphome.espidf``.
+
+Deliberately importable without the esp8266 component to avoid circular
+imports; the component wires these modules in via lazy imports.
+"""

esphome/arduino8266/framework.py

@@ -0,0 +1,335 @@
+"""Download and install the Arduino ESP8266 core, toolchain, and ninja.
+
+Artifacts land in a machine-global cache (shared across projects, like the
+ESP-IDF install in ``esphome.espidf.framework``):
+
+    <cache>/arduino8266/frameworks/<version>/   framework-arduinoespressif8266
+    <cache>/arduino8266/toolchains/<version>/   toolchain-xtensa (gcc 10.3)
+
+ninja itself comes from PATH or the ninja PyPI wheel (a requirements.txt
+dependency), so only the two packages above are downloaded here.
+
+Sources default to the PlatformIO registry (the exact packages the PlatformIO
+toolchain has always used, so the bits are identical); the
+``ESPHOME_ARDUINO8266_*_MIRRORS`` environment variables override the URLs with
+``{VERSION}`` / ``{SYSTEM}`` substitution.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Collection
+import functools
+import logging
+import os
+from pathlib import Path
+import platform
+import shutil
+
+import platformdirs
+
+import esphome.config_validation as cv
+from esphome.core import CORE, EsphomeError
+from esphome.framework_helpers import (
+    archive_extract_all,
+    download_from_mirrors,
+    download_with_resume,
+    rmdir,
+    str_to_lst_of_str,
+)
+from esphome.helpers import get_bool_env, get_str_env
+from esphome.platformio.library import ensure_list
+
+_LOGGER = logging.getLogger(__name__)
+
+FRAMEWORK_PACKAGE = "framework-arduinoespressif8266"
+TOOLCHAIN_PACKAGE = "toolchain-xtensa"
+# gcc 10.3, the toolchain Arduino core 3.x builds with. The compile flags in
+# the build generator are tuned to it; treat version changes as a full
+# reinstall (the install dir is keyed on the version).
+TOOLCHAIN_VERSION = "2.100300.220621"
+
+_REGISTRY_URL = (
+    "https://api.registry.platformio.org/v3/packages/platformio/tool/{package}"
+)
+
+ESPHOME_ARDUINO8266_FRAMEWORK_MIRRORS = str_to_lst_of_str(
+    os.environ.get("ESPHOME_ARDUINO8266_FRAMEWORK_MIRRORS", "")
+)
+ESPHOME_ARDUINO8266_TOOLCHAIN_MIRRORS = str_to_lst_of_str(
+    os.environ.get("ESPHOME_ARDUINO8266_TOOLCHAIN_MIRRORS", "")
+)
+
+
+def get_arduino8266_tools_path() -> Path:
+    # Treat an empty/whitespace prefix as unset: Path("") resolves to the CWD,
+    # which clean-all would then delete.
+    if prefix := get_str_env("ESPHOME_ARDUINO8266_PREFIX", "").strip():
+        path = Path(prefix).expanduser()
+    else:
+        # Machine-global so all projects share one install; see
+        # espidf.framework.get_idf_tools_path for the location rationale.
+        path = (
+            Path(platformdirs.user_cache_dir("esphome", appauthor=False))
+            / "arduino8266"
+        )
+    return path.resolve()
+
+
+# 3.1.1 rather than 3.1.0: the registry has no package for 3.1.0
+MIN_FRAMEWORK_VERSION = cv.Version(3, 1, 1)
+
+
+def framework_package_version(ver: cv.Version) -> str:
+    """Map an Arduino core version (e.g. 3.1.2) to its package version.
+
+    Same encoding as the PlatformIO package registry uses for core 3.x
+    releases (3.1.2 -> 3.30102.0). The native toolchain only supports core
+    >= MIN_FRAMEWORK_VERSION, so the 1.x/2.x encodings never apply here.
+    """
+    return f"3.{ver.major}{ver.minor:02d}{ver.patch:02d}.0"
+
+
+def get_framework_path(package_version: str) -> Path:
+    return get_arduino8266_tools_path() / "frameworks" / package_version
+
+
+def get_toolchain_path() -> Path:
+    return get_arduino8266_tools_path() / "toolchains" / TOOLCHAIN_VERSION
+
+
+def _downloads_path() -> Path:
+    path = get_arduino8266_tools_path() / "downloads"
+    path.mkdir(parents=True, exist_ok=True)
+    return path
+
+
+def _pio_system() -> str:
+    """The PlatformIO registry system tag for the current host.
+
+    Hand-rolled instead of ``platformio.util.get_systype()`` so this backend
+    never imports the PlatformIO package. The windows-arm64 and darwin-arm64
+    mappings are deliberate: the toolchain packages ship x86_64 binaries for
+    those hosts (Rosetta / x86 emulation).
+    """
+    sysname = platform.system().lower()
+    machine = platform.machine().lower()
+    if sysname == "darwin":
+        return "darwin_arm64" if machine == "arm64" else "darwin_x86_64"
+    if sysname == "windows":
+        return "windows_amd64" if machine in ("amd64", "arm64") else "windows_x86"
+    if sysname == "linux":
+        if machine in ("arm64", "aarch64"):
+            return "linux_aarch64"
+        if machine in ("i686", "i386", "x86"):
+            return "linux_i686"
+        if machine.startswith("arm"):
+            return f"linux_{machine}"
+        if machine in ("x86_64", "amd64"):
+            return "linux_x86_64"
+    # Fail here, near the cause, rather than installing a toolchain whose
+    # binaries cannot execute on this host.
+    raise EsphomeError(
+        f"No {sysname}/{machine} build of the ESP8266 toolchain exists; "
+        "use 'toolchain: platformio'"
+    )
+
+
+def _registry_download(package: str, version: str) -> tuple[str, str, int | None]:
+    """Resolve a package's download URL, sha256, and size via the PIO registry."""
+    import requests
+
+    url = _REGISTRY_URL.format(package=package)
+    last_err: Exception | None = None
+    for _ in range(3):
+        try:
+            resp = requests.get(url, timeout=30)
+            resp.raise_for_status()
+            data = resp.json()
+            break
+        except requests.RequestException as err:
+            last_err = err
+    else:
+        # A clean, retried error like the other download paths in the tree
+        raise EsphomeError(
+            f"Could not query the package registry for {package}: {last_err}"
+        ) from last_err
+    system = _pio_system()
+    for ver in data.get("versions", []):
+        if ver.get("name") != version:
+            continue
+        for file in ver.get("files", []):
+            # ensure_list: a bare string would make ``in`` a substring test
+            systems = ensure_list(file.get("system") or "*")
+            if "*" in systems or system in systems:
+                sha256 = (file.get("checksum") or {}).get("sha256")
+                if not sha256:
+                    # Never extract an unverified archive; the registry
+                    # publishes a checksum for every package file.
+                    raise EsphomeError(
+                        f"The package registry returned no sha256 for "
+                        f"{package} {version}; refusing the unverified download"
+                    )
+                return (file["download_url"], sha256, file.get("size"))
+        raise EsphomeError(f"No {package} {version} build for this platform ({system})")
+    raise EsphomeError(f"{package} {version} not found in the package registry")
+
+
+def _install_package(
+    name: str,
+    version: str,
+    dest: Path,
+    mirrors: list[str],
+    expect: Collection[str] = (),
+) -> None:
+    """Download, verify, and extract one package if not already installed.
+
+    The registry path is integrity-checked against the sha256 the registry
+    publishes; a mirror override is trusted as configured.
+    """
+    marker = dest / ".esphome_extracted"
+    if marker.is_file():
+        return
+    from filelock import FileLock
+
+    # The cache is machine-global; serialize concurrent cold builds so one
+    # process cannot wipe the directory another is extracting into (same
+    # filelock pattern as platformio/toolchain.py and git.py).
+    dest.parent.mkdir(parents=True, exist_ok=True)
+    with FileLock(f"{dest}.lock"):
+        if marker.is_file():
+            # Another process finished the install while we waited
+            return
+        rmdir(dest, msg=f"Clean up incomplete {name} install")
+        # A persistent download location (not a temp dir) so an interrupted
+        # download resumes across esphome runs via download_with_resume's
+        # .part file, mirroring the espidf dist/ convention.
+        archive = _downloads_path() / f"{name}-{version}"
+        _LOGGER.info("Downloading %s %s ...", name, version)
+        if mirrors:
+            _LOGGER.warning(
+                "Downloading %s from a mirror override; checksum verification "
+                "is skipped for mirrors",
+                name,
+            )
+            download_from_mirrors(
+                mirrors, {"VERSION": version, "SYSTEM": _pio_system()}, archive
+            )
+        else:
+            url, sha256, size = _registry_download(name, version)
+            download_with_resume(url, archive, sha256=sha256, size=size)
+        _LOGGER.info("Extracting %s ...", name)
+        archive_extract_all(archive, dest, progress_header="Extracting")
+        # Validate the layout before recording success, so an unexpected
+        # package is never cached as a working install.
+        for rel in expect:
+            if not (dest / rel).is_dir():
+                raise EsphomeError(
+                    f"{name} {version} extracted without the expected {rel} "
+                    "directory; run 'esphome clean-all' and retry"
+                )
+        marker.touch()
+        archive.unlink(missing_ok=True)
+
+
+def _find_ninja() -> Path:
+    """Locate the ninja binary: PATH first, else the ninja PyPI wheel.
+
+    The wheel is a requirements.txt dependency, so pip has already
+    integrity-checked it; no download logic is needed here.
+    """
+    if binary := shutil.which("ninja"):
+        return Path(binary)
+    try:
+        import ninja
+    except ImportError as err:
+        raise EsphomeError(
+            "ninja not found on PATH or in the ninja package; reinstall the "
+            "esphome Python environment"
+        ) from err
+
+    binary = Path(ninja.BIN_DIR) / ("ninja.exe" if os.name == "nt" else "ninja")
+    if not binary.is_file():
+        raise EsphomeError(
+            "ninja not found on PATH or in the ninja package; reinstall the "
+            "esphome Python environment"
+        )
+    return binary
+
+
+def check_and_install(framework_version: cv.Version) -> dict[str, Path]:
+    """Ensure framework, toolchain, and ninja are installed; return their paths."""
+    package_version = framework_package_version(framework_version)
+    framework_path = get_framework_path(package_version)
+    _install_package(
+        FRAMEWORK_PACKAGE,
+        package_version,
+        framework_path,
+        ESPHOME_ARDUINO8266_FRAMEWORK_MIRRORS,
+        expect=("cores/esp8266", "tools/sdk", "libraries"),
+    )
+    toolchain_path = get_toolchain_path()
+    _install_package(
+        TOOLCHAIN_PACKAGE,
+        TOOLCHAIN_VERSION,
+        toolchain_path,
+        ESPHOME_ARDUINO8266_TOOLCHAIN_MIRRORS,
+        expect=("bin",),
+    )
+    return {
+        "framework_path": framework_path,
+        "toolchain_path": toolchain_path,
+        "ninja_path": _find_ninja(),
+    }
+
+
+def get_build_env(toolchain_path: Path) -> dict[str, str]:
+    env = os.environ.copy()
+    env["PATH"] = str(toolchain_path / "bin") + os.pathsep + env.get("PATH", "")
+    env.update(ccache_env())
+    return env
+
+
+@functools.cache
+def ccache_path() -> str | None:
+    """The ccache binary to prefix compiles with, or None when disabled.
+
+    Same convention as the PlatformIO path: on by default when the binary is
+    on PATH, ``ESPHOME_CCACHE_ENABLE=0`` disables it, and an explicit ``=1``
+    warns when no binary is found and skips the runnability probe.
+    """
+    from esphome.platformio.toolchain import _ccache_runs, _strip_win_long_path_prefix
+
+    explicit = "ESPHOME_CCACHE_ENABLE" in os.environ
+    if explicit and not get_bool_env("ESPHOME_CCACHE_ENABLE"):
+        return None
+    ccache = shutil.which("ccache")
+    if ccache is None:
+        if explicit:
+            _LOGGER.warning(
+                "ESPHOME_CCACHE_ENABLE is set but no ccache binary is on PATH; "
+                "compiling without ccache"
+            )
+        return None
+    ccache = _strip_win_long_path_prefix(ccache)
+    if not explicit and not _ccache_runs(ccache):
+        return None
+    return ccache
+
+
+def ccache_env() -> dict[str, str]:
+    """Return ccache settings for the build subprocess (not os.environ).
+
+    Mirrors ``espidf.framework._ccache_env``: cache under the machine-global
+    tools dir, depend mode (gcc emits depfiles via -MMD), and CCACHE_BASEDIR
+    scoped to the build dir so devices share framework cache entries. Values
+    the user already set in the environment are respected.
+    """
+    if ccache_path() is None:
+        return {}
+    defaults = {
+        "CCACHE_DIR": str(get_arduino8266_tools_path() / "ccache"),
+        "CCACHE_NOHASHDIR": "true",
+        "CCACHE_DEPEND": "1",
+        "CCACHE_BASEDIR": str(Path(CORE.build_path).resolve()),
+    }
+    return {k: v for k, v in defaults.items() if k not in os.environ}

esphome/writer.py

@@ -660,11 +660,17 @@ def clean_all(configuration: list[str]):
     # that live outside it.
     import platformdirs
 
+    from esphome.arduino8266.framework import get_arduino8266_tools_path
     from esphome.components.nrf52.framework import get_sdk_nrf_tools_path
     from esphome.espidf.framework import get_idf_tools_path
 
     cache_root = Path(platformdirs.user_cache_dir("esphome", appauthor=False)).resolve()
-    for install_path in (cache_root, get_idf_tools_path(), get_sdk_nrf_tools_path()):
+    for install_path in (
+        cache_root,
+        get_idf_tools_path(),
+        get_sdk_nrf_tools_path(),
+        get_arduino8266_tools_path(),
+    ):
         if install_path.is_dir():
             _LOGGER.info("Deleting %s", install_path)
             rmtree(install_path)

requirements.txt

@@ -28,6 +28,7 @@ smpclient==7.2.0
 requests==2.34.2
 py7zr==1.1.3
 platformdirs==4.11.3  # native esp-idf toolchain global cache dir
+ninja==1.13.0  # native esp8266 arduino toolchain build driver
 filelock==3.32.3  # inter-process locks (PlatformIO cache heal, git clone cache); >=3.32 for FileLock(fallback_to_soft=...), older versions silently drop the kwarg
 
 # esp-idf >= 5.0 requires this

tests/unit_tests/test_arduino8266_framework.py

@@ -0,0 +1,431 @@
+"""Tests for esphome.arduino8266.framework (downloads and environment)."""
+
+from __future__ import annotations
+
+from contextlib import contextmanager
+import os
+from pathlib import Path
+import subprocess
+import sys
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from esphome.arduino8266 import framework
+import esphome.config_validation as cv
+from esphome.core import CORE, EsphomeError
+
+
+@pytest.fixture(autouse=True)
+def _clear_caches(tmp_path: Path) -> None:
+    framework.ccache_path.cache_clear()
+    CORE.build_path = tmp_path
+
+
+def test_framework_package_version() -> None:
+    assert framework.framework_package_version(cv.Version(3, 1, 2)) == "3.30102.0"
+    assert framework.framework_package_version(cv.Version(3, 2, 0)) == "3.30200.0"
+
+
+def test_tools_path_default_and_prefix(tmp_path: Path) -> None:
+    with patch.dict(os.environ, {"ESPHOME_ARDUINO8266_PREFIX": str(tmp_path)}):
+        assert framework.get_arduino8266_tools_path() == tmp_path.resolve()
+    # A blank prefix must be treated as unset, not as the CWD
+    with patch.dict(os.environ, {"ESPHOME_ARDUINO8266_PREFIX": "  "}):
+        path = framework.get_arduino8266_tools_path()
+    assert path.name == "arduino8266"
+    assert path != Path.cwd()
+
+
+@pytest.mark.parametrize(
+    ("system", "machine", "expected"),
+    [
+        ("Darwin", "arm64", "darwin_arm64"),
+        ("Darwin", "x86_64", "darwin_x86_64"),
+        ("Windows", "AMD64", "windows_amd64"),
+        ("Windows", "ARM64", "windows_amd64"),
+        ("Windows", "x86", "windows_x86"),
+        ("Linux", "x86_64", "linux_x86_64"),
+        ("Linux", "aarch64", "linux_aarch64"),
+        ("Linux", "i686", "linux_i686"),
+        ("Linux", "armv7l", "linux_armv7l"),
+    ],
+)
+def test_pio_system(system: str, machine: str, expected: str) -> None:
+    with (
+        patch("platform.system", return_value=system),
+        patch("platform.machine", return_value=machine),
+    ):
+        assert framework._pio_system() == expected
+
+
+@pytest.mark.parametrize(
+    ("system", "machine"),
+    [
+        ("FreeBSD", "amd64"),
+        ("Linux", "ppc64le"),
+    ],
+)
+def test_pio_system_unsupported_host_raises(system: str, machine: str) -> None:
+    # Fails at resolution rather than installing a toolchain that can't run
+    with (
+        patch("platform.system", return_value=system),
+        patch("platform.machine", return_value=machine),
+        pytest.raises(EsphomeError, match="use 'toolchain: platformio'"),
+    ):
+        framework._pio_system()
+
+
+def _registry_response(files: list[dict]) -> MagicMock:
+    resp = MagicMock()
+    resp.json.return_value = {"versions": [{"name": "1.0.0", "files": files}]}
+    return resp
+
+
+def test_registry_download_network_error_is_clean_and_retried() -> None:
+    """Registry failures raise EsphomeError after retries, not a traceback."""
+    import requests
+
+    with (
+        patch("requests.get", side_effect=requests.ConnectionError("boom")) as mock_get,
+        pytest.raises(EsphomeError, match="Could not query the package registry"),
+    ):
+        framework._registry_download("pkg", "1.0.0")
+    assert mock_get.call_count == 3
+
+
+def test_registry_download_retries_transient_error() -> None:
+    import requests
+
+    resp = _registry_response(
+        [
+            {
+                "system": ["linux_x86_64"],
+                "download_url": "http://x/linux",
+                "checksum": {"sha256": "abc123"},
+                "size": 42,
+            }
+        ]
+    )
+    with (
+        patch("requests.get", side_effect=[requests.ConnectionError("boom"), resp]),
+        patch.object(framework, "_pio_system", return_value="linux_x86_64"),
+    ):
+        assert framework._registry_download("pkg", "1.0.0") == (
+            "http://x/linux",
+            "abc123",
+            42,
+        )
+
+
+def test_registry_download_matches_system() -> None:
+    resp = _registry_response(
+        [
+            {"system": ["windows_amd64"], "download_url": "http://x/win"},
+            {
+                "system": ["linux_x86_64"],
+                "download_url": "http://x/linux",
+                "checksum": {"sha256": "abc123"},
+                "size": 42,
+            },
+        ]
+    )
+    with (
+        patch("requests.get", return_value=resp),
+        patch.object(framework, "_pio_system", return_value="linux_x86_64"),
+    ):
+        assert framework._registry_download("pkg", "1.0.0") == (
+            "http://x/linux",
+            "abc123",
+            42,
+        )
+
+
+def test_registry_download_bare_string_system() -> None:
+    """A bare-string system tag is an exact match, not a substring test."""
+    resp = _registry_response(
+        [
+            {"system": "linux_x86", "download_url": "http://x/x86"},
+            {
+                "system": "linux_x86_64",
+                "download_url": "http://x/x86_64",
+                "checksum": {"sha256": "abc"},
+            },
+        ]
+    )
+    with (
+        patch("requests.get", return_value=resp),
+        patch.object(framework, "_pio_system", return_value="linux_x86_64"),
+    ):
+        assert framework._registry_download("pkg", "1.0.0")[0] == "http://x/x86_64"
+
+
+def test_registry_download_wildcard_system() -> None:
+    resp = _registry_response(
+        [
+            {
+                "system": "*",
+                "download_url": "http://x/any",
+                "checksum": {"sha256": "abc"},
+                "size": 7,
+            }
+        ]
+    )
+    with patch("requests.get", return_value=resp):
+        assert framework._registry_download("pkg", "1.0.0") == (
+            "http://x/any",
+            "abc",
+            7,
+        )
+
+
+def test_registry_download_missing_checksum_raises() -> None:
+    """An unverifiable archive is refused, never silently extracted."""
+    resp = _registry_response([{"system": "*", "download_url": "http://x/any"}])
+    with (
+        patch("requests.get", return_value=resp),
+        pytest.raises(EsphomeError, match="no sha256"),
+    ):
+        framework._registry_download("pkg", "1.0.0")
+
+
+def test_registry_download_no_system_match() -> None:
+    resp = _registry_response(
+        [{"system": ["windows_amd64"], "download_url": "http://x/win"}]
+    )
+    with (
+        patch("requests.get", return_value=resp),
+        patch.object(framework, "_pio_system", return_value="linux_x86_64"),
+        pytest.raises(EsphomeError, match="No pkg 1.0.0 build"),
+    ):
+        framework._registry_download("pkg", "1.0.0")
+
+
+def test_registry_download_version_not_found() -> None:
+    resp = _registry_response([])
+    resp.json.return_value = {"versions": [{"name": "2.0.0", "files": []}]}
+    with (
+        patch("requests.get", return_value=resp),
+        pytest.raises(EsphomeError, match="not found"),
+    ):
+        framework._registry_download("pkg", "1.0.0")
+
+
+def test_install_package_skips_when_marker_exists(tmp_path: Path) -> None:
+    dest = tmp_path / "pkg"
+    dest.mkdir()
+    (dest / ".esphome_extracted").touch()
+    with patch.object(framework, "download_from_mirrors") as mock_download:
+        framework._install_package("pkg", "1.0.0", dest, [])
+    mock_download.assert_not_called()
+
+
+def test_install_package_downloads_via_mirrors(tmp_path: Path) -> None:
+    dest = tmp_path / "pkg"
+    mirrors = ["http://mirror/{VERSION}/{SYSTEM}.tar.gz"]
+    with (
+        patch.object(framework, "download_from_mirrors") as mock_download,
+        patch.object(framework, "archive_extract_all") as mock_extract,
+        patch.object(framework, "_pio_system", return_value="linux_x86_64"),
+    ):
+        # Extraction is expected to create the directory
+        mock_extract.side_effect = lambda *_a, **_kw: dest.mkdir()
+        framework._install_package("pkg", "1.0.0", dest, mirrors)
+    assert mock_download.call_args[0][0] is mirrors
+    assert mock_download.call_args[0][1] == {
+        "VERSION": "1.0.0",
+        "SYSTEM": "linux_x86_64",
+    }
+    assert (dest / ".esphome_extracted").is_file()
+
+
+def test_install_package_downloads_via_registry(tmp_path: Path) -> None:
+    """The registry path downloads with the registry's sha256 and size."""
+    dest = tmp_path / "pkg"
+    with (
+        patch.object(framework, "download_with_resume") as mock_download,
+        patch.object(framework, "archive_extract_all") as mock_extract,
+        patch.object(
+            framework,
+            "_registry_download",
+            return_value=("http://x/pkg.tar.gz", "abc123", 42),
+        ),
+    ):
+        mock_extract.side_effect = lambda *_a, **_kw: dest.mkdir()
+        framework._install_package("pkg", "1.0.0", dest, [])
+    assert mock_download.call_args[0][0] == "http://x/pkg.tar.gz"
+    assert mock_download.call_args[1] == {"sha256": "abc123", "size": 42}
+
+
+def test_find_ninja_prefers_path(tmp_path: Path) -> None:
+    with patch("shutil.which", return_value=str(tmp_path / "ninja")):
+        assert framework._find_ninja() == tmp_path / "ninja"
+
+
+def test_find_ninja_falls_back_to_wheel(tmp_path: Path) -> None:
+    """Without a PATH entry, the ninja PyPI wheel's binary is used."""
+    binary_name = "ninja.exe" if os.name == "nt" else "ninja"
+    (tmp_path / binary_name).touch()
+    wheel = MagicMock(BIN_DIR=str(tmp_path))
+    with (
+        patch("shutil.which", return_value=None),
+        patch.dict(sys.modules, {"ninja": wheel}),
+    ):
+        assert framework._find_ninja() == tmp_path / binary_name
+
+
+def test_find_ninja_package_not_installed() -> None:
+    """A missing ninja package raises the actionable message, not ImportError."""
+    with (
+        patch("shutil.which", return_value=None),
+        patch.dict(sys.modules, {"ninja": None}),
+        pytest.raises(EsphomeError, match="ninja not found"),
+    ):
+        framework._find_ninja()
+
+
+def test_find_ninja_missing_everywhere(tmp_path: Path) -> None:
+    wheel = MagicMock(BIN_DIR=str(tmp_path))
+    with (
+        patch("shutil.which", return_value=None),
+        patch.dict(sys.modules, {"ninja": wheel}),
+        pytest.raises(EsphomeError, match="ninja not found"),
+    ):
+        framework._find_ninja()
+
+
+def test_check_and_install_returns_paths(tmp_path: Path) -> None:
+    with (
+        patch.dict(os.environ, {"ESPHOME_ARDUINO8266_PREFIX": str(tmp_path)}),
+        patch.object(framework, "_install_package") as mock_install,
+        patch.object(framework, "_find_ninja", return_value=tmp_path / "ninja"),
+    ):
+        paths = framework.check_and_install(cv.Version(3, 1, 2))
+    assert paths["framework_path"] == tmp_path / "frameworks" / "3.30102.0"
+    assert (
+        paths["toolchain_path"] == tmp_path / "toolchains" / framework.TOOLCHAIN_VERSION
+    )
+    assert paths["ninja_path"] == tmp_path / "ninja"
+    assert mock_install.call_count == 2
+    # The layout checks cover the directories write_project needs, including
+    # the bundled libraries/ tree
+    fw_expect = mock_install.call_args_list[0].kwargs["expect"]
+    assert fw_expect == ("cores/esp8266", "tools/sdk", "libraries")
+    assert mock_install.call_args_list[1].kwargs["expect"] == ("bin",)
+
+
+def test_get_build_env_prepends_toolchain_bin(tmp_path: Path) -> None:
+    with patch.object(framework, "ccache_env", return_value={"CCACHE_DIR": "x"}):
+        env = framework.get_build_env(tmp_path)
+    assert env["PATH"].startswith(str(tmp_path / "bin") + os.pathsep)
+    assert env["CCACHE_DIR"] == "x"
+
+
+def test_ccache_path_disabled_by_env() -> None:
+    with patch.dict(os.environ, {"ESPHOME_CCACHE_ENABLE": "0"}):
+        assert framework.ccache_path() is None
+
+
+def test_ccache_path_no_binary(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.delenv("ESPHOME_CCACHE_ENABLE", raising=False)
+    with patch("shutil.which", return_value=None):
+        assert framework.ccache_path() is None
+
+
+def test_ccache_path_probe_failure(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.delenv("ESPHOME_CCACHE_ENABLE", raising=False)
+    with (
+        patch("shutil.which", return_value="/usr/bin/ccache"),
+        patch("subprocess.run", side_effect=subprocess.SubprocessError),
+    ):
+        assert framework.ccache_path() is None
+
+
+def test_ccache_path_ok(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.delenv("ESPHOME_CCACHE_ENABLE", raising=False)
+    with (
+        patch("shutil.which", return_value="/usr/bin/ccache"),
+        patch("subprocess.run"),
+    ):
+        assert framework.ccache_path() == "/usr/bin/ccache"
+
+
+def test_ccache_path_explicit_missing_binary_warns(
+    monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
+) -> None:
+    monkeypatch.setenv("ESPHOME_CCACHE_ENABLE", "1")
+    with patch("shutil.which", return_value=None):
+        assert framework.ccache_path() is None
+    assert "no ccache binary is on PATH" in caplog.text
+
+
+def test_ccache_path_explicit_skips_probe(monkeypatch: pytest.MonkeyPatch) -> None:
+    """An explicit opt-in trusts the binary without the runnability probe."""
+    monkeypatch.setenv("ESPHOME_CCACHE_ENABLE", "1")
+    with (
+        patch("shutil.which", return_value="/usr/bin/ccache"),
+        patch("esphome.platformio.toolchain._ccache_runs", side_effect=AssertionError),
+    ):
+        assert framework.ccache_path() == "/usr/bin/ccache"
+
+
+def test_ccache_env(tmp_path: Path) -> None:
+    with patch.object(framework, "ccache_path", return_value=None):
+        assert framework.ccache_env() == {}
+    with (
+        patch.object(framework, "ccache_path", return_value="/usr/bin/ccache"),
+        patch.dict(os.environ, {"CCACHE_NOHASHDIR": "false"}),
+    ):
+        env = framework.ccache_env()
+    # User-set values are respected; the rest get defaults
+    assert "CCACHE_NOHASHDIR" not in env
+    assert env["CCACHE_DEPEND"] == "1"
+    assert env["CCACHE_BASEDIR"] == str(Path(CORE.build_path).resolve())
+    assert env["CCACHE_DIR"].endswith("ccache")
+
+
+def test_install_package_validates_expected_layout(tmp_path: Path) -> None:
+    """The success marker is only written when the extracted tree is usable."""
+    dest = tmp_path / "pkg"
+    with (
+        patch.object(framework, "download_from_mirrors"),
+        patch.object(framework, "archive_extract_all") as mock_extract,
+        patch.object(framework, "_pio_system", return_value="linux_x86_64"),
+    ):
+        mock_extract.side_effect = lambda *_a, **_kw: (dest / "bin").mkdir(parents=True)
+        framework._install_package("pkg", "1.0.0", dest, ["http://m"], expect=("bin",))
+    assert (dest / ".esphome_extracted").is_file()
+
+
+def test_install_package_unexpected_layout_raises(tmp_path: Path) -> None:
+    dest = tmp_path / "pkg"
+    with (
+        patch.object(framework, "download_from_mirrors"),
+        patch.object(framework, "archive_extract_all") as mock_extract,
+        patch.object(framework, "_pio_system", return_value="linux_x86_64"),
+        pytest.raises(EsphomeError, match="without the expected bin"),
+    ):
+        mock_extract.side_effect = lambda *_a, **_kw: dest.mkdir()
+        framework._install_package("pkg", "1.0.0", dest, ["http://m"], expect=("bin",))
+    assert not (dest / ".esphome_extracted").exists()
+
+
+def test_install_package_marker_rechecked_under_lock(tmp_path: Path) -> None:
+    """A concurrent install finishing while we wait for the lock is detected."""
+    dest = tmp_path / "pkg"
+    marker = dest / ".esphome_extracted"
+
+    @contextmanager
+    def _fake_lock(*_a, **_kw):
+        dest.mkdir(parents=True, exist_ok=True)
+        marker.touch()
+        yield
+
+    with (
+        patch("filelock.FileLock", _fake_lock),
+        patch.object(framework, "download_from_mirrors") as mock_download,
+        patch.object(framework, "rmdir") as mock_rmdir,
+    ):
+        framework._install_package("pkg", "1.0.0", dest, ["http://m"])
+    mock_download.assert_not_called()
+    mock_rmdir.assert_not_called()