From 3e776547190d5fb2151cbde8156fb2b804ce91a3 Mon Sep 17 00:00:00 2001 From: "J. Nick Koston" Date: Thu, 20 Aug 2026 12:58:06 -0500 Subject: [PATCH] Add the ESP8266 Arduino framework and toolchain installer --- esphome/arduino8266/__init__.py | 9 + esphome/arduino8266/framework.py | 335 ++++++++++++++ esphome/writer.py | 8 +- requirements.txt | 1 + .../unit_tests/test_arduino8266_framework.py | 431 ++++++++++++++++++ 5 files changed, 783 insertions(+), 1 deletion(-) create mode 100644 esphome/arduino8266/__init__.py create mode 100644 esphome/arduino8266/framework.py create mode 100644 tests/unit_tests/test_arduino8266_framework.py diff --git a/esphome/arduino8266/__init__.py b/esphome/arduino8266/__init__.py new file mode 100644 index 0000000000..8f403a8553 --- /dev/null +++ b/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. +""" diff --git a/esphome/arduino8266/framework.py b/esphome/arduino8266/framework.py new file mode 100644 index 0000000000..066667806c --- /dev/null +++ b/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``): + + /arduino8266/frameworks// framework-arduinoespressif8266 + /arduino8266/toolchains// 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} diff --git a/esphome/writer.py b/esphome/writer.py index 866377d2f5..26d091a823 100644 --- a/esphome/writer.py +++ b/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) diff --git a/requirements.txt b/requirements.txt index 740a8c1a79..04844f67dc 100644 --- a/requirements.txt +++ b/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 diff --git a/tests/unit_tests/test_arduino8266_framework.py b/tests/unit_tests/test_arduino8266_framework.py new file mode 100644 index 0000000000..2e72dfb26a --- /dev/null +++ b/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()