diff --git a/esphome/core/config.py b/esphome/core/config.py index 472ca64c9a..67a7b5210e 100644 --- a/esphome/core/config.py +++ b/esphome/core/config.py @@ -55,6 +55,7 @@ from esphome.helpers import ( cpp_string_escape, fnv1a_32bit_hash, get_str_env, + get_usable_cpu_count, walk_files, ) from esphome.types import ConfigType @@ -205,16 +206,6 @@ def valid_project_name(value: str): return value -def get_usable_cpu_count() -> int: - """Return the number of CPUs that can be used for processes. - On Python 3.13+ this is the number of CPUs that can be used for processes. - On older Python versions this is the number of CPUs. - """ - return ( - os.process_cpu_count() if hasattr(os, "process_cpu_count") else os.cpu_count() - ) - - if "ESPHOME_DEFAULT_COMPILE_PROCESS_LIMIT" in os.environ: _compile_process_limit_default = min( int(os.environ["ESPHOME_DEFAULT_COMPILE_PROCESS_LIMIT"]), get_usable_cpu_count() diff --git a/esphome/helpers.py b/esphome/helpers.py index 4397111c2e..a38fcaf821 100644 --- a/esphome/helpers.py +++ b/esphome/helpers.py @@ -402,6 +402,15 @@ def sort_ip_addresses(address_list: list[str]) -> list[str]: return [socket.getnameinfo(r[4], socket.NI_NUMERICHOST)[0] for r in res] +def get_usable_cpu_count() -> int: + """Return the number of CPUs usable by this process (affinity-aware + on Python 3.13+); 1 when the count is undeterminable.""" + count = ( + os.process_cpu_count() if hasattr(os, "process_cpu_count") else os.cpu_count() + ) + return count or 1 + + def get_bool_env(var, default=False): """Read a boolean env var: the ``cv.boolean`` spellings plus ``1``/``0``; anything else falls through to ``bool(value)``.""" diff --git a/esphome/platformio/prefetch.py b/esphome/platformio/prefetch.py index f313c2f4d0..ef8c27c9aa 100644 --- a/esphome/platformio/prefetch.py +++ b/esphome/platformio/prefetch.py @@ -1,29 +1,35 @@ -"""Parallel prefetch of the packages a PlatformIO run would install. +"""Parallel prefetch and install of the packages a PlatformIO run needs. 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. +(identical ``compute_download_path`` keys), then installs them through +PlatformIO's own ``_install`` with one worker per usable core, so +extraction (the serial, single-core half of a cold install) parallelizes +too and ``pio run`` finds every package already installed. 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 and installs 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 +from contextlib import suppress import hashlib import json import logging import os from pathlib import Path +from queue import SimpleQueue +import signal import subprocess import sys import threading import time -from typing import Any +from typing import Any, NamedTuple from esphome.framework_helpers import ( content_length, @@ -33,7 +39,7 @@ from esphome.framework_helpers import ( run_batch_downloads, warn_prefetch_failures, ) -from esphome.helpers import get_bool_env +from esphome.helpers import get_bool_env, get_usable_cpu_count, rmtree _LOGGER = logging.getLogger(__name__) @@ -78,6 +84,18 @@ def _sweep_stale_sidecars(download_dir: Path, expire_seconds: int) -> None: _LOGGER.debug("Could not sweep %s", download_dir, exc_info=True) +class _Resolved(NamedTuple): + """A registry spec resolved to its archive; ``cached`` skips the download.""" + + spec: Any + name: str + size: int + url: str + dl_path: Path + checksum: str + cached: bool + + # Child records a no-work run; the parent skips the next spawn while valid _SENTINEL_NAME = ".esphome_prefetch.json" _SENTINEL_SCHEMA = 1 @@ -151,24 +169,75 @@ def prefetch_platformio_packages() -> None: 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 + # Not a with-block: the lifetime spans the wait/terminate arms + proc = subprocess.Popen(cmd, env=env) # pylint: disable=consider-using-with 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: + try: + returncode = proc.wait(timeout=_PREFETCH_TIMEOUT) + except subprocess.TimeoutExpired: + _stop_child(proc) + _LOGGER.warning("PlatformIO package prefetch timed out; continuing without it") + return + except BaseException as err: + # SIGKILL (subprocess.run's choice on interrupt) could land inside + # a package-directory copy pio run would then trust; ask first + _stop_child(proc) + if isinstance(err, Exception): + # An unexpected wait() failure must degrade, not fail the build + _LOGGER.warning( + "PlatformIO package prefetch skipped: %s", failure_reason(err) + ) + return + raise + if 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: + elif 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 - ) + _LOGGER.warning("PlatformIO package prefetch skipped (exit %d)", returncode) + + +def _stop_child(proc: subprocess.Popen) -> None: + """Stop the child without cutting an in-flight package install short. + + Wait first (a terminal interrupt already unwinds the child), then + SIGTERM for the clean unwind main() installs, then kill. On Windows + terminate() cannot reach the handler, so its arm is a plain wait. + """ + if proc.poll() is None: + _LOGGER.info("Waiting for the prefetch child to finish its current install") + try: + with suppress(subprocess.TimeoutExpired): + proc.wait(timeout=5) + return + if sys.platform != "win32": + proc.terminate() + with suppress(subprocess.TimeoutExpired): + proc.wait(timeout=30) + return + proc.kill() + proc.wait(timeout=5) + # The kill can land mid-copy; the uncertainty must be visible + _LOGGER.warning("Prefetch child killed; a package install may be incomplete") + except KeyboardInterrupt: + # Kill so an interrupted stop cannot orphan a still-writing child + # (BaseException: a further interrupt must not skip the kill), + # then re-raise so the build aborts + with suppress(BaseException): + proc.kill() + proc.wait(timeout=5) + if proc.poll() is None: + _LOGGER.warning("The prefetch child could not be confirmed stopped") + raise + except Exception: # noqa: BLE001 # pylint: disable=broad-exception-caught + # A surviving child may still be writing packages pio run trusts + _LOGGER.warning("The prefetch child could not be confirmed stopped") + _LOGGER.debug("Stop detail", exc_info=True) def _project_platform_and_config(ini: Path, env: str) -> tuple[str | None, Any]: @@ -183,25 +252,36 @@ def _project_platform_and_config(ini: Path, env: str) -> tuple[str | None, Any]: return config.get(f"env:{env}", "platform", None), config +def _sibling_manager(manager: Any) -> Any: + """A same-store manager equivalent to the shared one.""" + # Hard read: a renamed attribute must fail loudly, not silently drop + # the qualifiers wave-1 installs resolve with; is-not-None so a falsy + # PackageCompatibility still propagates + if (compatibility := manager.compatibility) is not None: + return manager.__class__(manager.package_dir, compatibility=compatibility) + return manager.__class__(manager.package_dir) + + def _registry_jobs( - manager, specs, seen: set[str] -) -> tuple[list[tuple[str, int, Any]], int]: + manager: Any, specs: list[Any], seen: set[str] +) -> tuple[list[tuple[str, int, Any]], int, list[tuple[str, Any]]]: """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). + returns how many resolutions errored (a clean skip is not an error) + and the ``(name, spec)`` pairs whose archives will be installable. """ from platformio.registry.mirror import RegistryFileMirrorIterator local = threading.local() errors: list[str] = [] - def _resolve(spec) -> tuple[str, int, str, Path, str] | object | None: + def _resolve(spec) -> _Resolved | 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__() + mgr = local.mgr = _sibling_manager(manager) try: packages = mgr.search_registry_packages(spec) if not packages: @@ -218,13 +298,13 @@ def _registry_jobs( 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 + cached = dl_path.is_file() # fetched by an earlier run size = pkgfile.get("size") - if not size: + if not cached and 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 + name = f"{package['name']}@{version['name']}" + return _Resolved(spec, name, size or 0, url, dl_path, checksum, cached) 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) @@ -239,20 +319,27 @@ def _registry_jobs( unique.setdefault((s.owner, s.name, str(s.requirements)), s) pending = list(unique.values()) if not pending: - return [], 0 + 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]] = [] + installable: list[tuple[str, 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)) + installable.append((res.name, res.spec)) + if res.cached or str(res.dl_path) in seen: + continue # already fetched, or a duplicate must not share a .part + seen.add(str(res.dl_path)) jobs.append( - (name, size, _registry_fetch_job(manager, url, dl_path, checksum, size)) + ( + res.name, + res.size, + _registry_fetch_job( + manager, res.url, res.dl_path, res.checksum, res.size + ), + ) ) if failed := len(errors): # Visible once per build, naming a cause so an API break does not @@ -264,18 +351,22 @@ def _registry_jobs( len(pending), errors[0], ) - return jobs, failed + return jobs, failed, installable -def _uri_jobs(manager, specs, seen: set[str]) -> tuple[list[tuple[str, int, Any]], int]: +def _uri_jobs( + manager: Any, specs: list[Any], seen: set[str] +) -> tuple[list[tuple[str, int, Any]], int, list[tuple[str, Any]]]: """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). + error) and the ``(name, spec)`` pairs whose archives will be + installable. """ from esphome.net_retry import fetch_with_retry, http_request - candidates: list[tuple[str, str, Path]] = [] + candidates: list[tuple[str, str, Path, Any]] = [] + installable: list[tuple[str, Any]] = [] for spec in specs: url = spec.uri if not url or not url.startswith(("http://", "https://")): @@ -284,12 +375,21 @@ def _uri_jobs(manager, specs, seen: set[str]) -> tuple[list[tuple[str, int, Any] continue # bare-URL VCS spec; PlatformIO clones it if manager.get_package(spec): continue + name = spec.name or url.rsplit("/", 1)[-1] # 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 + if dl_path.is_file(): + if spec.has_custom_name(): + # Only a custom name (Foo=https://...) is the destination + # dir; a URI-derived name's destination comes from the + # archive manifest, so its dedupe key could collide with + # another name and race one directory. pio run installs it. + installable.append((name, spec)) # fetched by an earlier run + continue + if str(dl_path) in seen: + continue # another spec already claimed this .part seen.add(str(dl_path)) - candidates.append((spec.name, url, dl_path)) + candidates.append((spec.name, url, dl_path, spec)) errors: list[str] = [] @@ -314,16 +414,19 @@ def _uri_jobs(manager, specs, seen: set[str]) -> tuple[list[tuple[str, int, Any] return content_length(resp) if not candidates: - return [], 0 + return [], 0, installable with ThreadPoolExecutor(max_workers=min(_RESOLVE_WORKERS, len(candidates))) as ex: - sizes = list(ex.map(_head_size, [url for _, url, _ in candidates])) + 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): + for (name, url, dl_path, spec), 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))) + if spec.has_custom_name(): + # See above: derived-name specs stay with pio run's installer + installable.append((name, spec)) else: # Missing or unusable Content-Length; visible under -v _LOGGER.debug("%s reports no usable length; PlatformIO fetches it", url) @@ -335,7 +438,7 @@ def _uri_jobs(manager, specs, seen: set[str]) -> tuple[list[tuple[str, int, Any] len(candidates), errors[0], ) - return jobs, failed + return jobs, failed, installable def _serialized_fetch_job( @@ -460,11 +563,233 @@ def _uri_fetch_job(manager: Any, url: str, dl_path: Path, size: int) -> Any: return run +# (name, spec) from wave 1, (name, spec, compatibility) from dep waves +_Entry = tuple[str, Any] | tuple[str, Any, Any] + + +def _dependency_entries( + manager: Any, entries: list[_Entry], seen_names: set[str] +) -> list[_Entry]: + """Registry dependencies of the installed entries, one per new name. + + Mostly local manifest reads; the builtin probe walks installed + platforms (each may run platform code). Name-only platform libs stay + with pio run. + """ + + # Hard read: losing this filter would pre-install incompatible + # packages pio run then trusts + compatibility = manager.compatibility + # Tool managers have no builtin table; the contract test pins the name + is_builtin = getattr(manager, "is_builtin_lib", None) + deps: dict[str, Any] = {} + skipped = 0 + for name, spec, *_ in entries: + try: + deps_of = _entry_dependencies(manager, spec, compatibility, is_builtin) + except Exception: # noqa: BLE001 # pylint: disable=broad-exception-caught + # One unreadable manifest must not drop the group's whole wave + _LOGGER.debug("Skipping dependencies of %s", name, exc_info=True) + skipped += 1 + continue + for key, entry in deps_of: + if key not in seen_names: + deps.setdefault(key, entry) + if skipped: + # Visible at default verbosity: a dropped subtree silently + # degrades the wave; per-entry detail stays at debug + _LOGGER.warning( + "Could not read dependencies of %d of %d package(s)", + skipped, + len(entries), + ) + return list(deps.values()) + + +def _entry_dependencies( + manager: Any, spec: Any, compatibility: Any, is_builtin: Any +) -> list[tuple[str, _Entry]]: + from platformio.package.meta import PackageCompatibility + + out: list[tuple[str, _Entry]] = [] + if (pkg := manager.get_package(spec)) is None: + # Only successful installs are walked, so this is a real anomaly + # (stale memcache, name/dir mismatch, a pio API change); raising + # folds it into the caller's aggregate dropped-subtree warning + raise RuntimeError(f"just-installed {spec} is not resolvable") + for dep in manager.get_pkg_dependencies(pkg) or []: + if not (dep.get("owner") or dep.get("version")): + continue + if compatibility and not PackageCompatibility.from_dependency( + dep + ).is_compatible(compatibility): + continue # pio's install_dependency would skip it too + dspec = manager.dependency_to_spec(dep) + if ( + is_builtin + and not dspec.owner + and not dspec.external + and is_builtin(dspec.name) + ): + # pio's LibraryPackageManager.install_dependency skips + # builtins; a registry copy would shadow the bundled one + continue + if not (key := (dspec.name or "").lower()): + _LOGGER.debug("Dependency %r of %s has no name; left to pio run", dep, spec) + continue + if manager.get_package(dspec) is not None: + continue # already installed + # Carry the dep's compatibility so _install searches the + # registry qualified, exactly like pio's install_dependency + out.append( + (key, (dspec.name, dspec, PackageCompatibility.from_dependency(dep))) + ) + return out + + +def _clean_failed_install(mgr: Any, name: str, spec: Any) -> None: + # A post-copy failure leaves a package pio run would trust; remove it + # so pio run genuinely reinstalls it + try: + mgr.memcache_reset() + if (pkg := mgr.get_package(spec)) is not None: + # Dropping the metadata is the invariant: pio's own install + # overwrites a metadata-less dir, so a stuck tree cannot be + # trusted. The rmtree is best-effort tidiness. + (Path(pkg.path) / ".piopm").unlink(missing_ok=True) + with suppress(OSError): + rmtree(pkg.path) + else: + # Nothing was moved into place; the common failure shape + _LOGGER.debug("No on-disk install of %s to remove", name) + except Exception as cleanup_err: # noqa: BLE001 # pylint: disable=broad-exception-caught + _LOGGER.warning( + "Could not remove the failed install of %s: %s", + name, + failure_reason(cleanup_err), + ) + + +def _preinstall( + manager: Any, entries: list[_Entry], seen_names: set[str] | None = None +) -> None: + """Install downloaded packages in parallel via pio's own ``_install``. + + ``entries`` are ``_Entry`` tuples, one per destination directory. + The lock is held around each wave's pool, safe only because pio's + private ``_install`` never re-acquires it (a same-process re-lock + would hang, not fail). Waves skip dependencies; the installed + manifests feed the next wave. Any failure falls back to pio run. + """ + workers = min(get_usable_cpu_count(), len(entries)) + # One manager per worker (_install mutates instance state); built + # serially because construction rewires the shared manager logger + managers: SimpleQueue = SimpleQueue() + for _ in range(workers): + managers.put(_sibling_manager(manager)) + local = threading.local() + + def _install_one(entry) -> bool: + # Wave-1 entries are (name, spec); dependency waves add compatibility + name, spec, *rest = entry + compat = rest[0] if rest else None + if (mgr := getattr(local, "mgr", None)) is None: + # at most `workers` pool threads, one dequeue each + mgr = local.mgr = managers.get_nowait() + try: + mgr._install( # pylint: disable=protected-access # noqa: SLF001 + spec, skip_dependencies=True, compatibility=compat + ) + return True + except Exception as err: # noqa: BLE001 # pylint: disable=broad-exception-caught + _LOGGER.warning("Could not pre-install %s: %s", name, failure_reason(err)) + _LOGGER.debug("Pre-install failure detail", exc_info=True) + _clean_failed_install(mgr, name, spec) + return False + except BaseException: + # A SystemExit from a postinstall must not skip the cleanup + # and leave a torn dir pio run trusts + _clean_failed_install(mgr, name, spec) + raise + + _LOGGER.info( + "Installing %d PlatformIO package(s) with %d extraction worker(s): %s", + len(entries), + workers, + ", ".join(name for name, *_ in entries), + ) + # Postinstall scripts chdir process-globally; the cwd is restored + # after the pool. Concurrent postinstalls can still race pio's + # non-reentrant fs.cd mid-pool; that install fails, warns, and is + # redone serially by pio run. Suppress interleaved progress bars. + os.environ.setdefault("PLATFORMIO_DISABLE_PROGRESSBAR", "true") + # get_tmp_dir/get_download_dir create without exist_ok; racing workers + # would FileExistsError, so create them serially first. Concurrent + # usage.db updates can drop download bookkeeping; never a bad build. + manager.get_tmp_dir() + manager.get_download_dir() + cwd = Path.cwd() + manager.lock() + try: + with ThreadPoolExecutor(max_workers=workers) as ex: + try: + results = list(ex.map(_install_one, entries)) + except BaseException: + # Drop queued installs; in-flight ones finish so no + # package directory is left half copied + ex.shutdown(wait=True, cancel_futures=True) + raise + finally: + # Cleanup must not mask an in-flight exception or skip a step + # Each step runs even if an earlier one fails, and none may + # displace the in-flight exception (SIGTERM's SystemExit + # included) with a downgradeable one + wave_ok = True + for step, label in ( + (manager.memcache_reset, "reset the storage cache"), + (manager.unlock, "release the manager lock"), + (lambda: os.chdir(cwd), "restore the working dir"), + ): + try: + step() + except Exception: # noqa: BLE001,PERF203 # pylint: disable=broad-exception-caught + wave_ok = False + _LOGGER.warning("Could not %s", label) + _LOGGER.debug("Teardown detail", exc_info=True) + if len(entries) > 1 and not any(results): + # A systematic fault, not one bad archive; pio run installs serially + _LOGGER.warning( + "Could not pre-install any of %d PlatformIO package(s)", len(entries) + ) + + seen = seen_names if seen_names is not None else set() + # All entries join seen (failures must not be re-queued); only + # successful installs feed the dependency walk + seen.update(name.split("@", 1)[0].lower() for name, *_ in entries) + installed = [e for e, ok in zip(entries, results, strict=True) if ok] + if not wave_ok: + # A stale cache, an unknown lock state, or a lost cwd would + # poison the next wave; pio run installs the rest cleanly + _LOGGER.warning("Skipping the dependency wave") + return + # The builtin probe may construct platforms whose setup rewrites + # sys.path (see _prefetch); restore it for later imports + saved_sys_path = list(sys.path) + try: + next_entries = _dependency_entries(manager, installed, seen) + finally: + sys.path[:] = saved_sys_path + if next_entries: + # Terminates without a cap: every wave admits only never-seen + # names, so a cycle yields an empty next wave + _preinstall(manager, next_entries, seen) + + 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.package.meta import PackageCompatibility, PackageSpec from platformio.platform.factory import PlatformFactory platform_spec, config = _project_platform_and_config( @@ -504,10 +829,16 @@ def _prefetch(build_dir: Path, env: str) -> None: ) ) lib_deps = config.get(f"env:{env}", "lib_deps", []) - # pio run's storage dir for this env: installed libraries skip by - # disk lookup + # pio run's storage dir for this env, with its compatibility + # qualifiers: an unqualified library install could land a different + # owner's package pio run would then trust + qualifiers: dict[str, Any] = {"platforms": [p.name]} + if framework := config.get(f"env:{env}", "framework", None): + qualifiers["frameworks"] = framework libdeps_dir = Path(config.get("platformio", "libdeps_dir")) / env - lm = LibraryPackageManager(str(libdeps_dir)) + lm = LibraryPackageManager( + str(libdeps_dir), compatibility=PackageCompatibility(**qualifiers) + ) # 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 @@ -520,35 +851,71 @@ def _prefetch(build_dir: Path, env: str) -> None: seen: set[str] = set() jobs: list[tuple[str, int, Any]] = [] + groups: list[tuple[Any, list[tuple[str, Any]]]] = [] unresolved = 0 for mgr, batch in ((p.pm, specs), (lm, lib_specs)): + entries: list[tuple[str, Any]] = [] for build_jobs in (_registry_jobs, _uri_jobs): - batch_jobs, failed = build_jobs(mgr, batch, seen) + batch_jobs, failed, installable = build_jobs(mgr, batch, seen) jobs += batch_jobs unresolved += failed + entries += installable + if entries: + groups.append((mgr, entries)) 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)) + if jobs or groups: + # Real work invalidates any previous no-work record + sentinel.unlink(missing_ok=True) + failed_names: set[str] = set() + if jobs: + _LOGGER.info( + "Prefetching %d PlatformIO package(s): %s", + len(jobs), + ", ".join(name for name, _, _ in jobs), + ) + # PlatformIO retries failed packages itself, without resume + failures = run_batch_downloads("Downloading PlatformIO packages", jobs) + warn_prefetch_failures(failures) + failed_names = {name for name, _ in failures} + elif not groups and 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", + ) + + for mgr, entries in groups: + # One install per destination: pio derives the directory from + # the package name, so key on the name part + to_install = { + name.split("@", 1)[0].lower(): (name, spec) + for name, spec in entries + if name not in failed_names + } + if to_install: + try: + _preinstall(mgr, list(to_install.values())) + except Exception as err: # noqa: BLE001 # pylint: disable=broad-exception-caught + # Each group degrades independently; pio run installs + # whatever this one did not + _LOGGER.warning( + "Pre-install failed for the %s group: %s", + mgr.__class__.__name__, + failure_reason(err), + ) + _LOGGER.debug("Pre-install group failure detail", exc_info=True) + + +def _sigterm(_signum, _frame) -> None: + # Raised in the main thread: the pool's BaseException arm cancels + # queued installs while in-flight copies finish, then finally runs + raise SystemExit(143) def main(argv: list[str]) -> int: @@ -556,6 +923,7 @@ def main(argv: list[str]) -> int: from esphome.core import CORE from esphome.log import setup_log + signal.signal(signal.SIGTERM, _sigterm) raw_level = os.environ.get("ESPHOME_PREFETCH_LOG_LEVEL") try: level = int(raw_level) if raw_level is not None else logging.INFO diff --git a/tests/unit_tests/core/test_config.py b/tests/unit_tests/core/test_config.py index e620f8ec7f..68b165c0d0 100644 --- a/tests/unit_tests/core/test_config.py +++ b/tests/unit_tests/core/test_config.py @@ -3,7 +3,6 @@ from collections.abc import Callable import os from pathlib import Path -import types from typing import Any from unittest.mock import MagicMock, Mock, patch @@ -705,33 +704,6 @@ def test_include_file_with_c_header( assert '#include "c_library.h"' in mock_raw_statement.text -def test_get_usable_cpu_count() -> None: - """Test get_usable_cpu_count returns CPU count.""" - count = config.get_usable_cpu_count() - assert isinstance(count, int) - assert count > 0 - - -def test_get_usable_cpu_count_with_process_cpu_count() -> None: - """Test get_usable_cpu_count uses process_cpu_count when available.""" - # Test with process_cpu_count (Python 3.13+) - # Create a mock os module with process_cpu_count - - mock_os = types.SimpleNamespace(process_cpu_count=lambda: 8, cpu_count=lambda: 4) - - with patch("esphome.core.config.os", mock_os): - # When process_cpu_count exists, it should be used - count = config.get_usable_cpu_count() - assert count == 8 - - # Test fallback to cpu_count when process_cpu_count not available - mock_os_no_process = types.SimpleNamespace(cpu_count=lambda: 4) - - with patch("esphome.core.config.os", mock_os_no_process): - count = config.get_usable_cpu_count() - assert count == 4 - - def test_list_target_platforms(tmp_path: Path) -> None: """Test _list_target_platforms returns available platforms.""" # Create mock components directory structure diff --git a/tests/unit_tests/test_helpers.py b/tests/unit_tests/test_helpers.py index 683fef22cf..53c326e0d0 100644 --- a/tests/unit_tests/test_helpers.py +++ b/tests/unit_tests/test_helpers.py @@ -4,6 +4,7 @@ import os from pathlib import Path import socket import stat +import types from unittest.mock import MagicMock, patch from aioesphomeapi.host_resolver import AddrInfo, IPv4Sockaddr, IPv6Sockaddr @@ -1154,3 +1155,26 @@ def test_progressbar_interrupt_keeps_finished_bar_done(monkeypatch) -> None: def test_format_duration(seconds: float, expected: str) -> None: """Test that durations are rendered as short human-readable strings.""" assert helpers.format_duration(seconds) == expected + + +def test_get_usable_cpu_count() -> None: + """Returns a positive int on the real host.""" + count = helpers.get_usable_cpu_count() + assert isinstance(count, int) + assert count > 0 + + +def test_get_usable_cpu_count_sources() -> None: + """Prefers process_cpu_count, falls back to cpu_count, degrades to 1.""" + mock_os = types.SimpleNamespace(process_cpu_count=lambda: 8, cpu_count=lambda: 4) + with patch("esphome.helpers.os", mock_os): + assert helpers.get_usable_cpu_count() == 8 + + mock_os_no_process = types.SimpleNamespace(cpu_count=lambda: 4) + with patch("esphome.helpers.os", mock_os_no_process): + assert helpers.get_usable_cpu_count() == 4 + + # An undeterminable count degrades to one worker, never zero + mock_os_unknown = types.SimpleNamespace(cpu_count=lambda: None) + with patch("esphome.helpers.os", mock_os_unknown): + assert helpers.get_usable_cpu_count() == 1 diff --git a/tests/unit_tests/test_platformio_prefetch.py b/tests/unit_tests/test_platformio_prefetch.py index 22e20d0bf6..91fb78c6af 100644 --- a/tests/unit_tests/test_platformio_prefetch.py +++ b/tests/unit_tests/test_platformio_prefetch.py @@ -1,14 +1,24 @@ """Tests for the parallel PlatformIO package prefetch.""" import errno +import inspect import json +import logging import os from pathlib import Path +import signal import sys +import threading from types import SimpleNamespace from unittest.mock import MagicMock, patch from filelock import Timeout +from platformio.package.manager._install import PackageManagerInstallMixin +from platformio.package.manager.base import BasePackageManager +from platformio.package.manager.library import LibraryPackageManager +from platformio.package.manager.platform import PlatformPackageManager +from platformio.package.manager.tool import ToolPackageManager +from platformio.package.meta import PackageCompatibility, PackageSpec import pytest from esphome.core import CORE @@ -20,13 +30,20 @@ def _core(tmp_path: Path): CORE.reset() CORE.build_path = str(tmp_path) CORE.name = "testenv" + saved_bar = os.environ.get("PLATFORMIO_DISABLE_PROGRESSBAR") + saved_sigterm = signal.getsignal(signal.SIGTERM) 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 + # _preinstall and main() set these process-wide; keep the suite hermetic + if saved_bar is None: + os.environ.pop("PLATFORMIO_DISABLE_PROGRESSBAR", None) + else: + os.environ["PLATFORMIO_DISABLE_PROGRESSBAR"] = saved_bar + signal.signal(signal.SIGTERM, saved_sigterm) for n, flag in saved_propagate.items(): pf.logging.getLogger(n).propagate = flag pf.logging.getLogger(n).filters[:] = saved_filters[n] @@ -37,17 +54,31 @@ class _FakeSpec(SimpleNamespace): """PackageSpec stand-in for the attributes the prefetch reads.""" def __init__( - self, *, owner=None, requirements=None, external=False, **kwargs + self, + *, + uri=None, + owner=None, + requirements=None, + external=False, + custom_name=False, + **kwargs, ) -> None: super().__init__( - owner=owner, requirements=requirements, external=external, **kwargs + uri=uri, owner=owner, requirements=requirements, external=external, **kwargs ) + self._custom_name = custom_name + + def has_custom_name(self) -> bool: + return self._custom_name def _fake_manager(tmp_path: Path) -> MagicMock: m = MagicMock() - m.__class__ = lambda: m # _resolve constructs a same-class instance + # _resolve and _preinstall construct same-class instances + m.__class__ = lambda package_dir=None, **kwargs: m m.get_package.return_value = None + m.compatibility = None + m.is_builtin_lib.return_value = False m.search_registry_packages.return_value = [{"any": 1}] m.find_best_registry_version.return_value = ( {"name": "toolchain-xtensa"}, @@ -86,11 +117,12 @@ 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() + jobs, failed, installable = pf._registry_jobs( + m, [_FakeSpec(name="toolchain-xtensa")], set() ) assert failed == 0 assert len(jobs) == 1 + assert [n for n, _ in installable] == ["toolchain-xtensa@2.0.0"] name, size, fetch = jobs[0] assert name == "toolchain-xtensa@2.0.0" assert size == 1000 @@ -114,21 +146,32 @@ def test_registry_jobs_skips(tmp_path: Path, method, attr, value) -> None: 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) + assert pf._registry_jobs(m, [_FakeSpec(name="x")], set()) == ( + [], + 0, + [], + ) def test_registry_jobs_skips_cached_and_sizeless(tmp_path: Path) -> None: - """Cached or sizeless files are left to PlatformIO.""" + """A cached archive needs no download but is still installable; a + sizeless uncached one is left to PlatformIO entirely.""" 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) + jobs, failed, installable = pf._registry_jobs(m, [_FakeSpec(name="x")], set()) + assert (jobs, failed) == ([], 0) + assert [n for n, _ in installable] == ["toolchain-xtensa@2.0.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) + assert pf._registry_jobs(m, [_FakeSpec(name="x")], set()) == ( + [], + 0, + [], + ) def test_registry_jobs_dedupes_download_paths(tmp_path: Path) -> None: @@ -136,10 +179,10 @@ def test_registry_jobs_dedupes_download_paths(tmp_path: Path) -> None: 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)] + specs = [_FakeSpec(name="dup"), _FakeSpec(name="dup")] + specs += [_FakeSpec(name=f"n{i}") for i in range(8)] with _mirror_patch(): - jobs, failed = pf._registry_jobs(m, specs, set()) + jobs, failed, _installable = 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 @@ -151,7 +194,7 @@ def test_registry_jobs_uri_specs_excluded(tmp_path: Path) -> None: m = _fake_manager(tmp_path) assert pf._registry_jobs( m, [_FakeSpec(uri="https://x/y.zip", name="y")], set() - ) == ([], 0) + ) == ([], 0, []) m.search_registry_packages.assert_not_called() @@ -159,8 +202,8 @@ 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"), + _FakeSpec(name="framework-x", owner="platformio"), + _FakeSpec(name="framework-x", owner="pioarduino"), ] with _mirror_patch(): pf._registry_jobs(m, specs, set()) @@ -174,12 +217,12 @@ def test_registry_jobs_all_failed_warns_once( m = _fake_manager(tmp_path) m.search_registry_packages.side_effect = RuntimeError("registry down") with _mirror_patch(): - jobs, failed = pf._registry_jobs( + jobs, failed, installable = pf._registry_jobs( m, - [_FakeSpec(uri=None, name="a"), _FakeSpec(uri=None, name="b")], + [_FakeSpec(name="a"), _FakeSpec(name="b")], set(), ) - assert (jobs, failed) == ([], 2) + assert (jobs, failed, installable) == ([], 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 @@ -533,13 +576,14 @@ def test_registry_jobs_one_bad_spec_keeps_the_rest(tmp_path: Path) -> None: [{"any": 1}], ] with _mirror_patch(): - jobs, failed = pf._registry_jobs( + jobs, failed, installable = pf._registry_jobs( m, - [_FakeSpec(uri=None, name="flaky"), _FakeSpec(uri=None, name="good")], + [_FakeSpec(name="flaky"), _FakeSpec(name="good")], set(), ) assert failed == 1 assert len(jobs) == 1 + assert len(installable) == 1 def test_uri_jobs_head_sizes_the_bar(tmp_path: Path) -> None: @@ -548,24 +592,25 @@ def test_uri_jobs_head_sizes_the_bar(tmp_path: Path) -> None: resp = MagicMock() resp.headers = {"content-length": "2222"} with patch("esphome.net_retry.http_request", return_value=resp): - jobs, failed = pf._uri_jobs( + jobs, failed, installable = pf._uri_jobs( m, [ - _FakeSpec(uri="https://x/big.zip", name="big"), + _FakeSpec(uri="https://x/big.zip", name="big", custom_name=True), _FakeSpec(uri="git+https://x/repo.git", name="repo"), _FakeSpec(uri="https://x/repo.git#v1", name="barevcs"), - _FakeSpec(uri=None, name="registry"), + _FakeSpec(name="registry"), ], set(), ) assert failed == 0 assert [(n, s) for n, s, _ in jobs] == [("big", 2222)] + assert [n for n, _ in installable] == ["big"] # 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) + ) == ([], 0, []) def test_uri_jobs_head_failure_counts_as_unresolved( @@ -577,25 +622,25 @@ def test_uri_jobs_head_failure_counts_as_unresolved( 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) + 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) + 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) + 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 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 pf._uri_jobs(m, spec, set()) == ([], 0, []) assert "returned 404" not in caplog.text @@ -605,7 +650,7 @@ def test_uri_jobs_dedupes_duplicate_urls(tmp_path: Path) -> None: 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( + jobs, failed, _installable = pf._uri_jobs( m, [ _FakeSpec(uri="https://x/a.zip", name="a"), @@ -621,22 +666,26 @@ def test_uri_jobs_dedupes_duplicate_urls(tmp_path: Path) -> None: 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) + spec = [_FakeSpec(uri="https://x/a.zip", name="a", custom_name=True)] + 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) + # cached: no download job, but still installable + jobs, failed, installable = pf._uri_jobs(m, spec, set()) + assert (jobs, failed) == ([], 0) + assert [n for n, _ in installable] == ["a"] dl.unlink() # a registry job already claimed this download path - assert pf._uri_jobs(m, spec, {str(dl)}) == ([], 0) + 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) + proc = MagicMock() + proc.wait.return_value = 0 order = MagicMock() order.run.return_value = proc with ( @@ -644,7 +693,7 @@ def test_prefetch_spawns_isolated_subprocess(tmp_path: Path) -> None: "esphome.platformio.toolchain.heal_platformio_python_env", order.heal, ), - patch.object(pf.subprocess, "run", order.run) as mock_run, + patch.object(pf.subprocess, "Popen", order.run) as mock_run, patch.dict("os.environ", {"PYTHONPATH": "/leak"}), ): pf.prefetch_platformio_packages() @@ -664,57 +713,425 @@ def test_prefetch_spawns_isolated_subprocess(tmp_path: Path) -> None: # 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 + proc.wait.assert_called_once_with(timeout=pf._PREFETCH_TIMEOUT) + + +def test_stop_child_windows_never_terminates() -> None: + """The Windows TerminateProcess cannot reach the SIGTERM handler, so + the graceful arm becomes a plain longer wait.""" + proc = MagicMock() + proc.wait.side_effect = [pf.subprocess.TimeoutExpired("x", 1), 0] + with patch.object(pf.sys, "platform", "win32"): + pf._stop_child(proc) + proc.terminate.assert_not_called() + proc.kill.assert_not_called() + + +def test_stop_child_surviving_child_warns(caplog: pytest.LogCaptureFixture) -> None: + """A child that outlives kill() may still be writing packages pio run + trusts; that must be visible at default verbosity.""" + timeout = pf.subprocess.TimeoutExpired("cmd", 5) + proc = MagicMock() + proc.poll.return_value = None # still running: the wait is announced + proc.wait.side_effect = [timeout, timeout, timeout] + with ( + patch.object(pf.sys, "platform", "linux"), + caplog.at_level(pf.logging.INFO), + ): + pf._stop_child(proc) + assert "Waiting for the prefetch child" in caplog.text + assert "could not be confirmed stopped" in caplog.text + + +def test_dependency_entries_isolate_a_bad_manifest(tmp_path: Path) -> None: + """One unreadable manifest skips that entry only, never the group.""" + m = _fake_manager(tmp_path) + m.get_package.side_effect = lambda spec: ( + SimpleNamespace(spec=spec) + if getattr(spec, "name", "") in ("bad", "good") + else None + ) + + def deps_for(pkg): + if pkg.spec.name == "bad": + raise RuntimeError("manifest unreadable") + return [{"owner": "o", "name": "dep", "version": "^1"}] + + m.get_pkg_dependencies.side_effect = deps_for + m.dependency_to_spec.side_effect = lambda dep: _FakeSpec(name=dep["name"]) + entries = pf._dependency_entries( + m, + [ + ("bad@1", _FakeSpec(name="bad")), + ("good@1", _FakeSpec(name="good")), + ], + set(), + ) + assert [name for name, *_ in entries] == ["dep"] + + +def test_dependency_entries_skip_nameless_spec( + tmp_path: Path, caplog: pytest.LogCaptureFixture +) -> None: + """A dependency whose spec has no name has no destination identity; + the drop is diagnosable under -v.""" + m = _fake_manager(tmp_path) + m.get_package.side_effect = lambda spec: ( + SimpleNamespace(spec=spec) if getattr(spec, "name", "") == "top" else None + ) + m.get_pkg_dependencies.return_value = [{"owner": "o", "version": "^1"}] + m.dependency_to_spec.side_effect = lambda dep: _FakeSpec(name=None) + with caplog.at_level(logging.DEBUG): + assert ( + pf._dependency_entries(m, [("top@1", _FakeSpec(name="top"))], set()) == [] + ) + assert "has no name; left to pio run" in caplog.text + + +def test_dependency_entries_filter_seen_names(tmp_path: Path) -> None: + """A dependency already waved under its name is not queued again.""" + m = _fake_manager(tmp_path) + m.get_package.side_effect = lambda spec: ( + SimpleNamespace(spec=spec) if getattr(spec, "name", "") == "top" else None + ) + m.get_pkg_dependencies.return_value = [ + {"owner": "o", "name": "dep", "version": "^1"} + ] + m.dependency_to_spec.side_effect = lambda dep: _FakeSpec(name=dep["name"]) + assert pf._dependency_entries(m, [("top@1", _FakeSpec(name="top"))], {"dep"}) == [] + + +def test_preinstall_cleanup_cannot_displace_the_inflight_error( + tmp_path: Path, caplog: pytest.LogCaptureFixture, monkeypatch: pytest.MonkeyPatch +) -> None: + """A failing unlock or cwd restore must not replace the pool's own + exception (SIGTERM's SystemExit included) with a downgradeable one.""" + m = _fake_manager(tmp_path) + m._install.side_effect = SystemExit(143) + m.unlock.side_effect = RuntimeError("flock broke") + real_chdir = pf.os.chdir + monkeypatch.setattr(pf.os, "chdir", MagicMock(side_effect=OSError("cwd removed"))) + try: + with pytest.raises(SystemExit): + pf._preinstall(m, [("a@1", _FakeSpec(name="a"))]) + finally: + monkeypatch.setattr(pf.os, "chdir", real_chdir) + assert "Could not release the manager lock" in caplog.text + + +def test_preinstall_memcache_failure_leaves_a_trace( + tmp_path: Path, caplog: pytest.LogCaptureFixture +) -> None: + """A failing cache reset warns and skips the dependency wave; the + wave itself still completes.""" + m = _fake_manager(tmp_path) + m.memcache_reset.side_effect = RuntimeError("cache broken") + pf._preinstall(m, [("a@1", _FakeSpec(name="a"))]) + assert "Could not reset the storage cache" in caplog.text + assert "Skipping the dependency wave" in caplog.text + m.get_pkg_dependencies.assert_not_called() + + +def test_prefetch_wait_failure_degrades(caplog: pytest.LogCaptureFixture) -> None: + """An unexpected wait() failure warns and continues; the prefetch must + never become a new way for the build to fail.""" + proc = MagicMock() + proc.wait.side_effect = [RuntimeError("wait broke"), 0] + with ( + patch("esphome.platformio.toolchain.heal_platformio_python_env"), + patch.object(pf.subprocess, "Popen", return_value=proc), + ): + pf.prefetch_platformio_packages() + assert "prefetch skipped" in caplog.text + + +def test_preinstall_stuck_lock_skips_dependency_wave( + tmp_path: Path, caplog: pytest.LogCaptureFixture +) -> None: + """A failed unlock leaves the lock state unknown; the recursive wave + would install under a lock() that silently no-ops.""" + m = _fake_manager(tmp_path) + m.unlock.side_effect = RuntimeError("flock broke") + pf._preinstall(m, [("a@1", _FakeSpec(name="a"))]) + assert "Skipping the dependency wave" in caplog.text + m.get_pkg_dependencies.assert_not_called() + + +def test_preinstall_lost_cwd_warns_and_skips_wave( + tmp_path: Path, caplog: pytest.LogCaptureFixture +) -> None: + """A failed cwd restore is process-global state loss: it warns and + the rest is left to pio run from a clean process.""" + m = _fake_manager(tmp_path) + with patch.object(pf.os, "chdir", side_effect=OSError("cwd gone")): + pf._preinstall(m, [("a@1", _FakeSpec(name="a"))]) + assert "Could not restore the working dir" in caplog.text + assert "Skipping the dependency wave" in caplog.text + m.get_pkg_dependencies.assert_not_called() + + +def test_stop_child_interrupted_and_still_alive_warns( + caplog: pytest.LogCaptureFixture, +) -> None: + """An interrupt triggers a best-effort kill, warns when the child + cannot be confirmed dead, and re-raises so the build aborts.""" + proc = MagicMock() + proc.wait.side_effect = KeyboardInterrupt() + proc.poll.return_value = None + with pytest.raises(KeyboardInterrupt): + pf._stop_child(proc) + proc.kill.assert_called_once_with() + assert "could not be confirmed stopped" in caplog.text + + +def test_uri_derived_name_spec_downloads_but_never_installs(tmp_path: Path) -> None: + """A URL spec whose name is derived from the URI installs into a dir + named by the archive manifest, not the derived name; its archive is + prefetched, but the install stays with pio run.""" + m = _fake_manager(tmp_path) + resp = MagicMock(ok=True) + resp.headers = {"content-length": "4"} + with patch("esphome.net_retry.http_request", return_value=resp): + jobs, failed, installable = pf._uri_jobs( + m, [_FakeSpec(uri="https://x/v1.zip", name="v1")], set() + ) + assert failed == 0 + assert len(jobs) == 1 # still prefetched + assert installable == [] + # Cached-from-an-earlier-run archives are skipped the same way + dl = Path(m.compute_download_path("https://x/v1.zip", "")) + dl.parent.mkdir(parents=True, exist_ok=True) + dl.touch() + assert pf._uri_jobs(m, [_FakeSpec(uri="https://x/v1.zip", name="v1")], set()) == ( + [], + 0, + [], + ) + + +def test_dependency_entries_warn_when_all_reads_fail( + tmp_path: Path, caplog: pytest.LogCaptureFixture +) -> None: + """Every manifest read failing is a systematic fault (a pio API + break), not one bad package; the waves must not vanish silently.""" + m = _fake_manager(tmp_path) + m.get_package.side_effect = lambda spec: SimpleNamespace(spec=spec) + m.get_pkg_dependencies.side_effect = RuntimeError("api break") + assert ( + pf._dependency_entries( + m, + [ + ("a@1", _FakeSpec(name="a")), + ("b@1", _FakeSpec(name="b")), + ], + set(), + ) + == [] + ) + assert "Could not read dependencies of 2 of 2" in caplog.text + + +def _proc(wait_effect) -> MagicMock: + proc = MagicMock() + if isinstance(wait_effect, BaseException): + proc.wait.side_effect = [wait_effect, 0] + else: + proc.wait.return_value = wait_effect + return proc def test_prefetch_passes_dashboard_flag(tmp_path: Path) -> None: """The dashboard flag reaches the child so its bar still draws.""" CORE.dashboard = True + proc = MagicMock() + proc.wait.return_value = 0 with ( patch("esphome.platformio.toolchain.heal_platformio_python_env"), - patch.object( - pf.subprocess, "run", return_value=MagicMock(returncode=0) - ) as mock_run, + patch.object(pf.subprocess, "Popen", return_value=proc) as mock_popen, ): pf.prefetch_platformio_packages() - assert mock_run.call_args[1]["env"]["ESPHOME_PREFETCH_DASHBOARD"] == "1" + assert mock_popen.call_args[1]["env"]["ESPHOME_PREFETCH_DASHBOARD"] == "1" @pytest.mark.parametrize( - ("run_effect", "expected"), + ("wait_effect", "spawn_error", "expected"), [ - ( - {"side_effect": pf.subprocess.TimeoutExpired("cmd", pf._PREFETCH_TIMEOUT)}, - "prefetch timed out", - ), - ({"return_value": MagicMock(returncode=4)}, "prefetch skipped (exit 4)"), + ("timeout", None, "prefetch timed out"), + (4, None, "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"), + (1, None, "prefetch skipped (exit 1)"), + (None, OSError("no exec"), "PlatformIO package prefetch skipped"), ], ) def test_prefetch_spawn_failures_warn_and_continue( - caplog: pytest.LogCaptureFixture, run_effect, expected + caplog: pytest.LogCaptureFixture, wait_effect, spawn_error, expected ) -> None: - """Timeouts, nonzero exits, and spawn failures each warn, never raise.""" + """Timeouts, nonzero exits, and spawn failures each warn, never raise; + a timed-out child is stopped gracefully. The mock is built per test: + a collection-time mock's consumable side_effect breaks reruns.""" + if spawn_error is not None: + popen_effect = {"side_effect": spawn_error} + elif wait_effect == "timeout": + popen_effect = { + "return_value": _proc( + pf.subprocess.TimeoutExpired("cmd", pf._PREFETCH_TIMEOUT) + ) + } + else: + popen_effect = {"return_value": _proc(wait_effect)} with ( patch("esphome.platformio.toolchain.heal_platformio_python_env"), - patch.object(pf.subprocess, "run", **run_effect), + patch.object(pf.subprocess, "Popen", **popen_effect), ): pf.prefetch_platformio_packages() assert expected in caplog.text +def test_stop_child_waits_terminates_then_kills() -> None: + """The stop sequence waits for a self-unwinding child first, then + SIGTERMs, and kills only a child that will not stop. The platform is + pinned: on Windows the terminate arm is deliberately skipped.""" + timeout = pf.subprocess.TimeoutExpired("cmd", 5) + with patch.object(pf.sys, "platform", "linux"): + # Child already unwinding from its own SIGINT: no signals at all + proc = MagicMock() + proc.wait.return_value = 0 + pf._stop_child(proc) + proc.terminate.assert_not_called() + # Child needs the SIGTERM unwind + proc = MagicMock() + proc.wait.side_effect = [timeout, 0] + pf._stop_child(proc) + proc.terminate.assert_called_once_with() + proc.kill.assert_not_called() + # Child ignoring SIGTERM is killed with a bounded reap + proc = MagicMock() + proc.wait.side_effect = [timeout, timeout, 0] + pf._stop_child(proc) + proc.kill.assert_called_once_with() + # An interrupt mid-stop re-raises so the build aborts + proc = MagicMock() + proc.wait.side_effect = KeyboardInterrupt() + with pytest.raises(KeyboardInterrupt): + pf._stop_child(proc) + + +def test_preinstall_failure_removes_torn_destination(tmp_path: Path) -> None: + """A failed install removes whatever get_package can see so pio run + genuinely reinstalls it; a cleanup failure warns.""" + m = _fake_manager(tmp_path) + m._install.side_effect = RuntimeError("postinstall failed") + # cleanup lookup first, then the dependency-wave lookup + m.get_package.side_effect = [SimpleNamespace(path=str(tmp_path / "torn")), None] + removed: list[str] = [] + with patch.object(pf, "rmtree", side_effect=removed.append): + pf._preinstall(m, [("bad@1", _FakeSpec(name="bad"))]) + assert removed == [str(tmp_path / "torn")] + + +def test_preinstall_system_exit_still_cleans(tmp_path: Path) -> None: + """A worker SystemExit runs the torn cleanup before propagating.""" + m = _fake_manager(tmp_path) + m._install.side_effect = SystemExit(143) + m.get_package.side_effect = [SimpleNamespace(path=str(tmp_path / "torn")), None] + removed: list[str] = [] + with ( + patch.object(pf, "rmtree", side_effect=removed.append), + pytest.raises(SystemExit), + ): + pf._preinstall(m, [("bad@1", _FakeSpec(name="bad"))]) + assert removed == [str(tmp_path / "torn")] + + +def test_preinstall_stuck_tree_drops_metadata( + tmp_path: Path, caplog: pytest.LogCaptureFixture +) -> None: + """An unremovable torn tree loses its .piopm so pio run reinstalls + it instead of trusting it forever.""" + m = _fake_manager(tmp_path) + m._install.side_effect = RuntimeError("boom") + torn = tmp_path / "torn" + torn.mkdir() + (torn / ".piopm").write_text("{}") + m.get_package.side_effect = [SimpleNamespace(path=str(torn)), None] + with patch.object(pf, "rmtree", side_effect=OSError("busy")): + pf._preinstall(m, [("bad@1", _FakeSpec(name="bad"))]) + assert not (torn / ".piopm").exists() + assert torn.exists() # tidiness is best-effort; metadata is the invariant + + +def test_preinstall_cleanup_failure_warns( + tmp_path: Path, caplog: pytest.LogCaptureFixture +) -> None: + m = _fake_manager(tmp_path) + m._install.side_effect = RuntimeError("boom") + m.get_package.side_effect = [OSError("scan failed"), None] + pf._preinstall(m, [("bad@1", _FakeSpec(name="bad"))]) + assert "Could not remove the failed install of bad@1" in caplog.text + + +def test_dependency_entries_honor_compatibility(tmp_path: Path) -> None: + """A dependency pio's install_dependency would skip as incompatible is + not pre-installed either.""" + m = _fake_manager(tmp_path) + m.compatibility = PackageCompatibility(platforms=["espressif32"]) + # only the top-level entry is installed; the deps are not + m.get_package.side_effect = lambda spec: ( + SimpleNamespace(spec=spec) if getattr(spec, "name", "") == "top" else None + ) + m.get_pkg_dependencies.return_value = [ + {"owner": "o", "name": "espdep", "version": "^1", "platforms": ["espressif32"]}, + {"owner": "o", "name": "avrdep", "version": "^1", "platforms": ["atmelavr"]}, + ] + m.dependency_to_spec.side_effect = lambda dep: _FakeSpec(name=dep["name"]) + entries = pf._dependency_entries(m, [("top@1", _FakeSpec(name="top"))], set()) + assert [name for name, *_ in entries] == ["espdep"] + + +def test_dependency_entries_skip_builtin_libs(tmp_path: Path) -> None: + """An owner-less versioned dep naming a framework builtin (the dict + manifest form of SPI/Wire) is skipped like pio's install_dependency; + a registry copy would shadow the bundled library.""" + m = _fake_manager(tmp_path) + m.is_builtin_lib.side_effect = lambda name: name == "SPI" + m.get_package.side_effect = lambda spec: ( + SimpleNamespace(spec=spec) if getattr(spec, "name", "") == "top" else None + ) + m.get_pkg_dependencies.return_value = [ + {"name": "SPI", "version": "*"}, + {"name": "realdep", "version": "^1"}, + ] + m.dependency_to_spec.side_effect = lambda dep: _FakeSpec(name=dep["name"]) + entries = pf._dependency_entries(m, [("top@1", _FakeSpec(name="top"))], set()) + assert [name for name, *_ in entries] == ["realdep"] + + +def test_prefetch_interrupt_stops_child_gracefully() -> None: + """On Ctrl-C the stop sequence waits first; a child that exits on its + own is never signalled, and the interrupt re-raises.""" + proc = MagicMock() + proc.wait.side_effect = [KeyboardInterrupt(), 0] + with ( + patch("esphome.platformio.toolchain.heal_platformio_python_env"), + patch.object(pf.subprocess, "Popen", return_value=proc), + pytest.raises(KeyboardInterrupt), + ): + pf.prefetch_platformio_packages() + # the stop sequence's first wait saw the child exit on its own + proc.terminate.assert_not_called() + proc.kill.assert_not_called() + + 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.""" + proc = MagicMock() + proc.wait.return_value = pf._EXIT_HANDLED with ( patch("esphome.platformio.toolchain.heal_platformio_python_env"), - patch.object( - pf.subprocess, "run", return_value=MagicMock(returncode=pf._EXIT_HANDLED) - ), + patch.object(pf.subprocess, "Popen", return_value=proc), ): pf.prefetch_platformio_packages() assert "prefetch skipped" not in caplog.text @@ -767,7 +1184,7 @@ def _pio_modules(tmp_path: Path, fake_platform, fake_pm, config, lib_captures=No fake_pm.get_download_dir.return_value = str(tmp_path / "downloads") fake_pm.DOWNLOAD_CACHE_EXPIRE = 86400 * 30 - def fake_lib_manager(storage_dir): + def fake_lib_manager(storage_dir, **kwargs): if lib_captures is not None: lib_captures.append(storage_dir) return _fake_manager(tmp_path) @@ -798,7 +1215,8 @@ def _pio_modules(tmp_path: Path, fake_platform, fake_pm, config, lib_captures=No 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])), - ) + ), + PackageCompatibility=SimpleNamespace, ), "platformio.platform": MagicMock(), "platformio.platform.factory": SimpleNamespace( @@ -837,12 +1255,14 @@ def test_prefetch_all_cached_is_quiet_and_writes_sentinel(tmp_path: Path) -> Non 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, "_registry_jobs", return_value=([], 0, [])), + patch.object(pf, "_uri_jobs", return_value=([], 0, [])), patch.object(pf, "run_batch_downloads") as mock_batch, + patch.object(pf, "_preinstall") as mock_install, ): pf._prefetch(tmp_path, "testenv") mock_batch.assert_not_called() + mock_install.assert_not_called() assert pf._prefetch_is_warm(tmp_path) @@ -856,8 +1276,8 @@ def test_prefetch_failed_resolution_is_not_cached_as_warm(tmp_path: Path) -> Non 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, "_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") @@ -890,10 +1310,10 @@ def test_prefetch_warm_sentinel_skips_spawn(tmp_path: Path) -> None: _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, + patch.object(pf.subprocess, "Popen") as mock_popen, ): pf.prefetch_platformio_packages() - mock_run.assert_not_called() + mock_popen.assert_not_called() def test_prefetch_end_to_end_wiring( @@ -922,6 +1342,7 @@ def test_prefetch_end_to_end_wiring( tmp_path, { "platform": "fake/platform@1.0", + "framework": "arduino", # 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}"], @@ -934,17 +1355,22 @@ def test_prefetch_end_to_end_wiring( 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 + return ( + [("toolchain-x@1", 10, lambda t: None)], + 0, + [("toolchain-x@1", specs[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, "_uri_jobs", return_value=([], 0, [])), patch.object( pf, "run_batch_downloads", return_value=[("toolchain-x@1", OSError("down"))], ) as mock_batch, + patch.object(pf, "_preinstall") as mock_install, ): pf._prefetch(tmp_path, "testenv") fake_pm.install.assert_called_once_with("fake/platform@1.0", skip_dependencies=True) @@ -957,6 +1383,279 @@ def test_prefetch_end_to_end_wiring( assert lib_dirs == [str(Path(tmp_path / "libdeps") / "testenv")] mock_batch.assert_called_once() assert "Could not prefetch toolchain-x@1" in caplog.text + # every installable failed its download; nothing to pre-install + mock_install.assert_not_called() + + +def test_prefetch_installs_cached_archives_without_downloads( + tmp_path: Path, +) -> None: + """Archives already in the download cache still pre-install (in + parallel) even when there is nothing to download, and no sentinel is + written until everything is installed.""" + _write_ini(tmp_path, "[env:testenv]\nplatform = fake/p@1\n") + fake_platform = MagicMock() + fake_platform.packages = {} + config = _fake_config(tmp_path, {"platform": "fake/p@1"}) + modules = _pio_modules(tmp_path, fake_platform, MagicMock(), config) + spec = _FakeSpec(name="cachedpkg") + with ( + patch.dict("sys.modules", modules), + patch.object( + pf, + "_registry_jobs", + side_effect=[([], 0, [("cachedpkg@1", spec)]), ([], 0, [])], + ), + patch.object(pf, "_uri_jobs", return_value=([], 0, [])), + patch.object(pf, "run_batch_downloads") as mock_batch, + patch.object(pf, "_preinstall") as mock_install, + ): + pf._prefetch(tmp_path, "testenv") + mock_batch.assert_not_called() + assert mock_install.call_count == 1 + assert mock_install.call_args[0][1] == [("cachedpkg@1", spec)] + assert not (tmp_path / pf._SENTINEL_NAME).exists() + + +def test_preinstall_extracts_in_parallel_under_one_lock(tmp_path: Path) -> None: + """The manager lock wraps the whole batch; per-thread managers share + its package dir; one failing install leaves the rest alone.""" + m = _fake_manager(tmp_path) + installed: list[str] = [] + + def fake_install(spec, skip_dependencies, compatibility=None): + # Dependencies must be skipped: a shared dep extracted from two + # threads would race one destination dir + assert skip_dependencies is True + if spec.name == "bad": + raise RuntimeError("corrupt archive") + installed.append(spec.name) + + m._install.side_effect = fake_install + entries = [ + ("a@1", _FakeSpec(name="a")), + ("bad@1", _FakeSpec(name="bad")), + ("b@1", _FakeSpec(name="b")), + ] + pf._preinstall(m, entries) + assert sorted(installed) == ["a", "b"] + m.lock.assert_called_once_with() + m.unlock.assert_called_once_with() + assert m.memcache_reset.call_count >= 1 + + +def test_preinstall_all_failed_warns_once( + tmp_path: Path, caplog: pytest.LogCaptureFixture +) -> None: + """Every install failing is a systemic fault, not archive noise.""" + m = _fake_manager(tmp_path) + m._install.side_effect = AttributeError("_install went away") + pf._preinstall( + m, + [ + ("a@1", _FakeSpec(name="a")), + ("b@1", _FakeSpec(name="b")), + ], + ) + assert "Could not pre-install a@1" in caplog.text + assert "Could not pre-install any of 2" in caplog.text + + +def test_preinstall_dedupes_names_across_entries(tmp_path: Path) -> None: + """Two entries with one name install once (one destination dir).""" + _write_ini(tmp_path, "[env:testenv]\nplatform = fake/p@1\n") + fake_platform = MagicMock() + fake_platform.packages = {} + config = _fake_config(tmp_path, {"platform": "fake/p@1"}) + modules = _pio_modules(tmp_path, fake_platform, MagicMock(), config) + s1 = _FakeSpec(name="dup") + s2 = _FakeSpec(name="dup") + with ( + patch.dict("sys.modules", modules), + patch.object( + pf, + "_registry_jobs", + side_effect=[([], 0, [("pkg@1", s1), ("pkg@1", s2)]), ([], 0, [])], + ), + patch.object(pf, "_uri_jobs", return_value=([], 0, [])), + patch.object(pf, "_preinstall") as mock_install, + ): + pf._prefetch(tmp_path, "testenv") + mock_install.assert_called_once() + (entry,) = mock_install.call_args[0][1] + assert entry[0] == "pkg@1" + assert entry[1] is s2 # the dict comprehension keeps the last duplicate + + +def test_preinstall_runs_dependency_waves(tmp_path: Path) -> None: + """Dependencies of installed packages install in a follow-up wave, + deduped by name; name-only platform libs stay with pio run.""" + m = _fake_manager(tmp_path) + installed: list[str] = [] + m._install.side_effect = lambda spec, skip_dependencies, compatibility=None: ( + installed.append(spec.name if hasattr(spec, "name") else str(spec)) + ) + pkg = SimpleNamespace(spec="noise-c") + m.get_package.side_effect = lambda spec: ( + pkg if getattr(spec, "name", None) == "noise-c" else None + ) + m.get_pkg_dependencies.return_value = [ + {"owner": "esphome", "name": "libsodium", "version": "^1.0"}, + {"owner": "esphome", "name": "libsodium", "version": "^1.0"}, + {"name": "SPI"}, + ] + m.dependency_to_spec.side_effect = lambda dep: _FakeSpec(name=dep["name"]) + pf._preinstall(m, [("noise-c@0.1.21", _FakeSpec(name="noise-c"))]) + assert installed == ["noise-c", "libsodium"] # dep deduped, SPI left out + # The dep wave carries its compatibility so _install searches qualified + dep_call = m._install.call_args_list[-1] + assert dep_call.kwargs["compatibility"] is not None + + +def test_preinstall_dependency_wave_skips_seen_names(tmp_path: Path) -> None: + """A dependency whose name matches an already-waved entry is not + reinstalled.""" + m = _fake_manager(tmp_path) + m.get_package.side_effect = lambda spec: SimpleNamespace(spec=spec) + m.get_pkg_dependencies.return_value = [ + {"owner": "esphome", "name": "noise-c", "version": "^0.1"}, + ] + m.dependency_to_spec.side_effect = lambda dep: _FakeSpec(name=dep["name"]) + installed: list[str] = [] + m._install.side_effect = lambda spec, skip_dependencies, compatibility=None: ( + installed.append(getattr(spec, "name", str(spec))) + ) + pf._preinstall(m, [("noise-c@0.1.21", _FakeSpec(name="noise-c"))]) + assert installed == ["noise-c"] + + +def test_preinstall_uses_distinct_managers_in_parallel(tmp_path: Path) -> None: + """Each worker thread gets its own pre-built manager and installs + genuinely overlap (the barrier deadlocks a serial pool). The worker + count is pinned so a 1-CPU host cannot serialize the pool.""" + barrier = threading.Barrier(2, timeout=5) + used: set = set() + + class _WaveManager: + package_dir = str(tmp_path) + compatibility = None + + def __init__(self, package_dir, **kwargs) -> None: + assert package_dir == str(tmp_path) + + def lock(self) -> None: + pass + + def unlock(self) -> None: + pass + + def memcache_reset(self) -> None: + pass + + def get_tmp_dir(self) -> str: + return str(tmp_path) + + def get_download_dir(self) -> str: + return str(tmp_path) + + def get_package(self, spec): + return None + + def get_pkg_dependencies(self, pkg): + return None + + def _install(self, spec, skip_dependencies, compatibility=None) -> None: + used.add(id(self)) + barrier.wait() + + seed = _WaveManager(str(tmp_path)) + with patch.object(pf, "get_usable_cpu_count", return_value=2): + pf._preinstall( + seed, + [ + ("a@1", _FakeSpec(name="a")), + ("b@1", _FakeSpec(name="b")), + ], + ) + assert len(used) == 2 + assert id(seed) not in used + + +def test_sibling_manager_and_sigterm() -> None: + """Sibling managers inherit compatibility; SIGTERM raises SystemExit.""" + calls = [] + m = MagicMock(package_dir="p", compatibility="qual") + m.__class__ = lambda package_dir, **kw: calls.append((package_dir, kw)) + pf._sibling_manager(m) + m.compatibility = None + pf._sibling_manager(m) + assert calls == [("p", {"compatibility": "qual"}), ("p", {})] + with pytest.raises(SystemExit): + pf._sigterm(15, None) + + +def test_dependency_entries_skip_installed(tmp_path: Path) -> None: + """A dependency a previous build installed stays off the destructive + failure path.""" + m = _fake_manager(tmp_path) + m.get_package.side_effect = lambda spec: SimpleNamespace(spec=spec) + m.get_pkg_dependencies.return_value = [ + {"owner": "o", "name": "already", "version": "^1"}, + ] + m.dependency_to_spec.side_effect = lambda dep: _FakeSpec(name=dep["name"]) + assert pf._dependency_entries(m, [("top@1", _FakeSpec(name="top"))], set()) == [] + + +def test_group_failure_does_not_skip_other_groups(tmp_path: Path) -> None: + """One group's pre-install failure degrades that group only.""" + _write_ini(tmp_path, "[env:testenv]\nplatform = fake/p@1\n") + fake_platform = MagicMock() + fake_platform.packages = {} + config = _fake_config(tmp_path, {"platform": "fake/p@1"}) + modules = _pio_modules(tmp_path, fake_platform, MagicMock(), config) + s1 = _FakeSpec(name="toolpkg") + s2 = _FakeSpec(name="libpkg") + with ( + patch.dict("sys.modules", modules), + patch.object( + pf, + "_registry_jobs", + side_effect=[ + ([], 0, [("toolpkg@1", s1)]), + ([], 0, [("libpkg@1", s2)]), + ], + ), + patch.object(pf, "_uri_jobs", return_value=([], 0, [])), + patch.object( + pf, "_preinstall", side_effect=[RuntimeError("group down"), None] + ) as mock_install, + ): + pf._prefetch(tmp_path, "testenv") + assert mock_install.call_count == 2 + + +def test_preinstall_unlocks_even_when_pool_fails(tmp_path: Path) -> None: + """A failure inside the pool cancels queued installs and releases the + lock; a failing executor construction still releases it.""" + m = _fake_manager(tmp_path) + boom = MagicMock() + boom.__enter__.return_value = boom + boom.map.side_effect = RuntimeError("no threads") + with ( + patch.object(pf, "ThreadPoolExecutor", return_value=boom), + pytest.raises(RuntimeError), + ): + pf._preinstall(m, [("a@1", _FakeSpec(name="a"))]) + m.unlock.assert_called_once_with() + assert boom.shutdown.call_args_list[0][1].get("cancel_futures") is True + m.reset_mock() + # A failing executor construction still releases the lock + with ( + patch.object(pf, "ThreadPoolExecutor", side_effect=RuntimeError("no")), + pytest.raises(RuntimeError), + ): + pf._preinstall(m, [("a@1", _FakeSpec(name="a"))]) + m.unlock.assert_called_once_with() def test_prefetch_skips_duplicate_tool_scons(tmp_path: Path) -> None: @@ -976,10 +1675,57 @@ def test_prefetch_skips_duplicate_tool_scons(tmp_path: Path) -> None: pf, "_registry_jobs", side_effect=lambda mgr, specs, seen: ( - batches.append([s.name for s in specs]) or ([], 0) + batches.append([s.name for s in specs]) or ([], 0, []) ), ), - patch.object(pf, "_uri_jobs", return_value=([], 0)), + patch.object(pf, "_uri_jobs", return_value=([], 0, [])), ): pf._prefetch(tmp_path, "testenv") assert batches[0] == ["tool-scons"] + + +def test_platformio_private_api_contract() -> None: + """The pinned PlatformIO still exposes what the pre-install drives. + + Also load-bearing but unpinnable by introspection: pio's private + _install must never re-acquire the manager's inter-process lock + (locking lives in the public install()); a re-lock would hang the + child for the full prefetch timeout, so re-check it on any bump. + + Everything else in this module mocks the managers, so this is the one + test that fails loudly when a requirements bump changes the private + surface instead of silently degrading the prefetch to a no-op. + """ + params = inspect.signature(PackageManagerInstallMixin._install).parameters + assert "spec" in params + assert "skip_dependencies" in params + assert "compatibility" in params + for cls in (ToolPackageManager, LibraryPackageManager, PlatformPackageManager): + assert "package_dir" in inspect.signature(cls.__init__).parameters + for name in ( + "lock", + "unlock", + "memcache_reset", + "get_package", + "compute_download_path", + "get_pkg_dependencies", + "dependency_to_spec", + ): + assert callable(getattr(BasePackageManager, name)) + # The dependency wave mirrors install_dependency's builtin skip + assert callable(LibraryPackageManager.is_builtin_lib) + # The pre-install passes these positionally / by keyword + assert "compatibility" in inspect.signature(BasePackageManager.__init__).parameters + lib_params = inspect.signature(LibraryPackageManager.__init__).parameters + # Capability, not implementation: an explicit compatibility= parameter + # would serve the call site just as well as **kwargs forwarding + assert "compatibility" in lib_params or any( + p.kind is inspect.Parameter.VAR_KEYWORD for p in lib_params.values() + ) + assert callable(PackageCompatibility.from_dependency) + assert callable(PackageCompatibility.is_compatible) + # Every URL spec derives a name from the URI; only a custom name + # (Foo=https://...) is also the destination dir the wave installs into + derived = PackageSpec("https://x/y/archive/master.zip") + assert derived.name and not derived.has_custom_name() + assert PackageSpec("Foo=https://x/y/archive/master.zip").has_custom_name()