From 2104096f02b313d7dfdd4eedd6d91ff5b1898838 Mon Sep 17 00:00:00 2001 From: "J. Nick Koston" <3060199+jesserockz@users.noreply.github.com> Date: Wed, 26 Aug 2026 21:42:30 -0500 Subject: [PATCH] [core] Prefetch PlatformIO packages in parallel (#18769) --- esphome/espidf/framework.py | 28 +- esphome/framework_helpers.py | 44 +- esphome/platformio/library.py | 13 +- esphome/platformio/prefetch.py | 600 +++++++++++ esphome/platformio/toolchain.py | 17 +- tests/unit_tests/test_espidf_framework.py | 20 +- tests/unit_tests/test_framework_helpers.py | 41 + tests/unit_tests/test_platformio_prefetch.py | 985 ++++++++++++++++++ tests/unit_tests/test_platformio_toolchain.py | 14 +- 9 files changed, 1713 insertions(+), 49 deletions(-) create mode 100644 esphome/platformio/prefetch.py create mode 100644 tests/unit_tests/test_platformio_prefetch.py diff --git a/esphome/espidf/framework.py b/esphome/espidf/framework.py index 239d874dbd..6c2a285360 100644 --- a/esphome/espidf/framework.py +++ b/esphome/espidf/framework.py @@ -2,7 +2,6 @@ from collections.abc import Callable from ctypes.util import find_library -from functools import partial import json import logging import os @@ -24,16 +23,17 @@ from esphome.framework_helpers import ( create_venv, download_and_extract, download_from_mirrors, - download_with_resume, failure_reason, get_python_env_executable_path, get_system_python_path, + resume_fetch_job, rmdir, run_batch_downloads, run_command, run_command_ok, str_to_lst_of_str, tool_version_runs, + warn_prefetch_failures, ) from esphome.helpers import write_file_if_changed @@ -686,18 +686,6 @@ def _patch_tools_json_demote_unused_tools(framework_path: Path) -> None: ) -def _download_tool( - dist_path: Path, entry: dict, tracker: Callable[[int], None] -) -> None: - download_with_resume( - entry["url"], - dist_path / entry["dest"], - sha256=entry["sha256"], - size=entry["size"], - progress=tracker, - ) - - def _prefetch_idf_tool_archives( framework_path: Path, targets_str: str, @@ -775,15 +763,17 @@ def _prefetch_idf_tool_archives( ( entry["name"], entry["size"], - partial(_download_tool, dist_path, entry), + resume_fetch_job( + entry["url"], + dist_path / entry["dest"], + sha256=entry["sha256"], + size=entry["size"], + ), ) for entry in entries ], ) - for name, e in failures: - # failure_reason: a message-less exception must not log blank - _LOGGER.warning("Could not prefetch %s: %s", name, failure_reason(e)) - _LOGGER.debug("Prefetch failure detail", exc_info=e) + warn_prefetch_failures(failures) if len(failures) == len(entries): # A systematic fault, not one flaky mirror: the resume # workaround (#17703) is off for this whole install diff --git a/esphome/framework_helpers.py b/esphome/framework_helpers.py index aab7acc0e8..82bc0d3727 100644 --- a/esphome/framework_helpers.py +++ b/esphome/framework_helpers.py @@ -701,7 +701,7 @@ def _write_download_meta( _LOGGER.debug("Could not update download metadata %s: %s", meta, e) -def _content_length(resp: "requests.Response") -> int: +def content_length(resp: "requests.Response") -> int: """Return the response's Content-Length, or 0 when absent or malformed. 0 means "unknown", which downstream disables the progress bar and the @@ -744,7 +744,7 @@ def _stream_response_to_file( """ f.seek(offset) f.truncate(offset) - total_size = size or offset + _content_length(resp) + total_size = size or offset + content_length(resp) downloaded = offset own_bar: ProgressBar | None = None if progress is None: @@ -909,6 +909,19 @@ def _part_path(dest: Path) -> Path: return dest.with_name(dest.name + ".part") +def discard_partial_download(dest: Path) -> None: + """Remove ``dest`` and the resume sidecars of an abandoned download.""" + part = _part_path(dest) + for stale in (dest, part, part.with_name(part.name + ".meta")): + try: + stale.unlink() + except FileNotFoundError: + continue + except OSError as err: + # The caller's cache is never pruned; leave a trace + _LOGGER.debug("Could not remove %s: %s", stale, err) + + def _cancellable_sleep( delay: float, progress: Callable[[int], None] | None, done: int ) -> None: @@ -922,6 +935,31 @@ def _cancellable_sleep( time.sleep(min(0.5, remaining)) +def resume_fetch_job( + url: str, dest: PathType, **kwargs +) -> Callable[[Callable[[int], None]], None]: + """A ``run_batch_downloads`` job callable wrapping ``download_with_resume``. + + Forwards the runner's positional tracker as the ``progress`` keyword. + """ + + def fetch(tracker: Callable[[int], None]) -> None: + download_with_resume(url, dest, progress=tracker, **kwargs) + + return fetch + + +def warn_prefetch_failures( + failures: list[tuple[str, BaseException]], + message: str = "Could not prefetch %s: %s", +) -> None: + """Warn per failed batch-prefetch job; the caller's installer retries them.""" + for name, err in failures: + # failure_reason: a message-less exception must not log blank + _LOGGER.warning(message, name, failure_reason(err)) + _LOGGER.debug("Prefetch failure detail", exc_info=err) + + def download_with_resume( url: str, dest: PathType, @@ -1022,7 +1060,7 @@ def download_with_resume( streamed = True if offset == 0: validator = _response_validator(resp) - expected_total = _content_length(resp) + expected_total = content_length(resp) # Recorded so a later run can prove an If-Range # resume of this part file safe. _write_download_meta(meta, url, validator, expected_total) diff --git a/esphome/platformio/library.py b/esphome/platformio/library.py index 1792647d6b..306f07854e 100644 --- a/esphome/platformio/library.py +++ b/esphome/platformio/library.py @@ -35,6 +35,7 @@ from esphome.framework_helpers import ( failure_reason, rmdir, run_batch_downloads, + warn_prefetch_failures, ) _LOGGER = logging.getLogger(__name__) @@ -977,14 +978,10 @@ def _prefetch_wave( for c in components ], ) - for name, err in failures: - # The sequential call below retries and raises the real error - _LOGGER.warning( - "Prefetch of %s failed (retrying sequentially): %s", - name, - failure_reason(err), - ) - _LOGGER.debug("Prefetch failure detail", exc_info=err) + # The sequential call below retries and raises the real error + warn_prefetch_failures( + failures, "Prefetch of %s failed (retrying sequentially): %s" + ) except Exception as err: # noqa: BLE001 # pylint: disable=broad-exception-caught # Same policy as the ESP-IDF twin: the prefetch must never become a # new way for the build to fail diff --git a/esphome/platformio/prefetch.py b/esphome/platformio/prefetch.py new file mode 100644 index 0000000000..f313c2f4d0 --- /dev/null +++ b/esphome/platformio/prefetch.py @@ -0,0 +1,600 @@ +"""Parallel prefetch of the packages a PlatformIO run would install. + +Downloads the archives concurrently into PlatformIO's own download cache +(identical ``compute_download_path`` keys) so the serial installer finds +them already cached. Runs in a subprocess like all PlatformIO execution: +loading a platform executes its code (pioarduino's penv setup rewrites +``sys.path``). A sentinel in the build dir lets warm builds skip the +spawn. Best-effort: any failure logs and PlatformIO downloads as before. +Across processes sharing a core dir every download destination is +serialized by a file lock; checksum-less URL downloads additionally +stage under a stable name and promote with an atomic rename. +""" + +from __future__ import annotations + +from concurrent.futures import ThreadPoolExecutor +import hashlib +import json +import logging +import os +from pathlib import Path +import subprocess +import sys +import threading +import time +from typing import Any + +from esphome.framework_helpers import ( + content_length, + discard_partial_download, + failure_reason, + resume_fetch_job, + run_batch_downloads, + warn_prefetch_failures, +) +from esphome.helpers import get_bool_env + +_LOGGER = logging.getLogger(__name__) + +# Concurrent registry resolutions / HEAD probes (each is network-bound) +_RESOLVE_WORKERS = 8 + +# A hung child must not block the build; downloads resume on the next run +_PREFETCH_TIMEOUT = 20 * 60 + +# Waiting on another process's URL download; past this, leave it to pio +_DOWNLOAD_LOCK_TIMEOUT = 60 + +# Child exit for a handled, already-warned failure; 1 would collide with +# the interpreter's own import-failure exit +_EXIT_HANDLED = 3 + +# Short lock-acquire slices so a waiting worker still observes Ctrl-C +_URI_LOCK_POLL = 1 + +# Resolution errored (vs a clean skip); suppresses the warm sentinel +_RESOLVE_FAILED = object() + + +def _sweep_stale_sidecars(download_dir: Path, expire_seconds: int) -> None: + """Prune resume sidecars pio's usage.db pruner cannot see. + + A version bump strands an aborted archive's sidecars forever. Lock + files stay: a held lock can carry an ancient mtime (O_TRUNC keeps + it), and unlinking one reopens the single-writer hole it guards. + """ + cutoff = time.time() - expire_seconds + try: + for f in download_dir.iterdir(): + if f.suffix not in (".part", ".meta", ".prefetch"): + continue + try: + if f.stat().st_mtime < cutoff: + f.unlink() + except OSError as err: + _LOGGER.debug("Could not remove %s: %s", f, err) + except OSError: + _LOGGER.debug("Could not sweep %s", download_dir, exc_info=True) + + +# Child records a no-work run; the parent skips the next spawn while valid +_SENTINEL_NAME = ".esphome_prefetch.json" +_SENTINEL_SCHEMA = 1 + + +def _ini_sha256(build_dir: Path) -> str: + return hashlib.sha256((build_dir / "platformio.ini").read_bytes()).hexdigest() + + +def _sentinel_state(build_dir: Path) -> dict[str, Any]: + """The environment fingerprint a sentinel must match to stay valid.""" + # Same fingerprint as the heal stamp: the sentinel's dirs die with its wipe + from esphome.platformio.toolchain import current_python_minor + + return { + "schema": _SENTINEL_SCHEMA, + "ini_sha256": _ini_sha256(build_dir), + "python": current_python_minor(), + "core_dir_env": os.environ.get("PLATFORMIO_CORE_DIR", ""), + } + + +def _prefetch_is_warm(build_dir: Path) -> bool: + """Whether the last prefetch found nothing to do and nothing changed since.""" + try: + data = json.loads((build_dir / _SENTINEL_NAME).read_text(encoding="utf-8")) + dirs = data.pop("dirs") + return ( + data == _sentinel_state(build_dir) + and bool(dirs) + and all(Path(d).is_dir() for d in dirs) + ) + except FileNotFoundError: + return False + except (OSError, ValueError, KeyError, AttributeError, TypeError): + _LOGGER.debug("Ignoring invalid prefetch sentinel", exc_info=True) + return False + + +def prefetch_platformio_packages() -> None: + """Warm PlatformIO's download cache for the current project, in parallel.""" + from esphome.core import CORE + from esphome.platformio.toolchain import ( + default_libdeps_dir, + heal_platformio_python_env, + ) + + # Heal first: its Python-version wipe would discard freshly warmed + # caches and the sentinel's dirs (the later heal call is a no-op) + heal_platformio_python_env() + build_dir = Path(CORE.build_path) + if _prefetch_is_warm(build_dir): + return + # The child is esphome itself: PYTHONPATH stays so it imports this + # tree's esphome (tests/integration pins the source tree through it) + env = dict(os.environ) + # Must match run_platformio_cli's default or warm builds re-resolve + # every library + env.setdefault("PLATFORMIO_LIBDEPS_DIR", default_libdeps_dir()) + # -v/-vv must reach the child's debug logging or the swallowed + # failure detail is undiagnosable in the field + env["ESPHOME_PREFETCH_LOG_LEVEL"] = str(logging.getLogger().getEffectiveLevel()) + if CORE.dashboard: + # The child's progress bar and log escaping key off CORE.dashboard + env["ESPHOME_PREFETCH_DASHBOARD"] = "1" + cmd = [ + sys.executable, + "-m", + "esphome.platformio.prefetch", + str(build_dir), + CORE.name, + ] + try: + proc = subprocess.run(cmd, env=env, check=False, timeout=_PREFETCH_TIMEOUT) + except subprocess.TimeoutExpired: + _LOGGER.warning("PlatformIO package prefetch timed out; continuing without it") + return + except Exception as err: # noqa: BLE001 # pylint: disable=broad-exception-caught + # The prefetch must never become a new way for the build to fail + _LOGGER.warning("PlatformIO package prefetch skipped: %s", failure_reason(err)) + _LOGGER.debug("Prefetch failure detail", exc_info=True) + return + if proc.returncode == _EXIT_HANDLED: + # The child already warned with the reason; a second line is noise + _LOGGER.debug("Prefetch child reported a handled failure") + elif proc.returncode != 0: + # Exit 1 stays here: the interpreter exits 1 for import/module + # failures before main() ever runs, a wiring break worth a warning + _LOGGER.warning( + "PlatformIO package prefetch skipped (exit %d)", proc.returncode + ) + + +def _project_platform_and_config(ini: Path, env: str) -> tuple[str | None, Any]: + """The env's platform spec and the ProjectConfig for the given ini.""" + from platformio import app + from platformio.project.config import ProjectConfig + + # PlatformBase.config reads the default ProjectConfig; it must see + # this ini's env options + app.set_session_var("custom_project_conf", str(ini)) + config = ProjectConfig.get_instance(str(ini)) + return config.get(f"env:{env}", "platform", None), config + + +def _registry_jobs( + manager, specs, seen: set[str] +) -> tuple[list[tuple[str, int, Any]], int]: + """Resolve registry specs to ``(name, size, fetch)`` batch jobs. + + Mirrors PlatformIO's install path: best version, systype file, first + mirror, and the same sha1(url + checksum) download-cache key. Also + returns how many resolutions errored (a clean skip is not an error). + """ + from platformio.registry.mirror import RegistryFileMirrorIterator + + local = threading.local() + errors: list[str] = [] + + def _resolve(spec) -> tuple[str, int, str, Path, str] | object | None: + # One manager (and registry HTTP session) per worker thread; + # installed-state was already checked on the shared manager + if (mgr := getattr(local, "mgr", None)) is None: + mgr = local.mgr = manager.__class__() + try: + packages = mgr.search_registry_packages(spec) + if not packages: + _LOGGER.debug("%s is unknown to the registry", spec) + return None # let PlatformIO report it + package, version = mgr.find_best_registry_version(packages, spec) + if not package or not version: + _LOGGER.debug("%s has no matching registry version", spec) + return None + pkgfile = mgr.pick_compatible_pkg_file(version["files"]) + if not pkgfile: + _LOGGER.debug("%s has no file for this systype", spec) + return None + url, checksum = next(RegistryFileMirrorIterator(pkgfile["download_url"])) + checksum = checksum or pkgfile["checksum"]["sha256"] + dl_path = Path(mgr.compute_download_path(url, checksum)) + if dl_path.is_file(): + return None # cached from an earlier run + size = pkgfile.get("size") + if not size: + _LOGGER.debug("%s has no size; PlatformIO fetches it", spec) + return None # no size, no bar share + return f"{package['name']}@{version['name']}", size, url, dl_path, checksum + except Exception as err: # noqa: BLE001 # pylint: disable=broad-exception-caught + # One flaky spec must not discard the rest of the batch + _LOGGER.debug("Could not resolve %s", spec, exc_info=True) + errors.append(failure_reason(err)) + return _RESOLVE_FAILED + + # Serial disk lookups on the shared manager: a fully warm build + # resolves nothing, and duplicate specs resolve once + unique: dict[tuple[str | None, str, str], Any] = {} + for s in specs: + if not s.uri and not manager.get_package(s): + unique.setdefault((s.owner, s.name, str(s.requirements)), s) + pending = list(unique.values()) + if not pending: + return [], 0 + # Serial resolutions (registry GET + mirror HEAD each) dominate + with ThreadPoolExecutor(max_workers=min(_RESOLVE_WORKERS, len(pending))) as ex: + results = list(ex.map(_resolve, pending)) + jobs: list[tuple[str, int, Any]] = [] + for res in results: + if res is None or res is _RESOLVE_FAILED: + continue + name, size, url, dl_path, checksum = res + if str(dl_path) in seen: + continue # duplicate spec; two workers must not share a .part + seen.add(str(dl_path)) + jobs.append( + (name, size, _registry_fetch_job(manager, url, dl_path, checksum, size)) + ) + if failed := len(errors): + # Visible once per build, naming a cause so an API break does not + # read as an outage; per-spec detail stays at debug + _LOGGER.warning( + "Could not resolve %d of %d PlatformIO package(s) (%s); " + "PlatformIO will download them serially", + failed, + len(pending), + errors[0], + ) + return jobs, failed + + +def _uri_jobs(manager, specs, seen: set[str]) -> tuple[list[tuple[str, int, Any]], int]: + """Jobs for direct-URL specs; a HEAD sizes each for the combined bar. + + Also returns how many HEAD probes errored (an absent length is not an + error). + """ + from esphome.net_retry import fetch_with_retry, http_request + + candidates: list[tuple[str, str, Path]] = [] + for spec in specs: + url = spec.uri + if not url or not url.startswith(("http://", "https://")): + continue # git+/file specs are cloned/copied, not downloaded + if url.split("#", 1)[0].endswith(".git"): + continue # bare-URL VCS spec; PlatformIO clones it + if manager.get_package(spec): + continue + # PlatformIO downloads URL specs with no checksum + dl_path = Path(manager.compute_download_path(url, "")) + if dl_path.is_file() or str(dl_path) in seen: + continue # cached, or another spec already claimed this .part + seen.add(str(dl_path)) + candidates.append((spec.name, url, dl_path)) + + errors: list[str] = [] + + def _head_size(url: str) -> int: + try: + resp = fetch_with_retry(url, lambda: http_request("HEAD", url, timeout=30)) + except Exception as err: # noqa: BLE001 # pylint: disable=broad-exception-caught + _LOGGER.debug("HEAD %s failed", url, exc_info=True) + errors.append(failure_reason(err)) + return -1 + if not resp.ok: + # An error page's Content-Length is not a download size + _LOGGER.debug("HEAD %s returned %s", url, resp.status_code) + if resp.status_code in (401, 403, 408, 429) or resp.status_code >= 500: + # 401/403 included: registries rate-limit with them + errors.append(f"HTTP {resp.status_code}") + return -1 # transient; must not be cached as warm + # Permanent (405/501 HEAD-unsupported, 401/403/404): a clean + # skip so the warm sentinel is not disabled forever; pio run + # surfaces a genuinely broken URL when it downloads + return 0 + return content_length(resp) + + if not candidates: + return [], 0 + with ThreadPoolExecutor(max_workers=min(_RESOLVE_WORKERS, len(candidates))) as ex: + sizes = list(ex.map(_head_size, [url for _, url, _ in candidates])) + jobs: list[tuple[str, int, Any]] = [] + failed = 0 + for (name, url, dl_path), size in zip(candidates, sizes, strict=True): + if size < 0: + failed += 1 + elif size: + jobs.append((name, size, _uri_fetch_job(manager, url, dl_path, size))) + else: + # Missing or unusable Content-Length; visible under -v + _LOGGER.debug("%s reports no usable length; PlatformIO fetches it", url) + if failed: + _LOGGER.warning( + "Could not size %d of %d PlatformIO package URL(s) (%s); " + "PlatformIO will download them serially", + failed, + len(candidates), + errors[0], + ) + return jobs, failed + + +def _serialized_fetch_job( + dl_path: Path, lock_path: str, body: Any, unlocked_ok: bool = True +) -> Any: + """Wrap ``body`` so the shared destination is single-writer. + + Interleaved writers truncate each other's ``.part`` bytes (see + registry.py). The bounded poll observes Ctrl-C via the tracker; a + blown deadline is a clean skip (the holder's copy is what the build + needs). On a lock-less filesystem a sha256-verified body runs + unlocked with one warning; a checksum-less one + (``unlocked_ok=False``) is a counted failure instead. + """ + + def run(tracker: Any) -> None: + from filelock import FileLock, Timeout + + # fallback_to_soft would leave a stale marker on lock-less + # filesystems that blocks every later build (see git.py) + lock = FileLock(lock_path, fallback_to_soft=False) + deadline = time.monotonic() + _DOWNLOAD_LOCK_TIMEOUT + while True: + try: + lock.acquire(timeout=_URI_LOCK_POLL) + break + except Timeout: + tracker(0) # raises when the batch is cancelled + if time.monotonic() >= deadline: + # Another process is fetching this same file; its copy + # is what the build needs (a large framework archive + # can hold the lock far longer than this deadline) + _LOGGER.debug("Leaving %s to its current downloader", dl_path.name) + return + except OSError as err: + if not unlocked_ok: + # A body with no checksum to catch interleaved corruption + raise + lock = None + _LOGGER.warning( + "Could not lock %s (%s); downloading unlocked", + dl_path.name, + err, + ) + break + try: + if dl_path.is_file(): + return # another process finished it while we waited + body(tracker) + finally: + if lock is not None: + lock.release() + + return run + + +# usage.db is a whole-file rewrite behind pio's self-unlinking LockFile; +# concurrent writers could reset every recorded entry +_REGISTER_LOCK = threading.Lock() + + +def _register_download(manager: Any, dl_path: Path) -> None: + """Hand the archive to pio's usage.db pruner; an unregistered one is + never expired (disk garbage, never a bad build).""" + try: + with _REGISTER_LOCK: + manager.set_download_utime(str(dl_path)) + except Exception as err: # noqa: BLE001 # pylint: disable=broad-exception-caught + _LOGGER.debug("Could not register %s with pio's cache: %s", dl_path, err) + + +def _registry_fetch_job( + manager: Any, url: str, dl_path: Path, checksum: str, size: int +) -> Any: + """A locked fetch straight to the cache path; sha256 verifies it.""" + # .esphome.lock: pio's own LockFile(dl_path) owns .lock and + # deletes it on release, which would unlink a held filelock + fetch = _serialized_fetch_job( + dl_path, + f"{dl_path}.esphome.lock", + resume_fetch_job(url, dl_path, sha256=checksum, size=size), + ) + + def run(tracker: Any) -> None: + fetch(tracker) + if dl_path.is_file(): + # The deadline skip can end with no archive landed + _register_download(manager, dl_path) + + return run + + +def _uri_fetch_job(manager: Any, url: str, dl_path: Path, size: int) -> Any: + """Fetch to a locked staging path, then rename into the cache. + + The stable staging name keeps resume working across interrupted + runs; the rename makes the promotion atomic. + """ + tmp = dl_path.with_name(f"{dl_path.name}.prefetch") + # attempts=2: the size is only a HEAD probe's word, and a HEAD/GET + # disagreement would otherwise re-download the archive five times + fetch = resume_fetch_job(url, tmp, size=size, attempts=2) + + def promote(tracker: Any) -> None: + fetch(tracker) + if (actual := tmp.stat().st_size) != size: + # A wrong-length checksum-less body must never be published + discard_partial_download(tmp) + raise ValueError(f"expected {size} bytes, fetched {actual}") + tmp.replace(dl_path) + + def run(tracker: Any) -> None: + _serialized_fetch_job(dl_path, f"{tmp}.lock", promote, unlocked_ok=False)( + tracker + ) + if dl_path.is_file(): + # Won or lost, the race is over; staging files left behind + # are dead weight PlatformIO's cache never prunes + discard_partial_download(tmp) + _register_download(manager, dl_path) + + return run + + +def _prefetch(build_dir: Path, env: str) -> None: + from platformio.dependencies import get_core_dependencies + from platformio.package.manager.library import LibraryPackageManager + from platformio.package.manager.platform import PlatformPackageManager + from platformio.package.meta import PackageSpec + from platformio.platform.factory import PlatformFactory + + platform_spec, config = _project_platform_and_config( + build_dir / "platformio.ini", env + ) + if not platform_spec: + # An env mismatch must not disable the feature with no trace + _LOGGER.debug( + "No platform for env %s in %s; nothing to prefetch", env, build_dir + ) + return + + # The platform (manifest plus build scripts) installs first and + # resolves the rest. Its setup may rewrite sys.path (pioarduino's penv + # setup does); restore it so later imports here still resolve. + saved_sys_path = list(sys.path) + pm = PlatformPackageManager() + _sweep_stale_sidecars(Path(pm.get_download_dir()), pm.DOWNLOAD_CACHE_EXPIRE) + pkg = pm.install(platform_spec, skip_dependencies=True) + p = PlatformFactory.new(pkg) + p.configure_project_packages(env, ["run"]) + sys.path[:] = saved_sys_path + + specs = [ + p.get_package_spec(name) + for name, opts in p.packages.items() + if not opts.get("optional") + ] + # PIO's build engine installs outside the platform package list; + # skipped when the platform lists it itself + if not any(s.name == "tool-scons" for s in specs): + specs.append( + PackageSpec( + owner="platformio", + name="tool-scons", + requirements=get_core_dependencies()["tool-scons"], + ) + ) + lib_deps = config.get(f"env:{env}", "lib_deps", []) + # pio run's storage dir for this env: installed libraries skip by + # disk lookup + libdeps_dir = Path(config.get("platformio", "libdeps_dir")) / env + lm = LibraryPackageManager(str(libdeps_dir)) + # A bare name is usually a framework built-in (WiFi, SPI); with no + # lib builders here to tell built-in from registry, skip it. The only + # cost is that an owner-less user library is not prefetched + lib_specs = [ + spec + for dep in lib_deps + if dep and not dep.startswith("$") + if (spec := PackageSpec(dep)).external or spec.owner + ] + + seen: set[str] = set() + jobs: list[tuple[str, int, Any]] = [] + unresolved = 0 + for mgr, batch in ((p.pm, specs), (lm, lib_specs)): + for build_jobs in (_registry_jobs, _uri_jobs): + batch_jobs, failed = build_jobs(mgr, batch, seen) + jobs += batch_jobs + unresolved += failed + + sentinel = build_dir / _SENTINEL_NAME + if not jobs: + if not unresolved: + # Record the no-work run so the parent skips the next spawn. + # A failed resolution is not "no work": a registry outage must + # not be cached as warm. + dirs = [config.get("platformio", "packages_dir")] + if lib_specs: + dirs.append(str(libdeps_dir)) + sentinel.write_text( + json.dumps({**_sentinel_state(build_dir), "dirs": dirs}), + encoding="utf-8", + ) + return + sentinel.unlink(missing_ok=True) + _LOGGER.info( + "Prefetching %d PlatformIO package(s): %s", + len(jobs), + ", ".join(name for name, _, _ in jobs), + ) + # PlatformIO retries failed packages itself, without resume + warn_prefetch_failures(run_batch_downloads("Downloading PlatformIO packages", jobs)) + + +def main(argv: list[str]) -> int: + """Subprocess entry point: ``prefetch ``.""" + from esphome.core import CORE + from esphome.log import setup_log + + raw_level = os.environ.get("ESPHOME_PREFETCH_LOG_LEVEL") + try: + level = int(raw_level) if raw_level is not None else logging.INFO + except ValueError: + level = logging.INFO + # Mirror the parent's log setup: warnings keep their level prefix and + # color, and the download bar still draws under the dashboard + CORE.dashboard = get_bool_env("ESPHOME_PREFETCH_DASHBOARD") + setup_log(level) + # pio's managers attach their own handler and still propagate; without + # this every manager line also prints through the root handler. Their + # construction re-pins the logger to INFO, so a logger-level filter + # (which survives pio's handler reset) enforces a quiet level instead. + for cls_name in ( + "ToolPackageManager", + "LibraryPackageManager", + "PlatformPackageManager", + ): + manager_logger = logging.getLogger(cls_name.replace("Package", " ")) + manager_logger.propagate = False + manager_logger.addFilter(lambda record: record.levelno >= level) + if len(argv) != 2: + # A wiring bug, not a network failure; make it distinguishable + _LOGGER.warning("prefetch usage: ") + return 2 + build_dir, env = argv + try: + _prefetch(Path(build_dir), env) + except KeyboardInterrupt: + # Shared process group: exit quietly, no traceback on the terminal + _LOGGER.debug("Prefetch interrupted", exc_info=True) + return 130 + except Exception as err: # noqa: BLE001 # pylint: disable=broad-exception-caught + # The parent treats any exit as warn-and-continue, never a failure + _LOGGER.warning("PlatformIO package prefetch skipped: %s", failure_reason(err)) + _LOGGER.debug("Prefetch failure detail", exc_info=True) + return _EXIT_HANDLED + return 0 + + +if __name__ == "__main__": # pragma: no cover + sys.exit(main(sys.argv[1:])) diff --git a/esphome/platformio/toolchain.py b/esphome/platformio/toolchain.py index cf2094dfe0..97b32420da 100644 --- a/esphome/platformio/toolchain.py +++ b/esphome/platformio/toolchain.py @@ -96,7 +96,7 @@ def _clean_platformio_python_env(config: "ProjectConfig", core_dir: Path) -> Non rmtree(penv) -def _current_python_minor() -> str: +def current_python_minor() -> str: """Return the running interpreter's ``major.minor`` (e.g. ``3.13``).""" return f"{sys.version_info.major}.{sys.version_info.minor}" @@ -161,7 +161,7 @@ def heal_platformio_python_env() -> None: def _check_platformio_python_stamp(config: "ProjectConfig") -> None: """Compare the stamp to the running interpreter; wipe and restamp on mismatch.""" - current = _current_python_minor() + current = current_python_minor() stamp_dir = _pio_stamp_dir(config) # Host the stamp/lock even before PlatformIO's first run creates the dir. stamp_dir.mkdir(parents=True, exist_ok=True) @@ -289,6 +289,12 @@ def copy_ccache_script() -> None: ) +def default_libdeps_dir() -> str: + """The PLATFORMIO_LIBDEPS_DIR value a pio run defaults to; the package + prefetch must resolve installed libraries against the same dir.""" + return str(CORE.relative_piolibdeps_path().absolute()) + + def run_platformio_cli(*args, **kwargs) -> str | int: # Re-provision the PlatformIO cache if the interpreter's major.minor changed # since it was last built; a stale platform otherwise rejects the new Python @@ -296,9 +302,7 @@ def run_platformio_cli(*args, **kwargs) -> str | int: heal_platformio_python_env() os.environ["PLATFORMIO_FORCE_COLOR"] = "true" os.environ["PLATFORMIO_BUILD_DIR"] = str(CORE.relative_pioenvs_path().absolute()) - os.environ.setdefault( - "PLATFORMIO_LIBDEPS_DIR", str(CORE.relative_piolibdeps_path().absolute()) - ) + os.environ.setdefault("PLATFORMIO_LIBDEPS_DIR", default_libdeps_dir()) # Suppress Python syntax warnings from third-party scripts during compilation os.environ.setdefault("PYTHONWARNINGS", "ignore::SyntaxWarning") # Increase uv retry count to handle transient network errors (default is 3) @@ -346,6 +350,9 @@ def run_platformio_cli_run(config, verbose, *args, **kwargs) -> str | int: def run_compile(config, verbose): + from esphome.platformio.prefetch import prefetch_platformio_packages + + prefetch_platformio_packages() args = [] if CONF_COMPILE_PROCESS_LIMIT in config[CONF_ESPHOME]: args += [f"-j{config[CONF_ESPHOME][CONF_COMPILE_PROCESS_LIMIT]}"] diff --git a/tests/unit_tests/test_espidf_framework.py b/tests/unit_tests/test_espidf_framework.py index 45a971ca01..afa4433aa1 100644 --- a/tests/unit_tests/test_espidf_framework.py +++ b/tests/unit_tests/test_espidf_framework.py @@ -911,7 +911,7 @@ def test_prefetch_leaves_unverifiable_entries_to_the_installer( "esphome.espidf.framework.run_command", return_value=(True, json.dumps(entries), ""), ), - patch("esphome.espidf.framework.download_with_resume") as download, + patch("esphome.framework_helpers.download_with_resume") as download, patch("esphome.espidf.framework.get_system_python_path", return_value="python"), patch("esphome.framework_helpers._BatchDownloadProgress") as progress_cls, ): @@ -934,7 +934,7 @@ def test_prefetch_all_entries_unverifiable_is_a_noop(tmp_path: Path) -> None: "esphome.espidf.framework.run_command", return_value=(True, json.dumps(entries), ""), ), - patch("esphome.espidf.framework.download_with_resume") as download, + patch("esphome.framework_helpers.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) @@ -952,7 +952,7 @@ def test_prefetch_dedupes_entries_by_dest(tmp_path: Path) -> None: "esphome.espidf.framework.run_command", return_value=(True, json.dumps(entries), ""), ), - patch("esphome.espidf.framework.download_with_resume") as download, + patch("esphome.framework_helpers.download_with_resume") as download, patch("esphome.espidf.framework.get_system_python_path", return_value="python"), patch("esphome.framework_helpers._BatchDownloadProgress"), ): @@ -967,7 +967,7 @@ def test_prefetch_downloads_each_archive_with_resume(tmp_path: Path) -> None: "esphome.espidf.framework.run_command", return_value=(True, _PREFETCH_JSON, ""), ), - patch("esphome.espidf.framework.download_with_resume") as download, + patch("esphome.framework_helpers.download_with_resume") as download, patch("esphome.espidf.framework.get_system_python_path", return_value="python"), patch("esphome.framework_helpers._BatchDownloadProgress") as progress_cls, ): @@ -1011,7 +1011,7 @@ def test_prefetch_downloads_archives_concurrently(tmp_path: Path) -> None: "esphome.espidf.framework.run_command", return_value=(True, json.dumps(entries), ""), ), - patch("esphome.espidf.framework.download_with_resume") as download, + patch("esphome.framework_helpers.download_with_resume") as download, patch("esphome.espidf.framework.get_system_python_path", return_value="python"), patch( "esphome.framework_helpers.ThreadPoolExecutor", wraps=ThreadPoolExecutor @@ -1032,7 +1032,7 @@ def test_prefetch_skips_already_downloaded_archives(tmp_path: Path) -> None: "esphome.espidf.framework.run_command", return_value=(True, _PREFETCH_JSON, ""), ), - patch("esphome.espidf.framework.download_with_resume") as download, + patch("esphome.framework_helpers.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) @@ -1065,7 +1065,7 @@ def test_prefetch_failures_never_raise( with ( patch("esphome.espidf.framework.run_command", return_value=run_result), patch( - "esphome.espidf.framework.download_with_resume", + "esphome.framework_helpers.download_with_resume", side_effect=download_error, ), patch("esphome.espidf.framework.get_system_python_path", return_value="python"), @@ -1087,7 +1087,7 @@ def test_prefetch_total_failure_logs_error( return_value=(True, _PREFETCH_JSON, ""), ), patch( - "esphome.espidf.framework.download_with_resume", + "esphome.framework_helpers.download_with_resume", side_effect=OSError("proxy refuses everything"), ), patch("esphome.espidf.framework.get_system_python_path", return_value="python"), @@ -1112,7 +1112,7 @@ def test_prefetch_one_failed_archive_does_not_stop_the_rest( return_value=(True, _PREFETCH_JSON, ""), ), patch( - "esphome.espidf.framework.download_with_resume", + "esphome.framework_helpers.download_with_resume", side_effect=_fail_cmake_download, ) as download, patch("esphome.espidf.framework.get_system_python_path", return_value="python"), @@ -1133,7 +1133,7 @@ def test_prefetch_finishes_progress_bar_and_cancels_queue(tmp_path: Path) -> Non "esphome.espidf.framework.run_command", return_value=(True, _PREFETCH_JSON, ""), ), - patch("esphome.espidf.framework.download_with_resume"), + patch("esphome.framework_helpers.download_with_resume"), patch("esphome.espidf.framework.get_system_python_path", return_value="python"), patch("esphome.framework_helpers._BatchDownloadProgress") as progress_cls, patch("esphome.framework_helpers.ThreadPoolExecutor") as pool_cls, diff --git a/tests/unit_tests/test_framework_helpers.py b/tests/unit_tests/test_framework_helpers.py index 8844212600..fcc5572f51 100644 --- a/tests/unit_tests/test_framework_helpers.py +++ b/tests/unit_tests/test_framework_helpers.py @@ -2280,6 +2280,32 @@ class TestGetProjectCxxCompileFlags: assert get_project_cxx_compile_flags() == [] +def test_resume_fetch_job_threads_tracker(tmp_path: Path) -> None: + """The batch runner passes the tracker positionally; the shared adapter + must deliver it as download_with_resume's progress keyword.""" + from esphome.framework_helpers import resume_fetch_job + + with patch("esphome.framework_helpers.download_with_resume") as mock_download: + fetch = resume_fetch_job("https://x/a.zip", tmp_path / "a", sha256="ff", size=9) + tracker = lambda done: None # noqa: E731 + fetch(tracker) + mock_download.assert_called_once_with( + "https://x/a.zip", tmp_path / "a", progress=tracker, sha256="ff", size=9 + ) + + +def test_warn_prefetch_failures_names_each_failure( + caplog: pytest.LogCaptureFixture, +) -> None: + """The shared failure loop warns per job with the failure reason.""" + from esphome.framework_helpers import warn_prefetch_failures + + warn_prefetch_failures([("toolchain-x@1", OSError("down"))]) + assert "Could not prefetch toolchain-x@1: down" in caplog.text + warn_prefetch_failures([("lib", OSError("gone"))], "Prefetch of %s failed: %s") + assert "Prefetch of lib failed: gone" in caplog.text + + @pytest.mark.parametrize( ("platform", "input_path", "expected"), [ @@ -2312,3 +2338,18 @@ def test_strip_win_long_path_prefix( r"""``\\?\`` and ``\\?\UNC\`` prefixes are stripped only on win32.""" with patch("esphome.framework_helpers.sys.platform", platform): assert framework_helpers.strip_win_long_path_prefix(input_path) == expected + + +def test_discard_partial_download_logs_undeletable( + tmp_path: Path, caplog: pytest.LogCaptureFixture +) -> None: + """An unremovable staging file leaves a debug trace; the caller's + cache is never pruned, so silence would hide unbounded growth.""" + dest = tmp_path / "archive" + dest.write_bytes(b"stale") + with ( + patch.object(Path, "unlink", side_effect=OSError("busy")), + caplog.at_level(logging.DEBUG), + ): + framework_helpers.discard_partial_download(dest) + assert "Could not remove" in caplog.text diff --git a/tests/unit_tests/test_platformio_prefetch.py b/tests/unit_tests/test_platformio_prefetch.py new file mode 100644 index 0000000000..22e20d0bf6 --- /dev/null +++ b/tests/unit_tests/test_platformio_prefetch.py @@ -0,0 +1,985 @@ +"""Tests for the parallel PlatformIO package prefetch.""" + +import errno +import json +import os +from pathlib import Path +import sys +from types import SimpleNamespace +from unittest.mock import MagicMock, patch + +from filelock import Timeout +import pytest + +from esphome.core import CORE +import esphome.platformio.prefetch as pf + + +@pytest.fixture(autouse=True) +def _core(tmp_path: Path): + CORE.reset() + CORE.build_path = str(tmp_path) + CORE.name = "testenv" + pio_loggers = ("Tool Manager", "Library Manager", "Platform Manager") + saved_propagate = {n: pf.logging.getLogger(n).propagate for n in pio_loggers} + saved_filters = {n: list(pf.logging.getLogger(n).filters) for n in pio_loggers} + # The real setup_log would swap pytest's root-handler formatter + with patch("esphome.log.setup_log"): + yield + # main() flips these process-wide; keep the suite hermetic + for n, flag in saved_propagate.items(): + pf.logging.getLogger(n).propagate = flag + pf.logging.getLogger(n).filters[:] = saved_filters[n] + CORE.reset() + + +class _FakeSpec(SimpleNamespace): + """PackageSpec stand-in for the attributes the prefetch reads.""" + + def __init__( + self, *, owner=None, requirements=None, external=False, **kwargs + ) -> None: + super().__init__( + owner=owner, requirements=requirements, external=external, **kwargs + ) + + +def _fake_manager(tmp_path: Path) -> MagicMock: + m = MagicMock() + m.__class__ = lambda: m # _resolve constructs a same-class instance + m.get_package.return_value = None + m.search_registry_packages.return_value = [{"any": 1}] + m.find_best_registry_version.return_value = ( + {"name": "toolchain-xtensa"}, + { + "name": "2.0.0", + "files": [ + { + "download_url": "https://dl.example/t.tar.gz", + "checksum": {"sha256": "cafe"}, + "size": 1000, + } + ], + }, + ) + m.pick_compatible_pkg_file.side_effect = lambda files: files[0] + m.compute_download_path.side_effect = lambda url, checksum: str( + tmp_path / "dl" / f"{abs(hash((url, checksum)))}" + ) + return m + + +def _mirror_patch(): + return patch.dict( + "sys.modules", + { + "platformio.registry.mirror": SimpleNamespace( + RegistryFileMirrorIterator=lambda url: iter( + [("https://mirror.example/t.tar.gz", "beef")] + ) + ) + }, + ) + + +def test_registry_jobs_resolves_like_platformio(tmp_path: Path) -> None: + """A registry spec resolves to a job keyed by mirror URL and checksum.""" + m = _fake_manager(tmp_path) + with _mirror_patch(): + jobs, failed = pf._registry_jobs( + m, [_FakeSpec(uri=None, name="toolchain-xtensa")], set() + ) + assert failed == 0 + assert len(jobs) == 1 + name, size, fetch = jobs[0] + assert name == "toolchain-xtensa@2.0.0" + assert size == 1000 + m.compute_download_path.assert_called_once_with( + "https://mirror.example/t.tar.gz", "beef" + ) + assert callable(fetch) + + +@pytest.mark.parametrize( + ("method", "attr", "value"), + [ + ("get_package", "return_value", object()), # already installed + ("search_registry_packages", "return_value", []), # unknown package + ("find_best_registry_version", "return_value", (None, None)), # no match + ("pick_compatible_pkg_file", "side_effect", lambda files: None), # no file + ], +) +def test_registry_jobs_skips(tmp_path: Path, method, attr, value) -> None: + """Entries PlatformIO would not download produce no job.""" + m = _fake_manager(tmp_path) + setattr(getattr(m, method), attr, value) + with _mirror_patch(): + assert pf._registry_jobs(m, [_FakeSpec(uri=None, name="x")], set()) == ([], 0) + + +def test_registry_jobs_skips_cached_and_sizeless(tmp_path: Path) -> None: + """Cached or sizeless files are left to PlatformIO.""" + m = _fake_manager(tmp_path) + dl = Path(m.compute_download_path("https://mirror.example/t.tar.gz", "beef")) + dl.parent.mkdir(parents=True, exist_ok=True) + dl.touch() + with _mirror_patch(): + assert pf._registry_jobs(m, [_FakeSpec(uri=None, name="x")], set()) == ([], 0) + dl.unlink() + m.find_best_registry_version.return_value[1]["files"][0]["size"] = 0 + with _mirror_patch(): + assert pf._registry_jobs(m, [_FakeSpec(uri=None, name="x")], set()) == ([], 0) + + +def test_registry_jobs_dedupes_download_paths(tmp_path: Path) -> None: + """Duplicate specs resolve once and one archive yields one job (two + workers must never share a .part); nine specs against eight workers + also exercise the thread-local manager reuse.""" + m = _fake_manager(tmp_path) + specs = [_FakeSpec(uri=None, name="dup"), _FakeSpec(uri=None, name="dup")] + specs += [_FakeSpec(uri=None, name=f"n{i}") for i in range(8)] + with _mirror_patch(): + jobs, failed = pf._registry_jobs(m, specs, set()) + # the fake resolves every spec to the same mirror URL and checksum + assert failed == 0 + assert len(jobs) == 1 + assert m.search_registry_packages.call_count == 9 # dup resolved once + + +def test_registry_jobs_uri_specs_excluded(tmp_path: Path) -> None: + """URL specs never reach the registry resolution.""" + m = _fake_manager(tmp_path) + assert pf._registry_jobs( + m, [_FakeSpec(uri="https://x/y.zip", name="y")], set() + ) == ([], 0) + m.search_registry_packages.assert_not_called() + + +def test_registry_jobs_dedup_keeps_distinct_owners(tmp_path: Path) -> None: + """platformio/x and pioarduino/x are different packages.""" + m = _fake_manager(tmp_path) + specs = [ + _FakeSpec(uri=None, name="framework-x", owner="platformio"), + _FakeSpec(uri=None, name="framework-x", owner="pioarduino"), + ] + with _mirror_patch(): + pf._registry_jobs(m, specs, set()) + assert m.search_registry_packages.call_count == 2 + + +def test_registry_jobs_all_failed_warns_once( + tmp_path: Path, caplog: pytest.LogCaptureFixture +) -> None: + """A whole-batch failure is a systemic fault and must be visible.""" + m = _fake_manager(tmp_path) + m.search_registry_packages.side_effect = RuntimeError("registry down") + with _mirror_patch(): + jobs, failed = pf._registry_jobs( + m, + [_FakeSpec(uri=None, name="a"), _FakeSpec(uri=None, name="b")], + set(), + ) + assert (jobs, failed) == ([], 2) + # The aggregate warning names a cause so an API break does not read + # as a registry outage + assert "Could not resolve 2 of 2" in caplog.text + assert "registry down" in caplog.text + + +def test_uri_fetch_job_promotes_atomically(tmp_path: Path) -> None: + """Checksum-less URL archives land via a locked staging file and an + atomic rename (the stable name is what keeps .part resume working).""" + dl_path = tmp_path / "archive" + + def fake_download(url, dest, progress=None, **kwargs): + Path(dest).write_bytes(b"data") + + manager = MagicMock() + with patch( + "esphome.framework_helpers.download_with_resume", side_effect=fake_download + ): + pf._uri_fetch_job(manager, "https://x/a.zip", dl_path, 4)(lambda done: None) + assert dl_path.read_bytes() == b"data" + # The archive is handed to pio's usage.db pruner + manager.set_download_utime.assert_called_once_with(str(dl_path)) + # no orphaned staging file; the lock file may or may not persist + # (filelock removes it on release on some platforms) + leftovers = {f.name for f in tmp_path.iterdir()} + assert leftovers - {f"{dl_path.name}.prefetch.lock"} == {dl_path.name} + + +def test_registry_fetch_job_skips_when_cached(tmp_path: Path) -> None: + """A destination another process completed is not re-downloaded.""" + dl_path = tmp_path / "archive" + dl_path.write_bytes(b"done") + with patch("esphome.framework_helpers.download_with_resume") as mock_download: + pf._registry_fetch_job( + MagicMock(), "https://x/a.tar.gz", dl_path, "ab" * 32, 4 + )(lambda done: None) + mock_download.assert_not_called() + assert dl_path.read_bytes() == b"done" + + +def test_registry_fetch_job_downloads_under_lock(tmp_path: Path) -> None: + """Registry downloads write the shared cache path under the same lock + the URL path uses; interleaved writers would corrupt the archive.""" + dl_path = tmp_path / "archive" + order: list[str] = [] + with ( + patch( + "esphome.framework_helpers.download_with_resume", + side_effect=lambda url, dest, progress=None, **kw: ( + order.append("fetch"), + Path(dest).write_bytes(b"data"), # registration needs a real file + ), + ), + patch( + "filelock.FileLock.acquire", + side_effect=lambda *a, **k: order.append("lock"), + ), + patch( + "filelock.FileLock.release", + side_effect=lambda *a, **k: order.append("unlock"), + ), + ): + manager = MagicMock() + pf._registry_fetch_job(manager, "https://x/a.tar.gz", dl_path, "ab" * 32, 4)( + lambda done: None + ) + # FileLock.__del__ may add a trailing release; the contract is the order + assert order[:2] == ["lock", "fetch"] + assert "unlock" in order[2:] + manager.set_download_utime.assert_called_once_with(str(dl_path)) + + +def test_lockless_filesystem_downloads_unlocked( + tmp_path: Path, caplog: pytest.LogCaptureFixture +) -> None: + """A filesystem without lock support (ENOSYS/EPERM) degrades to an + unlocked download with one warning, never a per-package failure.""" + dl_path = tmp_path / "archive" + with ( + patch("esphome.framework_helpers.download_with_resume") as mock_download, + patch( + "filelock.FileLock.acquire", + side_effect=OSError(errno.ENOSYS, "no locks"), + ), + patch("filelock.FileLock.release"), + ): + pf._registry_fetch_job( + MagicMock(), "https://x/a.tar.gz", dl_path, "ab" * 32, 4 + )(lambda done: None) + mock_download.assert_called_once() + assert "downloading unlocked" in caplog.text + + +def test_uri_fetch_job_failed_download_keeps_staging(tmp_path: Path) -> None: + """A failed fetch keeps the .part staging bytes for the next resume.""" + dl_path = tmp_path / "archive" + part = tmp_path / "archive.prefetch.part" + part.write_bytes(b"partial") + with ( + patch( + "esphome.framework_helpers.download_with_resume", + side_effect=OSError("network gone"), + ), + pytest.raises(OSError, match="network gone"), + ): + pf._uri_fetch_job(MagicMock(), "https://x/a.zip", dl_path, 4)(lambda done: None) + assert part.read_bytes() == b"partial" + assert not dl_path.exists() + + +def test_uri_fetch_job_rejects_wrong_length(tmp_path: Path) -> None: + """A checksum-less body of the wrong length is never published under a + cache key pio would trust forever.""" + dl_path = tmp_path / "archive" + + def fake_download(url, dest, progress=None, **kwargs): + Path(dest).write_bytes(b"short") + + with ( + patch( + "esphome.framework_helpers.download_with_resume", side_effect=fake_download + ), + pytest.raises(ValueError, match="expected 9999 bytes"), + ): + pf._uri_fetch_job(MagicMock(), "https://x/a.zip", dl_path, 9999)( + lambda done: None + ) + assert not dl_path.exists() + assert not (tmp_path / "archive.prefetch").exists() + + +def test_sweep_stale_sidecars(tmp_path: Path) -> None: + """Sidecars past pio's own expiry are pruned; fresh and foreign files + stay.""" + old_time = pf.time.time() - 110 + stale = tmp_path / "a.tar.gz.part" + stale.write_bytes(b"x") + os.utime(stale, (old_time, old_time)) + fresh = tmp_path / "b.tar.gz.part" + fresh.write_bytes(b"x") + keep = tmp_path / "c.tar.gz" + keep.write_bytes(b"x") + os.utime(keep, (old_time, old_time)) + # A held lock can carry an ancient mtime (O_TRUNC keeps it); locks + # must never be swept or the single-writer guarantee reopens + held_lock = tmp_path / "d.tar.gz.esphome.lock" + held_lock.write_bytes(b"") + os.utime(held_lock, (old_time, old_time)) + pf._sweep_stale_sidecars(tmp_path, 100) + assert not stale.exists() + assert fresh.exists() + assert keep.exists() + assert held_lock.exists() + pf._sweep_stale_sidecars(tmp_path / "missing", 100) # tolerated + + +def test_register_download_failure_leaves_a_trace( + tmp_path: Path, caplog: pytest.LogCaptureFixture +) -> None: + """A failed usage.db registration is traced; an unregistered archive + is never pruned, so silence would hide the leak coming back.""" + manager = MagicMock() + manager.set_download_utime.side_effect = RuntimeError("db locked") + with caplog.at_level(pf.logging.DEBUG): + pf._register_download(manager, tmp_path / "a.tar.gz") + assert "Could not register" in caplog.text + + +def test_sweep_logs_unprunable_files( + tmp_path: Path, caplog: pytest.LogCaptureFixture +) -> None: + """A sidecar that cannot be removed leaves a trace; a sweep that never + prunes must not look like a clean sweep.""" + old_time = pf.time.time() - 110 + stale = tmp_path / "a.tar.gz.part" + stale.write_bytes(b"x") + os.utime(stale, (old_time, old_time)) + with ( + patch.object(Path, "unlink", side_effect=OSError("busy")), + caplog.at_level(pf.logging.DEBUG), + ): + pf._sweep_stale_sidecars(tmp_path, 100) + assert "Could not remove" in caplog.text + + +def test_uri_lock_failure_is_a_counted_failure(tmp_path: Path) -> None: + """The checksum-less URL path never degrades to an unlocked shared + write; interleaved right-length corruption would go undetected.""" + dl_path = tmp_path / "archive" + with ( + patch("esphome.framework_helpers.download_with_resume") as mock_download, + patch( + "filelock.FileLock.acquire", + side_effect=OSError(errno.ENOSYS, "no locks"), + ), + pytest.raises(OSError), + ): + pf._uri_fetch_job(MagicMock(), "https://x/a.zip", dl_path, 4)(lambda done: None) + mock_download.assert_not_called() + + +def test_uri_fetch_job_no_discard_without_a_file(tmp_path: Path) -> None: + """When no archive landed (degraded serialized run), the staging bytes + stay for the next resume instead of being discarded.""" + dl_path = tmp_path / "archive" + part = tmp_path / "archive.prefetch.part" + part.write_bytes(b"partial") + with patch.object(pf, "_serialized_fetch_job", return_value=lambda tracker: None): + pf._uri_fetch_job(MagicMock(), "https://x/a.zip", dl_path, 4)(lambda done: None) + assert part.read_bytes() == b"partial" + + +def test_uri_fetch_job_waits_out_a_briefly_held_lock(tmp_path: Path) -> None: + """A lock freed within the deadline lets the job proceed normally.""" + dl_path = tmp_path / "archive" + + def fake_download(url, dest, progress=None, **kwargs): + Path(dest).write_bytes(b"data") + + with ( + patch( + "esphome.framework_helpers.download_with_resume", side_effect=fake_download + ), + patch("filelock.FileLock.acquire", side_effect=[Timeout("held"), None]), + patch("filelock.FileLock.release"), + ): + pf._uri_fetch_job(MagicMock(), "https://x/a.zip", dl_path, 4)(lambda done: None) + assert dl_path.read_bytes() == b"data" + + +def test_lock_deadline_leaves_download_to_the_holder(tmp_path: Path) -> None: + """A lock held past the deadline means another process is fetching the + same file; skipping cleanly beats a misleading failure warning. The + tracker is still polled so a parked worker observes cancellation.""" + dl_path = tmp_path / "archive" + ticks: list[int] = [] + with ( + patch("esphome.framework_helpers.download_with_resume") as mock_download, + patch("filelock.FileLock.acquire", side_effect=Timeout("held")), + patch.object(pf, "_DOWNLOAD_LOCK_TIMEOUT", 0), + ): + pf._uri_fetch_job(MagicMock(), "https://x/a.zip", dl_path, 4)(ticks.append) + mock_download.assert_not_called() + assert ticks == [0] + assert not dl_path.exists() + + +def test_registry_lock_deadline_skips_registration(tmp_path: Path) -> None: + """A registry job that lost the download race to another process + must not stamp a nonexistent archive into pio's usage.db.""" + manager = MagicMock() + dl_path = tmp_path / "archive" + with ( + patch("esphome.framework_helpers.download_with_resume") as mock_download, + patch("filelock.FileLock.acquire", side_effect=Timeout("held")), + patch.object(pf, "_DOWNLOAD_LOCK_TIMEOUT", 0), + ): + pf._registry_fetch_job(manager, "https://x/a.tar.gz", dl_path, "ab" * 32, 4)( + lambda done: None + ) + mock_download.assert_not_called() + manager.set_download_utime.assert_not_called() + + +def test_main_interrupt_exits_quietly(tmp_path: Path) -> None: + """Ctrl-C reaches the child via the shared process group; it must exit + without a traceback.""" + with ( + patch("esphome.log.setup_log"), + patch.object(pf, "_prefetch", side_effect=KeyboardInterrupt), + ): + assert pf.main([str(tmp_path), "testenv"]) == 130 + + +def test_main_bad_log_level_falls_back(tmp_path: Path) -> None: + with ( + patch.dict("os.environ", {"ESPHOME_PREFETCH_LOG_LEVEL": "verbose"}), + patch("esphome.log.setup_log") as mock_setup, + patch.object(pf, "_prefetch"), + ): + assert pf.main([str(tmp_path), "testenv"]) == 0 + assert mock_setup.call_args[0][0] == pf.logging.INFO + + +def test_main_silences_pio_manager_propagation(tmp_path: Path) -> None: + """The pio manager loggers carry their own handler; propagation to + the root handler would print every install line twice.""" + with patch("esphome.log.setup_log"), patch.object(pf, "_prefetch"): + assert pf.main([str(tmp_path), "testenv"]) == 0 + for name in ("Tool Manager", "Library Manager", "Platform Manager"): + assert pf.logging.getLogger(name).propagate is False + + +def test_main_quiet_level_reaches_pio_manager_loggers(tmp_path: Path) -> None: + """Manager construction re-pins its logger to INFO, so a quiet run + needs the logger-level filter to keep per-package lines out.""" + with ( + patch.dict("os.environ", {"ESPHOME_PREFETCH_LOG_LEVEL": "30"}), + patch("esphome.log.setup_log"), + patch.object(pf, "_prefetch"), + ): + assert pf.main([str(tmp_path), "testenv"]) == 0 + lib_logger = pf.logging.getLogger("Library Manager") + lib_logger.setLevel(pf.logging.INFO) # what pio's _setup_logger does + info = pf.logging.LogRecord("Library Manager", 20, __file__, 1, "x", (), None) + warning = pf.logging.LogRecord("Library Manager", 30, __file__, 1, "x", (), None) + # Logger.filter returns falsy to drop, the record itself to pass + assert not lib_logger.filter(info) + assert lib_logger.filter(warning) + + +def test_main_mirrors_parent_log_setup(tmp_path: Path) -> None: + """The child adopts the parent's dashboard flag and log formatter so + its warnings and progress bar match the parent's.""" + with ( + patch.dict( + "os.environ", + {"ESPHOME_PREFETCH_LOG_LEVEL": "30", "ESPHOME_PREFETCH_DASHBOARD": "1"}, + ), + patch("esphome.log.setup_log") as mock_setup, + patch.object(pf, "_prefetch"), + ): + assert pf.main([str(tmp_path), "testenv"]) == 0 + mock_setup.assert_called_once_with(30) + assert CORE.dashboard is True + + +def test_uri_fetch_job_skips_when_another_process_won(tmp_path: Path) -> None: + """A lost race discards the staging files; the cache never prunes them.""" + dl_path = tmp_path / "archive" + dl_path.write_bytes(b"done") + stale = [ + tmp_path / "archive.prefetch", + tmp_path / "archive.prefetch.part", + tmp_path / "archive.prefetch.part.meta", + ] + for f in stale: + f.write_bytes(b"stale") + with patch("esphome.framework_helpers.download_with_resume") as mock_download: + pf._uri_fetch_job(MagicMock(), "https://x/a.zip", dl_path, 4)(lambda done: None) + mock_download.assert_not_called() + assert dl_path.read_bytes() == b"done" + assert not any(f.exists() for f in stale) + + +def test_registry_jobs_one_bad_spec_keeps_the_rest(tmp_path: Path) -> None: + """A flaky resolution counts as failed without discarding the batch.""" + m = _fake_manager(tmp_path) + m.search_registry_packages.side_effect = [ + RuntimeError("registry 500"), + [{"any": 1}], + ] + with _mirror_patch(): + jobs, failed = pf._registry_jobs( + m, + [_FakeSpec(uri=None, name="flaky"), _FakeSpec(uri=None, name="good")], + set(), + ) + assert failed == 1 + assert len(jobs) == 1 + + +def test_uri_jobs_head_sizes_the_bar(tmp_path: Path) -> None: + """HEAD sizes direct-URL specs; git and unreachable URLs are skipped.""" + m = _fake_manager(tmp_path) + resp = MagicMock() + resp.headers = {"content-length": "2222"} + with patch("esphome.net_retry.http_request", return_value=resp): + jobs, failed = pf._uri_jobs( + m, + [ + _FakeSpec(uri="https://x/big.zip", name="big"), + _FakeSpec(uri="git+https://x/repo.git", name="repo"), + _FakeSpec(uri="https://x/repo.git#v1", name="barevcs"), + _FakeSpec(uri=None, name="registry"), + ], + set(), + ) + assert failed == 0 + assert [(n, s) for n, s, _ in jobs] == [("big", 2222)] + # a successful HEAD with no Content-Length is a clean skip + resp.headers = {} + with patch("esphome.net_retry.http_request", return_value=resp): + assert pf._uri_jobs( + m, [_FakeSpec(uri="https://x/nolen.zip", name="nolen")], set() + ) == ([], 0) + + +def test_uri_jobs_head_failure_counts_as_unresolved( + tmp_path: Path, caplog: pytest.LogCaptureFixture +) -> None: + """Network errors and transient statuses count as unresolved (and warn); + any permanent error status is a clean skip so the sentinel can still + be written (pio run names a broken URL when it downloads).""" + m = _fake_manager(tmp_path) + spec = [_FakeSpec(uri="https://x/a.zip", name="a")] + with patch("esphome.net_retry.http_request", side_effect=OSError("no route")): + assert pf._uri_jobs(m, spec, set()) == ([], 1) + resp = MagicMock(ok=False, status_code=503) + resp.headers = {"content-length": "999"} + with patch("esphome.net_retry.http_request", return_value=resp): + assert pf._uri_jobs(m, spec, set()) == ([], 1) + # 403 is how registries rate-limit; it must not be cached as warm + resp = MagicMock(ok=False, status_code=403) + resp.headers = {"content-length": "999"} + with patch("esphome.net_retry.http_request", return_value=resp): + assert pf._uri_jobs(m, spec, set()) == ([], 1) + resp = MagicMock(ok=False, status_code=405) + resp.headers = {"content-length": "999"} + with patch("esphome.net_retry.http_request", return_value=resp): + assert pf._uri_jobs(m, spec, set()) == ([], 0) + assert "HEAD https://x/a.zip" not in caplog.text + resp = MagicMock(ok=False, status_code=404) + resp.headers = {"content-length": "999"} + with patch("esphome.net_retry.http_request", return_value=resp): + assert pf._uri_jobs(m, spec, set()) == ([], 0) + assert "returned 404" not in caplog.text + + +def test_uri_jobs_dedupes_duplicate_urls(tmp_path: Path) -> None: + """Two specs with one URL yield one HEAD and one job.""" + m = _fake_manager(tmp_path) + resp = MagicMock() + resp.headers = {"content-length": "5"} + with patch("esphome.net_retry.http_request", return_value=resp) as mock_head: + jobs, failed = pf._uri_jobs( + m, + [ + _FakeSpec(uri="https://x/a.zip", name="a"), + _FakeSpec(uri="https://x/a.zip", name="a"), + ], + set(), + ) + assert failed == 0 + assert len(jobs) == 1 + mock_head.assert_called_once() + + +def test_uri_jobs_skips_installed_cached_and_seen(tmp_path: Path) -> None: + m = _fake_manager(tmp_path) + m.get_package.return_value = object() + spec = [_FakeSpec(uri="https://x/a.zip", name="a")] + assert pf._uri_jobs(m, spec, set()) == ([], 0) + m.get_package.return_value = None + dl = Path(m.compute_download_path("https://x/a.zip", "")) + dl.parent.mkdir(parents=True, exist_ok=True) + dl.touch() + assert pf._uri_jobs(m, spec, set()) == ([], 0) + dl.unlink() + # a registry job already claimed this download path + assert pf._uri_jobs(m, spec, {str(dl)}) == ([], 0) + + +def test_prefetch_spawns_isolated_subprocess(tmp_path: Path) -> None: + """Heal runs first, then the subprocess spawns with pio run's libdeps + dir and the parent's PYTHONPATH preserved (the child is esphome).""" + proc = MagicMock(returncode=0) + order = MagicMock() + order.run.return_value = proc + with ( + patch( + "esphome.platformio.toolchain.heal_platformio_python_env", + order.heal, + ), + patch.object(pf.subprocess, "run", order.run) as mock_run, + patch.dict("os.environ", {"PYTHONPATH": "/leak"}), + ): + pf.prefetch_platformio_packages() + assert [c[0] for c in order.mock_calls[:2]] == ["heal", "run"] + (cmd,), kwargs = mock_run.call_args + assert cmd == [ + sys.executable, + "-m", + "esphome.platformio.prefetch", + str(CORE.build_path), + "testenv", + ] + assert kwargs["env"]["PLATFORMIO_LIBDEPS_DIR"] == str( + CORE.relative_piolibdeps_path().absolute() + ) + # The child is esphome itself; PYTHONPATH must survive so it imports + # the same tree (tests/integration pins the source tree through it) + assert kwargs["env"]["PYTHONPATH"] == "/leak" + assert "ESPHOME_PREFETCH_DASHBOARD" not in kwargs["env"] + assert kwargs["timeout"] == pf._PREFETCH_TIMEOUT + + +def test_prefetch_passes_dashboard_flag(tmp_path: Path) -> None: + """The dashboard flag reaches the child so its bar still draws.""" + CORE.dashboard = True + with ( + patch("esphome.platformio.toolchain.heal_platformio_python_env"), + patch.object( + pf.subprocess, "run", return_value=MagicMock(returncode=0) + ) as mock_run, + ): + pf.prefetch_platformio_packages() + assert mock_run.call_args[1]["env"]["ESPHOME_PREFETCH_DASHBOARD"] == "1" + + +@pytest.mark.parametrize( + ("run_effect", "expected"), + [ + ( + {"side_effect": pf.subprocess.TimeoutExpired("cmd", pf._PREFETCH_TIMEOUT)}, + "prefetch timed out", + ), + ({"return_value": MagicMock(returncode=4)}, "prefetch skipped (exit 4)"), + # Exit 1 is the interpreter's own import-failure code, never quiet + ({"return_value": MagicMock(returncode=1)}, "prefetch skipped (exit 1)"), + ({"side_effect": OSError("no exec")}, "PlatformIO package prefetch skipped"), + ], +) +def test_prefetch_spawn_failures_warn_and_continue( + caplog: pytest.LogCaptureFixture, run_effect, expected +) -> None: + """Timeouts, nonzero exits, and spawn failures each warn, never raise.""" + with ( + patch("esphome.platformio.toolchain.heal_platformio_python_env"), + patch.object(pf.subprocess, "run", **run_effect), + ): + pf.prefetch_platformio_packages() + assert expected in caplog.text + + +def test_prefetch_child_handled_failure_is_quiet( + caplog: pytest.LogCaptureFixture, +) -> None: + """Exit _EXIT_HANDLED (3) means the child already warned with the + reason; the parent adds no second warning.""" + with ( + patch("esphome.platformio.toolchain.heal_platformio_python_env"), + patch.object( + pf.subprocess, "run", return_value=MagicMock(returncode=pf._EXIT_HANDLED) + ), + ): + pf.prefetch_platformio_packages() + assert "prefetch skipped" not in caplog.text + + +def test_main_guards_and_exits_nonzero(caplog: pytest.LogCaptureFixture) -> None: + """A swallowed failure still reaches the parent as a nonzero exit; the + parent warns and continues, never failing the build.""" + with patch.object(pf, "_prefetch", side_effect=RuntimeError("boom")): + assert pf.main(["/b", "testenv"]) == pf._EXIT_HANDLED + assert "PlatformIO package prefetch skipped" in caplog.text + + +def test_main_runs_prefetch(tmp_path: Path) -> None: + with patch.object(pf, "_prefetch") as mock_prefetch: + assert pf.main([str(tmp_path), "testenv"]) == 0 + mock_prefetch.assert_called_once_with(tmp_path, "testenv") + + +def test_main_bad_argv_is_a_distinct_exit( + caplog: pytest.LogCaptureFixture, +) -> None: + """A parent/child wiring bug must not look like a network failure.""" + with patch.object(pf, "_prefetch") as mock_prefetch: + assert pf.main(["only-one"]) == 2 + mock_prefetch.assert_not_called() + assert "prefetch usage" in caplog.text + + +def _write_ini(tmp_path: Path, body: str) -> None: + (tmp_path / "platformio.ini").write_text(body) + + +def _write_valid_sentinel(tmp_path: Path, dirs: list[str]) -> None: + (tmp_path / pf._SENTINEL_NAME).write_text( + json.dumps({**pf._sentinel_state(tmp_path), "dirs": dirs}), encoding="utf-8" + ) + + +def test_prefetch_no_platform_returns(tmp_path: Path) -> None: + _write_ini(tmp_path, "[env:testenv]\n") + with patch.object(pf, "_registry_jobs") as mock_jobs: + pf._prefetch(tmp_path, "testenv") + mock_jobs.assert_not_called() + + +def _pio_modules(tmp_path: Path, fake_platform, fake_pm, config, lib_captures=None): + # A bare MagicMock's get_download_dir would fspath to '' and point the + # sidecar sweep at the process cwd + fake_pm.get_download_dir.return_value = str(tmp_path / "downloads") + fake_pm.DOWNLOAD_CACHE_EXPIRE = 86400 * 30 + + def fake_lib_manager(storage_dir): + if lib_captures is not None: + lib_captures.append(storage_dir) + return _fake_manager(tmp_path) + + modules = { + "platformio": MagicMock(), + "platformio.app": MagicMock(), + "platformio.project": MagicMock(), + "platformio.project.config": MagicMock(), + "platformio.dependencies": SimpleNamespace( + get_core_dependencies=lambda: { + "tool-scons": "~4.0", + "contrib-piohome": "~3", + } + ), + "platformio.package": MagicMock(), + "platformio.package.manager": MagicMock(), + "platformio.package.manager.library": SimpleNamespace( + LibraryPackageManager=fake_lib_manager + ), + "platformio.package.manager.platform": SimpleNamespace( + PlatformPackageManager=lambda: fake_pm + ), + "platformio.package.meta": SimpleNamespace( + PackageSpec=lambda *a, **kw: _FakeSpec( + uri=None, + name=kw.get("name") or (a[0] if a else None), + owner=kw.get("owner") + or (str(a[0]).split("/")[0] if a and "/" in str(a[0]) else None), + external=bool(a and "://" in str(a[0])), + ) + ), + "platformio.platform": MagicMock(), + "platformio.platform.factory": SimpleNamespace( + PlatformFactory=SimpleNamespace(new=lambda pkg: fake_platform) + ), + } + modules[ + "platformio.project.config" + ].ProjectConfig.get_instance.return_value = config + return modules + + +def _fake_config(tmp_path: Path, env_options: dict): + config = MagicMock() + options = { + "libdeps_dir": str(tmp_path / "libdeps"), + "packages_dir": str(tmp_path / "packages"), + **env_options, + } + config.get.side_effect = lambda section, key, default=None: options.get( + key, default + ) + return config + + +def test_prefetch_all_cached_is_quiet_and_writes_sentinel(tmp_path: Path) -> None: + """A no-work run neither logs nor batches, and records the sentinel.""" + _write_ini(tmp_path, "[env:testenv]\nplatform = fake/p@1\n") + (tmp_path / "packages").mkdir() + (tmp_path / "libdeps" / "testenv").mkdir(parents=True) + fake_platform = MagicMock() + fake_platform.packages = {} + config = _fake_config( + tmp_path, {"platform": "fake/p@1", "lib_deps": ["esphome/noise-c@1.0"]} + ) + modules = _pio_modules(tmp_path, fake_platform, MagicMock(), config) + with ( + patch.dict("sys.modules", modules), + patch.object(pf, "_registry_jobs", return_value=([], 0)), + patch.object(pf, "_uri_jobs", return_value=([], 0)), + patch.object(pf, "run_batch_downloads") as mock_batch, + ): + pf._prefetch(tmp_path, "testenv") + mock_batch.assert_not_called() + assert pf._prefetch_is_warm(tmp_path) + + +def test_prefetch_failed_resolution_is_not_cached_as_warm(tmp_path: Path) -> None: + """A registry outage must not write the sentinel.""" + _write_ini(tmp_path, "[env:testenv]\nplatform = fake/p@1\n") + (tmp_path / "packages").mkdir() + fake_platform = MagicMock() + fake_platform.packages = {} + config = _fake_config(tmp_path, {"platform": "fake/p@1"}) + modules = _pio_modules(tmp_path, fake_platform, MagicMock(), config) + with ( + patch.dict("sys.modules", modules), + patch.object(pf, "_registry_jobs", return_value=([], 1)), + patch.object(pf, "_uri_jobs", return_value=([], 0)), + patch.object(pf, "run_batch_downloads") as mock_batch, + ): + pf._prefetch(tmp_path, "testenv") + mock_batch.assert_not_called() + assert not (tmp_path / pf._SENTINEL_NAME).exists() + + +def test_sentinel_invalidation(tmp_path: Path) -> None: + """Ini changes, missing dirs, and garbage sentinels all read as cold.""" + _write_ini(tmp_path, "[env:testenv]\nplatform = fake/p@1\n") + pkg_dir = tmp_path / "packages" + pkg_dir.mkdir() + assert not pf._prefetch_is_warm(tmp_path) # no sentinel yet + _write_valid_sentinel(tmp_path, [str(pkg_dir)]) + assert pf._prefetch_is_warm(tmp_path) + _write_ini(tmp_path, "[env:testenv]\nplatform = fake/p@2\n") + assert not pf._prefetch_is_warm(tmp_path) # ini changed + _write_ini(tmp_path, "[env:testenv]\nplatform = fake/p@1\n") + pkg_dir.rmdir() + assert not pf._prefetch_is_warm(tmp_path) # recorded dir gone + (tmp_path / pf._SENTINEL_NAME).write_text("not json", encoding="utf-8") + assert not pf._prefetch_is_warm(tmp_path) + + +def test_prefetch_warm_sentinel_skips_spawn(tmp_path: Path) -> None: + """A valid sentinel skips the subprocess entirely.""" + _write_ini(tmp_path, "[env:testenv]\nplatform = fake/p@1\n") + pkg_dir = tmp_path / "packages" + pkg_dir.mkdir() + _write_valid_sentinel(tmp_path, [str(pkg_dir)]) + with ( + patch("esphome.platformio.toolchain.heal_platformio_python_env"), + patch.object(pf.subprocess, "run") as mock_run, + ): + pf.prefetch_platformio_packages() + mock_run.assert_not_called() + + +def test_prefetch_end_to_end_wiring( + tmp_path: Path, caplog: pytest.LogCaptureFixture +) -> None: + """Platform installs dep-free, non-optional packages plus tool-scons + resolve, libraries use the env libdeps dir, a platform sys.path rewrite + is undone, and failures warn by name.""" + _write_ini(tmp_path, "[env:testenv]\nplatform = fake/platform@1.0\n") + fake_platform = MagicMock() + fake_platform.packages = { + "toolchain-x": {"optional": False}, + "framework-y": {"optional": True}, + } + fake_platform.get_package_spec.side_effect = lambda name: _FakeSpec( + uri=None, name=name + ) + # Platform setup code rewrites sys.path (pioarduino penv); _prefetch + # must restore it + bogus = str(tmp_path / "penv-site-packages") + fake_platform.configure_project_packages.side_effect = lambda env, targets: ( + sys.path.insert(0, bogus) + ) + fake_pm = MagicMock() + config = _fake_config( + tmp_path, + { + "platform": "fake/platform@1.0", + # the bare built-in name and the interpolation are skipped; + # only the owner-qualified library resolves + "lib_deps": ["esphome/noise-c@1.0", "WiFi", "${common.lib_deps}"], + }, + ) + lib_dirs: list[str] = [] + modules = _pio_modules(tmp_path, fake_platform, fake_pm, config, lib_dirs) + (tmp_path / pf._SENTINEL_NAME).write_text("{}", encoding="utf-8") + captured: dict = {} + + def fake_registry_jobs(manager, specs, seen): + captured.setdefault("spec_batches", []).append([s.name for s in specs]) + return [("toolchain-x@1", 10, lambda t: None)], 0 + + with ( + patch.dict("sys.modules", modules), + patch.object(pf, "_registry_jobs", side_effect=fake_registry_jobs), + patch.object(pf, "_uri_jobs", return_value=([], 0)), + patch.object( + pf, + "run_batch_downloads", + return_value=[("toolchain-x@1", OSError("down"))], + ) as mock_batch, + ): + pf._prefetch(tmp_path, "testenv") + fake_pm.install.assert_called_once_with("fake/platform@1.0", skip_dependencies=True) + assert not (tmp_path / pf._SENTINEL_NAME).exists() # stale sentinel removed + fake_platform.configure_project_packages.assert_called_once_with("testenv", ["run"]) + assert bogus not in sys.path + # non-optional platform package + tool-scons (never piohome), then libs + assert captured["spec_batches"][0] == ["toolchain-x", "tool-scons"] + assert captured["spec_batches"][1] == ["esphome/noise-c@1.0"] + assert lib_dirs == [str(Path(tmp_path / "libdeps") / "testenv")] + mock_batch.assert_called_once() + assert "Could not prefetch toolchain-x@1" in caplog.text + + +def test_prefetch_skips_duplicate_tool_scons(tmp_path: Path) -> None: + """A platform that lists tool-scons itself does not get it appended.""" + _write_ini(tmp_path, "[env:testenv]\nplatform = fake/p@1\n") + fake_platform = MagicMock() + fake_platform.packages = {"tool-scons": {"optional": False}} + fake_platform.get_package_spec.side_effect = lambda name: _FakeSpec( + uri=None, name=name + ) + config = _fake_config(tmp_path, {"platform": "fake/p@1"}) + modules = _pio_modules(tmp_path, fake_platform, MagicMock(), config) + batches: list[list[str]] = [] + with ( + patch.dict("sys.modules", modules), + patch.object( + pf, + "_registry_jobs", + side_effect=lambda mgr, specs, seen: ( + batches.append([s.name for s in specs]) or ([], 0) + ), + ), + patch.object(pf, "_uri_jobs", return_value=([], 0)), + ): + pf._prefetch(tmp_path, "testenv") + assert batches[0] == ["tool-scons"] diff --git a/tests/unit_tests/test_platformio_toolchain.py b/tests/unit_tests/test_platformio_toolchain.py index 63c40f3609..fb99ea9208 100644 --- a/tests/unit_tests/test_platformio_toolchain.py +++ b/tests/unit_tests/test_platformio_toolchain.py @@ -932,8 +932,13 @@ def test_run_compile(setup_core: Path, mock_run_platformio_cli_run: Mock) -> Non config = {CONF_ESPHOME: {CONF_COMPILE_PROCESS_LIMIT: 4}} mock_run_platformio_cli_run.return_value = 0 - toolchain.run_compile(config, verbose=True) + with patch( + "esphome.platformio.prefetch.prefetch_platformio_packages" + ) as mock_prefetch: + toolchain.run_compile(config, verbose=True) + # The only wiring of the prefetch into a build lives here + mock_prefetch.assert_called_once_with() mock_run_platformio_cli_run.assert_called_once_with(config, True, "-j4") @@ -947,7 +952,8 @@ def test_run_compile_without_process_limit( config = {CONF_ESPHOME: {}} mock_run_platformio_cli_run.return_value = 0 - toolchain.run_compile(config, verbose=False) + with patch("esphome.platformio.prefetch.prefetch_platformio_packages"): + toolchain.run_compile(config, verbose=False) mock_run_platformio_cli_run.assert_called_once_with(config, False) @@ -1677,8 +1683,8 @@ def pio_core_dir(tmp_path: Path) -> Path: def test_current_python_minor_matches_running_interpreter() -> None: - """_current_python_minor returns major.minor of the running interpreter.""" - assert toolchain._current_python_minor() == _CURRENT_MINOR + """current_python_minor returns major.minor of the running interpreter.""" + assert toolchain.current_python_minor() == _CURRENT_MINOR def test_pio_stamp_round_trip(tmp_path: Path) -> None: