Merge branch 'esp8266-native-build-infra' into esp8266-native-framework-installer

This commit is contained in:
J. Nick Koston
2026-08-23 13:09:54 -05:00
10 changed files with 261 additions and 357 deletions
+37 -29
View File
@@ -2,6 +2,7 @@
from collections.abc import Callable
from ctypes.util import find_library
from functools import partial
import json
import logging
import os
@@ -19,7 +20,6 @@ from esphome.build_helpers.ccache import (
from esphome.build_helpers.tools_cache import IDF_TOOLS_CACHE, tools_cache_path
from esphome.core import Version
from esphome.framework_helpers import (
BatchDownloadProgress,
PathType,
archive_extract_all,
create_venv,
@@ -684,6 +684,18 @@ 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,
@@ -726,22 +738,23 @@ def _prefetch_idf_tool_archives(
for entry in json.loads(stdout):
if (dist_path / entry["dest"]).is_file():
continue
if entry["dest"] in seen_dests:
# Two workers on one .part file would interleave
# seek/truncate writes; mirror the library prefetch's dedupe
continue
seen_dests.add(entry["dest"])
# tools.json always carries sha256 and size; an entry missing
# either must not be downloaded unverified here, so leave it to
# the installer (which fails loudly on a bad archive).
if entry.get("sha256") and entry.get("size"):
entries.append(entry)
else:
# Never download unverified: an entry without sha256/size is
# left to the installer, which fails loudly on a bad archive.
# Checked before the dedupe so it cannot shadow a verifiable
# duplicate of the same dest.
if not (entry.get("sha256") and entry.get("size")):
_LOGGER.warning(
"Tool %s has no sha256/size in the download list; "
"leaving it to the installer",
entry["name"],
)
continue
if entry["dest"] in seen_dests:
# Two workers on one .part file would interleave
# seek/truncate writes; mirror the library prefetch's dedupe
continue
seen_dests.add(entry["dest"])
entries.append(entry)
if not entries:
return
_LOGGER.info(
@@ -750,29 +763,24 @@ def _prefetch_idf_tool_archives(
", ".join(entry["name"] for entry in entries),
)
# Every entry carries a size (checked above), so the combined bar can
# be trusted. Unlike the library prefetch there is no sequential
# fallback: per-file bars from several threads would interleave, and
# skipping the prefetch would lose the resume workaround for #17703.
def _download(entry: dict):
return lambda tracker: download_with_resume(
entry["url"],
dist_path / entry["dest"],
sha256=entry["sha256"],
size=entry["size"],
progress=tracker,
)
# No sequential fallback here: skipping the prefetch would lose the
# resume workaround for #17703, and every entry has a size (above).
# A failed archive is retried by the installer itself (without
# resume); keep prefetching the rest.
failures = run_batch_downloads(
BatchDownloadProgress(
"Downloading ESP-IDF tools", sum(entry["size"] for entry in entries)
),
[(entry["name"], _download(entry)) for entry in entries],
"Downloading ESP-IDF tools",
[
(
entry["name"],
entry["size"],
partial(_download_tool, dist_path, entry),
)
for entry in entries
],
)
for name, e in failures:
_LOGGER.warning("Could not prefetch %s: %s", name, e)
_LOGGER.debug("Prefetch failure detail", exc_info=e)
except Exception as e: # noqa: BLE001 # pylint: disable=broad-exception-caught
# The installer downloads anything missing itself; never let the
# prefetch become a new way for the install to fail.
+27 -40
View File
@@ -736,9 +736,8 @@ def _stream_response_to_file(
(effective offset 0) discards the stale bytes. ``offset`` also seeds the
progress bar so a resumed download shows overall progress. ``size`` is
the known full file size; when None it is derived from the response's
content-length, and without either there is no progress bar. With
``progress`` set, no bar is drawn here; the callback gets the absolute
byte count, seeded with ``offset`` and then after each chunk.
content-length, and without either there is no bar. With ``progress``
set no bar is drawn here; the callback gets the absolute byte count.
"""
f.seek(offset)
f.truncate(offset)
@@ -768,25 +767,25 @@ BATCH_DOWNLOAD_WORKERS = 4
def run_batch_downloads(
progress: "BatchDownloadProgress",
jobs: list[tuple[str, Callable[[Callable[[int], None]], None]]],
header: str,
jobs: list[tuple[str, int, Callable[[Callable[[int], None]], None]]],
max_workers: int = BATCH_DOWNLOAD_WORKERS,
) -> list[tuple[str, Exception]]:
"""Run download jobs concurrently, reporting into one combined bar.
"""Run ``(name, size, fetch)`` download jobs concurrently under one bar.
``jobs`` holds ``(name, fetch)`` pairs where ``fetch(tracker)`` performs
one download reporting absolute byte counts to ``tracker``. Failures are
collected (list.append is atomic under the GIL) and returned after the
bar is done, so the caller's warnings never land on the bar's row; a
failed job credits its tracker 0 so the bar can still complete. Ctrl-C
drops queued jobs and aborts in-flight ones at their next progress
tick; resumable ``.part`` files keep the bytes already fetched.
Each ``fetch(tracker)`` reports absolute byte counts; the bar total is
the sum of the sizes. Failures are returned after the bar is done so
warnings never land on its row. Ctrl-C drops queued jobs and aborts
in-flight ones at their next tick; resumable destinations
(``download_with_resume``) keep their fetched ``.part`` bytes.
``jobs`` must be non-empty.
"""
failures: list[tuple[str, Exception]] = []
progress = _BatchDownloadProgress(header, sum(size for _, size, _ in jobs))
cancelled = threading.Event()
def _run(name: str, fetch: Callable[[Callable[[int], None]], None]) -> None:
def _run(
name: str, fetch: Callable[[Callable[[int], None]], None]
) -> tuple[str, Exception] | None:
tracker = progress.tracker()
def checked(done: int) -> None:
@@ -799,13 +798,14 @@ def run_batch_downloads(
except _BatchDownloadCancelled:
tracker(0)
except Exception as err: # noqa: BLE001 # pylint: disable=broad-exception-caught
failures.append((name, err))
tracker(0)
return (name, err)
return None
ex = ThreadPoolExecutor(max_workers=min(max_workers, len(jobs)))
ex = ThreadPoolExecutor(max_workers=max_workers)
try:
for future in [ex.submit(_run, name, fetch) for name, fetch in jobs]:
future.result()
futures = [ex.submit(_run, name, fetch) for name, _, fetch in jobs]
return [failure for f in futures if (failure := f.result()) is not None]
except BaseException:
# Without this the non-daemon workers download to completion before
# the interpreter can exit, making Ctrl-C ineffective for minutes
@@ -814,22 +814,18 @@ def run_batch_downloads(
finally:
ex.shutdown(wait=True, cancel_futures=True)
progress.done()
return failures
class _BatchDownloadCancelled(Exception):
"""Raised inside a download job to abandon it after Ctrl-C."""
class BatchDownloadProgress:
"""One progress bar across several concurrent ``download_with_resume`` calls.
class _BatchDownloadProgress:
"""One bar across several concurrent downloads, summing tracker bytes.
Each ``tracker()`` is a ``progress`` callback for one download; it reports
that file's absolute byte count and the bar shows the sum over ``total``.
The lock also serialises the bar's stderr writes, so worker threads never
interleave frames. With an unknown ``total`` (0) nothing is drawn. Call
``done()`` once every download has finished (or failed) so a bar that
never reached 100% still ends its line before the next log message.
The lock also serialises stderr writes so workers never interleave
frames; a ``total`` of 0 draws nothing. Call ``done()`` at the end so a
bar short of 100% still ends its line.
"""
def __init__(self, header: str, total: int) -> None:
@@ -853,13 +849,7 @@ class BatchDownloadProgress:
return update
def done(self) -> None:
# Nothing to end unless a frame was drawn and it was not the final
# one (update(1) already emitted its own newline).
if (
self._bar is not None
and self._bar.last_progress is not None
and self._bar.last_progress != 100
):
if self._bar is not None:
self._bar.done()
@@ -896,11 +886,8 @@ def download_with_resume(
of consuming attempts — for callers with their own fallback, like
``download_from_mirrors``.
``progress``, when given, replaces the built-in progress bar: it is called
with the absolute number of bytes of ``dest`` obtained so far (including
a resumed prefix, and the final size once the file is verified), so a
caller running several downloads at once can draw one combined bar (see
``BatchDownloadProgress``).
``progress`` replaces the built-in bar: it receives the absolute bytes of
``dest`` obtained so far (see ``BatchDownloadProgress``).
Raises EsphomeError when all attempts are exhausted.
"""
+2 -1
View File
@@ -735,7 +735,8 @@ class ProgressBar:
sys.stderr.flush()
def done(self) -> None:
if not self.enabled:
# No frame drawn, or the 100% frame already ended its own line
if not self.enabled or self.last_progress is None or self.last_progress == 100:
return
sys.stderr.write("\n")
sys.stderr.flush()
+6 -14
View File
@@ -30,10 +30,7 @@ def apply_extra_script(
pio_platform: str,
) -> None:
"""Run a library's ``extraScript`` and fold its captured env vars into
``component.data["build"]["flags"]``.
``board_mcu`` is a callable so its lookup runs only when a script will.
"""
``build.flags``; ``board_mcu`` is a callable so it resolves lazily."""
extra_script = component.data.get("build", {}).get("extraScript")
if not extra_script:
return
@@ -177,12 +174,10 @@ def run_extra_script(
board_mcu: str,
pio_platform: str,
) -> ExtraScriptResult:
"""Execute ``script_path`` with a fake SCons env and return captured vars.
"""Execute ``script_path`` with a fake SCons env, ``library_dir`` as CWD.
Runs with ``library_dir`` as CWD so relative lookups resolve against
the library tree. A crashed script warns and returns an empty result,
never a partial capture.
"""
A crashed script warns and returns an empty result, never a partial
capture."""
env = _FakeSConsEnv(
board_mcu=board_mcu,
pio_env=f"esphome_{board_mcu}",
@@ -248,11 +243,8 @@ def run_extra_script(
def captured_as_build_flags(
result: ExtraScriptResult, *, library_dir: Path
) -> list[str]:
"""Translate captured env vars into -L/-l/-D/raw build flags.
``LIBPATH`` entries are made relative to ``library_dir`` so the
generated build files stay portable.
"""
"""Translate captured env vars into -L/-l/-D/raw build flags; path
entries anchor to ``library_dir`` so the build files stay portable."""
flags: list[str] = []
def _strs(bucket: list, kind: str) -> list[str]:
+93 -158
View File
@@ -14,8 +14,8 @@ regardless of which toolchain consumes the result.
from collections import deque
from collections.abc import Callable, Iterable
from concurrent.futures import ThreadPoolExecutor
from dataclasses import dataclass, field
from functools import partial
import glob
import hashlib
import itertools
@@ -32,8 +32,6 @@ from urllib.request import url2pathname
from esphome import git
from esphome.core import CORE, EsphomeError, Library
from esphome.framework_helpers import (
BATCH_DOWNLOAD_WORKERS,
BatchDownloadProgress,
archive_extract_all,
download_from_mirrors,
rmdir,
@@ -55,12 +53,9 @@ DEFAULT_BUILD_SRC_FILTER = (
DEFAULT_BUILD_SRC_DIRS = "src"
DEFAULT_BUILD_INCLUDE_DIR = "include"
DEFAULT_BUILD_FLAGS = []
# Suffix -> compiler kind (PlatformIO's CSUFFIXES/CXXSUFFIXES/ASSUFFIXES).
# "asm" merges SCons's AS and ASPP sets: all compile as assembler-with-cpp.
# The kind values drive the ESP8266 native ninja rules (later in this
# chain); existing backends consume only the keys. Note .C/.C++ join the
# suffix set here per CXXSUFFIXES; SCons demotes .C to C on
# case-insensitive filesystems, we always treat it as C++.
# Suffix -> compiler kind (PlatformIO's CSUFFIXES/CXXSUFFIXES/ASSUFFIXES);
# "asm" merges SCons's AS and ASPP sets. Per CXXSUFFIXES .C/.C++ are C++
# here, even where SCons demotes .C on case-insensitive filesystems.
SOURCE_KIND_FOR_SUFFIX: dict[str, str] = {
".c": "c",
".cpp": "cxx",
@@ -87,12 +82,7 @@ ESPHOME_DATA_EXTRA_CMAKE_KEY = "EXTRA_CMAKE"
class Source:
def download(
self,
dir_suffix: str,
force: bool = False,
salt: str = "",
namespace: str = "",
progress: Callable[[int], None] | None = None,
self, dir_suffix: str, force: bool = False, salt: str = "", namespace: str = ""
) -> Path:
raise NotImplementedError
@@ -106,8 +96,11 @@ class Source:
class URLSource(Source):
def __init__(self, url: str):
def __init__(self, url: str, size: int | None = None):
self.url = url
# Archive size as reported by the registry, when known; sizes the
# combined prefetch bar without any extra network probe
self.size = size
def _cache_dir(self, dir_suffix: str, salt: str, namespace: str) -> Path:
# Namespace the cache per backend (e.g. pio_components/idf, .../zephyr) so
@@ -169,12 +162,7 @@ class GitSource(Source):
self.ref = ref
def download(
self,
dir_suffix: str,
force: bool = False,
salt: str = "",
namespace: str = "",
progress: Callable[[int], None] | None = None,
self, dir_suffix: str, force: bool = False, salt: str = "", namespace: str = ""
) -> Path:
domain = DOMAIN
if namespace:
@@ -209,12 +197,7 @@ class LocalSource(Source):
self.local_path = path
def download(
self,
dir_suffix: str,
force: bool = False,
salt: str = "",
namespace: str = "",
progress: Callable[[int], None] | None = None,
self, dir_suffix: str, force: bool = False, salt: str = "", namespace: str = ""
) -> Path:
src = Path(self.local_path)
if not src.is_dir():
@@ -252,11 +235,8 @@ class InvalidLibrary(Exception):
class IncompatiblePlatform(InvalidLibrary):
"""The manifest's platform filter rejected the target platform.
A distinct type so callers can treat the routine cross-platform skip
differently from other manifest problems without matching message text.
"""
"""The routine cross-platform skip, typed so callers need not match
message text."""
class ConvertedLibrary:
@@ -307,13 +287,7 @@ class ConvertedLibrary:
def get_require_name(self):
return self.get_sanitized_name().replace("/", "__")
def download(
self,
force: bool = False,
salt: str = "",
namespace: str = "",
progress: Callable[[int], None] | None = None,
):
def download(self, force: bool = False, salt: str = "", namespace: str = ""):
"""Fetch the library into the shared cache and record its ``path``.
The cache directory is named after the sanitized library name; backends
@@ -322,11 +296,7 @@ class ConvertedLibrary:
``get_require_name``). ``namespace`` keeps each backend's cache separate.
"""
self.path = self.source.download(
self.get_sanitized_name(),
force=force,
salt=salt,
namespace=namespace,
progress=progress,
self.get_sanitized_name(), force=force, salt=salt, namespace=namespace
)
self.source_path = self.source.source_root(self.path)
@@ -584,9 +554,10 @@ def _make_registry_client() -> Any:
def _resolve_registry_version(
owner: str | None, pkgname: str, requirements: set[str]
) -> tuple[str, str, str, str]:
) -> tuple[str, str, str, str, int | None]:
"""Resolve a registry package to the single highest version satisfying ALL
the given requirements; return ``(owner, name, version, download_url)``.
the given requirements; return ``(owner, name, version, download_url,
size)`` (``size`` is None when the registry omits it).
Intersecting every requirement (rather than resolving each consumer in
isolation) makes the result independent of processing order and guarantees
@@ -616,7 +587,7 @@ def _resolve_registry_version(
pkgfile = registry.pick_compatible_pkg_file(best["files"])
if not pkgfile:
raise RuntimeError(f"No package file for {owner}/{name}@{best['name']}")
return owner, name, best["name"], pkgfile["download_url"]
return owner, name, best["name"], pkgfile["download_url"], pkgfile.get("size")
def split_flag_entry(entry: Any, owner: str) -> list[str]:
@@ -635,9 +606,8 @@ def split_flag_entry(entry: Any, owner: str) -> list[str]:
def lex_build_flags(entries: str | list[str], owner: str) -> list[str]:
"""Shell-lex ``build.flags`` entries the way PlatformIO's ParseFlags
does; bare -I/-L/-l/-D tokens re-glue to their argument."""
# Join per entry, as SCons's ParseFlags lexes each string independently:
# a dangling -I ending one entry must warn, not absorb the next entry's
# first token.
# Lex per entry as ParseFlags does: a dangling -I must warn, not absorb
# the next entry's first token
return [
token
for entry in ensure_list(entries)
@@ -650,13 +620,9 @@ BARE_ARG_FLAGS = frozenset({"-I", "-L", "-l", "-D"})
def join_flag_args(tokens: Iterable[str], owner: str) -> list[str]:
"""Join a bare ``-I``/``-L``/``-l``/``-D`` with its following token,
the way PlatformIO's ParseFlags lexes them.
A trailing or empty argument (``-D ""``) is warned and dropped: the
bare flag would make gcc eat the next flag as its argument (or add
the CWD for ``-L``); always a typo.
"""
"""Join a bare ``-I``/``-L``/``-l``/``-D`` with its following token, as
PlatformIO's ParseFlags does. A trailing or empty argument is warned and
dropped: the bare flag would make gcc eat the next flag."""
out: list[str] = []
it = iter(tokens)
for tok in it:
@@ -676,11 +642,8 @@ def join_flag_args(tokens: Iterable[str], owner: str) -> list[str]:
def warn_properties_depends(name: str, data: object) -> None:
"""Warn when a manifest declares dependencies only as ``depends=``.
The dependency walk reads the JSON ``dependencies`` key; the raw
``library.properties`` spelling would otherwise drop silently.
"""
"""Warn for ``depends=``-only manifests; the walk reads only the JSON
``dependencies`` key, so they would otherwise drop silently."""
if isinstance(data, dict) and not data.get("dependencies") and data.get("depends"):
# INFO: common and unactionable for transitive libraries; a WARNING
# on every build would train users to ignore the stream
@@ -711,13 +674,9 @@ def dependency_is_usable(
def _valid_dependency_entry(entry: dict, manifest_name: str) -> bool:
"""Whether a normalized entry carries a usable name and version.
The name must be a non-empty string (every consumer indexes or joins
it); a present version must be a string (a container would raise from
``set.add()``, an int fails opaquely inside the registry resolution).
Invalid entries warn naming the manifest.
"""
"""Whether a normalized entry carries a usable name (non-empty string)
and version (string, if present); invalid entries warn naming the
manifest."""
name = entry.get("name")
if (
isinstance(name, str)
@@ -914,29 +873,17 @@ def is_lib_ignored(name: str | None, lib_ignore: set[str]) -> bool:
)
def _content_lengths(urls: list[str]) -> list[int | None]:
"""Content-Length per URL via HEAD requests; None when unknown."""
import requests
from esphome.happy_eyeballs import ensure_happy_eyeballs
# Same convention as every other network call: without it a broken-IPv6
# network burns the full timeout per HEAD before falling back
ensure_happy_eyeballs()
def head(url: str) -> int | None:
try:
resp = requests.head(url, timeout=10, allow_redirects=True)
if not resp.ok:
_LOGGER.debug("HEAD %s returned %s", url, resp.status_code)
return None
return int(resp.headers.get("content-length", 0)) or None
except (requests.RequestException, ValueError) as err:
_LOGGER.debug("HEAD %s failed: %s", url, err)
return None
with ThreadPoolExecutor(max_workers=min(BATCH_DOWNLOAD_WORKERS, len(urls))) as ex:
return list(ex.map(head, urls))
def _fetch_source(
component: ConvertedLibrary,
salt: str,
namespace: str,
tracker: Callable[[int], None],
) -> None:
# Straight to URLSource: only it takes progress, and mutating the
# shared component from a worker is the authoritative loop's job
component.source.download(
component.get_sanitized_name(), salt=salt, namespace=namespace, progress=tracker
)
def _prefetch_wave(
@@ -944,68 +891,58 @@ def _prefetch_wave(
) -> None:
"""Best-effort parallel download of a wave's registry archives.
The walk's own ``download()`` call stays authoritative (it surfaces real
failures, with resume); bars are suppressed since parallel bars would
interleave. Duplicate URLs prefetch once so two threads never extract
into the same cache directory.
The walk's own ``download()`` stays authoritative; duplicate URLs
prefetch once so two threads never share a cache directory. Archives
whose size the registry did not report are left to the sequential
loop, whose per-file bars don't interleave.
"""
components: list[ConvertedLibrary] = []
seen: set[str] = set()
for _key, component in wave:
if not isinstance(component.source, URLSource):
continue
if component.source.url in seen:
continue
seen.add(component.source.url)
try:
cached = component.source.is_cached(
component.get_sanitized_name(), salt=salt, namespace=namespace
)
except Exception as err: # noqa: BLE001 # pylint: disable=broad-exception-caught
# Best-effort: a failing probe prefetches (and re-downloads)
_LOGGER.debug("Cache probe for %s failed: %s", component.name, err)
cached = False
if cached:
# A completed extraction downloads nothing; a warm build must
# stay silent
continue
components.append(component)
if len(components) < 2:
return
# One combined bar over the batch, sized by HEAD requests. An unknown
# size would mean a silent multi-MB download; fall back to sequential
# downloads with their per-file bars instead.
sizes = _content_lengths([c.source.url for c in components])
if not all(sizes):
# Announced before the sequential per-file downloads take over, so
# the fallback is distinguishable from a hang
try:
components: list[ConvertedLibrary] = []
seen: set[str] = set()
for _key, component in wave:
source = component.source
if not isinstance(source, URLSource) or not source.size:
continue
if source.url in seen:
continue
seen.add(source.url)
try:
cached = source.is_cached(
component.get_sanitized_name(), salt=salt, namespace=namespace
)
except Exception as err: # noqa: BLE001 # pylint: disable=broad-exception-caught
# Best-effort: a failing probe prefetches (and re-downloads)
_LOGGER.debug("Cache probe for %s failed: %s", component.name, err)
cached = False
if cached:
# A warm build must stay silent
continue
components.append(component)
if len(components) < 2:
return
_LOGGER.info(
"No Content-Length for %s; downloading sequentially",
", ".join(
c.source.url
for c, size in zip(components, sizes, strict=True)
if not size
),
"Downloading %d libraries: %s",
len(components),
", ".join(c.name for c in components),
)
return
_LOGGER.info(
"Downloading %d libraries: %s",
len(components),
", ".join(c.name for c in components),
)
def _fetch(component: ConvertedLibrary):
return lambda tracker: component.download(
salt=salt, namespace=namespace, progress=tracker
failures = run_batch_downloads(
"Downloading libraries",
[
(c.name, c.source.size, partial(_fetch_source, c, salt, namespace))
for c in components
],
)
failures = run_batch_downloads(
BatchDownloadProgress("Downloading libraries", sum(sizes)),
[(component.name, _fetch(component)) for component 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, err)
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, err
)
_LOGGER.debug("Prefetch failure detail", exc_info=err)
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
_LOGGER.warning("Library prefetch failed: %s", err)
_LOGGER.debug("Prefetch failure detail", exc_info=True)
def convert_libraries(
@@ -1105,8 +1042,7 @@ def convert_libraries(
worklist = deque(dict.fromkeys(top_level))
while worklist:
# Drain the frontier sequentially (spec resolution mutates shared
# node state), then prefetch the wave's registry archives in
# parallel; the per-component download() below stays authoritative.
# state), then prefetch the wave in parallel
wave: list[tuple[str, ConvertedLibrary]] = []
while worklist:
key = worklist.popleft()
@@ -1124,20 +1060,19 @@ def convert_libraries(
elif node.is_local:
component = ConvertedLibrary(key, "*", LocalSource(node.local_path))
else:
owner, name, version, url = _resolve_registry_version(
owner, name, version, url, size = _resolve_registry_version(
node.owner, node.pkgname, node.requirements
)
component = ConvertedLibrary(
_owner_pkgname_to_name(owner, name), version, URLSource(url)
_owner_pkgname_to_name(owner, name), version, URLSource(url, size)
)
wave.append((key, component))
_prefetch_wave(wave, salt, backend.cache_key)
for key, component in wave:
node = nodes[key]
if frozenset(node.requirements) != resolved_requirements[key]:
# An earlier wave entry grew this node's requirements after
# the drain resolved it; skip parsing and walking a manifest
# the next wave will replace (its archive is already fetched)
# Requirements grew mid-wave: skip parsing a manifest the
# next wave will re-resolve and replace
worklist.append(key)
continue
component.download(salt=salt, namespace=backend.cache_key)
+13 -5
View File
@@ -614,7 +614,7 @@ def _patch_registry(monkeypatch, versions):
def test_resolve_registry_version_intersects_constraints(monkeypatch):
_patch_registry(monkeypatch, ["1.10018.1", "1.10021.0", "1.10021.1"])
owner, name, version, url = _resolve_registry_version(
owner, name, version, url, _size = _resolve_registry_version(
"esphome", "libsodium", {"==1.10021.0", "^1.10018.1"}
)
assert (owner, name, version) == ("esphome", "libsodium", "1.10021.0")
@@ -623,7 +623,9 @@ def test_resolve_registry_version_intersects_constraints(monkeypatch):
def test_resolve_registry_version_picks_highest_satisfying(monkeypatch):
_patch_registry(monkeypatch, ["1.0.0", "1.5.0", "2.0.0"])
_owner, _name, version, _url = _resolve_registry_version("o", "p", {"^1.0.0"})
_owner, _name, version, _url, _size = _resolve_registry_version(
"o", "p", {"^1.0.0"}
)
assert version == "1.5.0"
@@ -673,7 +675,7 @@ def test_generate_idf_components_dedupes_shared_dependency(
resolve_calls.append(pkgname)
captured[f"{owner}/{pkgname}"] = set(requirements)
version = "1.10021.0" if pkgname == "C" else "1.0.0"
return owner, pkgname, version, f"http://x/{pkgname}.tar.gz"
return owner, pkgname, version, f"http://x/{pkgname}.tar.gz", None
monkeypatch.setattr(
esphome.platformio.library, "_resolve_registry_version", fake_resolve
@@ -732,7 +734,7 @@ def test_generate_idf_components_lib_ignore_filters_top_level_and_dependencies(
def fake_resolve(owner, pkgname, requirements):
resolve_calls.append(pkgname)
return owner, pkgname, "1.0.0", f"http://x/{pkgname}.tar.gz"
return owner, pkgname, "1.0.0", f"http://x/{pkgname}.tar.gz", None
monkeypatch.setattr(
esphome.platformio.library, "_resolve_registry_version", fake_resolve
@@ -788,6 +790,7 @@ def test_generate_idf_components_handles_dependency_cycle(
pkgname,
"1.0.0",
f"http://x/{pkgname}.tar.gz",
None,
),
)
@@ -845,6 +848,7 @@ def test_generate_idf_components_git_overrides_registry_warns(
pkgname,
"1.0.0",
f"http://x/{pkgname}.tar.gz",
None,
),
)
@@ -881,6 +885,7 @@ def test_generate_idf_components_missing_manifest_raises(
pkgname,
"1.0.0",
f"http://x/{pkgname}.tar.gz",
None,
),
)
@@ -925,6 +930,7 @@ def test_generate_idf_components_warns_on_noncanonical_duplicate(
pkgname,
"1.0.0",
f"http://x/{pkgname}.tar.gz",
None,
),
)
@@ -958,6 +964,7 @@ def test_generate_idf_components_incompatible_top_level_raises(
pkgname,
"1.0.0",
f"http://x/{pkgname}.tar.gz",
None,
),
)
@@ -994,6 +1001,7 @@ def test_generate_idf_components_incompatible_dependency_skipped(
pkgname,
"1.0.0",
f"http://x/{pkgname}.tar.gz",
None,
),
)
@@ -1066,7 +1074,7 @@ def test_idf_component_download_passes_salt() -> None:
c.download(force=True, salt="abcd1234", namespace="idf")
source.download.assert_called_once_with(
"owner/name", force=True, salt="abcd1234", namespace="idf", progress=None
"owner/name", force=True, salt="abcd1234", namespace="idf"
)
assert c.path == Path("/converted/owner/name")
+5 -26
View File
@@ -912,7 +912,7 @@ def test_prefetch_leaves_unverifiable_entries_to_the_installer(
),
patch("esphome.espidf.framework.download_with_resume") as download,
patch("esphome.espidf.framework.get_system_python_path", return_value="python"),
patch("esphome.espidf.framework.BatchDownloadProgress") as progress_cls,
patch("esphome.framework_helpers._BatchDownloadProgress") as progress_cls,
):
_prefetch_idf_tool_archives(tmp_path, "esp32", ["required"], None)
assert [call[0][0] for call in download.call_args_list] == [
@@ -953,7 +953,7 @@ def test_prefetch_dedupes_entries_by_dest(tmp_path: Path) -> None:
),
patch("esphome.espidf.framework.download_with_resume") as download,
patch("esphome.espidf.framework.get_system_python_path", return_value="python"),
patch("esphome.espidf.framework.BatchDownloadProgress"),
patch("esphome.framework_helpers._BatchDownloadProgress"),
):
_prefetch_idf_tool_archives(tmp_path, "esp32", ["required"], None)
dests = [call[0][1].name for call in download.call_args_list]
@@ -968,7 +968,7 @@ def test_prefetch_downloads_each_archive_with_resume(tmp_path: Path) -> None:
),
patch("esphome.espidf.framework.download_with_resume") as download,
patch("esphome.espidf.framework.get_system_python_path", return_value="python"),
patch("esphome.espidf.framework.BatchDownloadProgress") as progress_cls,
patch("esphome.framework_helpers._BatchDownloadProgress") as progress_cls,
):
# Materialize the lazy mock before threads race its first creation
tracker = progress_cls.return_value.tracker.return_value
@@ -1022,25 +1022,6 @@ def test_prefetch_downloads_archives_concurrently(tmp_path: Path) -> None:
assert download.call_count == 6
def test_prefetch_single_archive_uses_one_worker(tmp_path: Path) -> None:
entries = json.loads(_PREFETCH_JSON)[:1]
with (
patch(
"esphome.espidf.framework.run_command",
return_value=(True, json.dumps(entries), ""),
),
patch("esphome.espidf.framework.download_with_resume") as download,
patch("esphome.espidf.framework.get_system_python_path", return_value="python"),
patch(
"esphome.framework_helpers.ThreadPoolExecutor", wraps=ThreadPoolExecutor
) as pool,
):
_prefetch_idf_tool_archives(tmp_path, "esp32", ["required"], None)
pool.assert_called_once_with(max_workers=1)
assert download.call_count == 1
def test_prefetch_skips_already_downloaded_archives(tmp_path: Path) -> None:
dist = get_idf_tools_path() / "dist"
dist.mkdir(parents=True)
@@ -1130,10 +1111,8 @@ def test_prefetch_finishes_progress_bar_and_cancels_queue(tmp_path: Path) -> Non
),
patch("esphome.espidf.framework.download_with_resume"),
patch("esphome.espidf.framework.get_system_python_path", return_value="python"),
patch("esphome.espidf.framework.BatchDownloadProgress") as progress_cls,
patch(
"esphome.framework_helpers.ThreadPoolExecutor", wraps=ThreadPoolExecutor
) as pool_cls,
patch("esphome.framework_helpers._BatchDownloadProgress") as progress_cls,
patch("esphome.framework_helpers.ThreadPoolExecutor") as pool_cls,
):
pool = MagicMock(wraps=ThreadPoolExecutor(max_workers=2))
pool_cls.return_value = pool
+6 -2
View File
@@ -998,10 +998,10 @@ def test_progress_bar(capsys: CaptureFixture[str]) -> None:
assert "100%" in captured.err
assert "Done" in captured.err
# Test done method
# done() after the 100% frame adds nothing; that frame ended its line
progress.done()
captured = capsys.readouterr()
assert captured.err == "\n"
assert captured.err == ""
# Test same progress doesn't update
progress.update(0.5)
@@ -1010,6 +1010,10 @@ def test_progress_bar(capsys: CaptureFixture[str]) -> None:
# Should only see one update (second call shouldn't write)
assert captured.err.count("50%") == 1
# done() after a mid-way frame ends the line
progress.done()
assert capsys.readouterr().err == "\n"
# Tests for SHA256 authentication
@pytest.mark.usefixtures("mock_time")
+29 -15
View File
@@ -12,6 +12,8 @@ from pathlib import Path
import subprocess
import sys
import tarfile
import threading
import time
from unittest.mock import MagicMock, Mock, call, patch
import zipfile
@@ -21,8 +23,8 @@ import requests as req
from esphome import framework_helpers
from esphome.core import EsphomeError
from esphome.framework_helpers import (
BatchDownloadProgress,
_7z_extract_all,
_BatchDownloadProgress,
_detect_archive_root,
_is_transient_download_error,
_rename_with_retry,
@@ -38,6 +40,7 @@ from esphome.framework_helpers import (
get_python_env_executable_path,
get_system_python_path,
rmdir,
run_batch_downloads,
run_command,
run_command_ok,
str_to_lst_of_str,
@@ -1161,11 +1164,6 @@ class TestDownloadWithResume:
def test_run_batch_downloads_ctrl_c_aborts_in_flight_jobs() -> None:
"""Ctrl-C cancels in-flight downloads at their next tick instead of
letting non-daemon workers download to completion."""
import threading
import time
from esphome.framework_helpers import BatchDownloadProgress, run_batch_downloads
started = threading.Event()
ticks: list[int] = []
@@ -1183,8 +1181,8 @@ def test_run_batch_downloads_ctrl_c_aborts_in_flight_jobs() -> None:
t0 = time.monotonic()
with pytest.raises(KeyboardInterrupt):
run_batch_downloads(
BatchDownloadProgress("Downloading", 0),
[("boom", interrupter), ("slow", slow_download)],
"Downloading",
[("boom", 0, interrupter), ("slow", 0, slow_download)],
max_workers=2,
)
# Uncancelled, slow_download alone takes ~5s
@@ -1192,10 +1190,10 @@ def test_run_batch_downloads_ctrl_c_aborts_in_flight_jobs() -> None:
assert len(ticks) < 500
class TestBatchDownloadProgress:
class Test_BatchDownloadProgress:
def test_sums_trackers_into_one_bar(self) -> None:
with patch("esphome.framework_helpers.ProgressBar") as bar_cls:
progress = BatchDownloadProgress("Downloading", 100)
progress = _BatchDownloadProgress("Downloading", 100)
a = progress.tracker()
b = progress.tracker()
a(10)
@@ -1209,13 +1207,13 @@ class TestBatchDownloadProgress:
def test_clamps_at_one(self) -> None:
"""Sizes are advisory; an over-delivering server never pushes past 100%."""
with patch("esphome.framework_helpers.ProgressBar") as bar_cls:
progress = BatchDownloadProgress("Downloading", 10)
progress = _BatchDownloadProgress("Downloading", 10)
progress.tracker()(25)
assert bar_cls.return_value.update.call_args[0][0] == 1
def test_unknown_total_draws_nothing(self) -> None:
with patch("esphome.framework_helpers.ProgressBar") as bar_cls:
progress = BatchDownloadProgress("Downloading", 0)
progress = _BatchDownloadProgress("Downloading", 0)
progress.tracker()(5)
progress.done()
bar_cls.assert_not_called()
@@ -1226,7 +1224,7 @@ class TestBatchDownloadProgress:
stream = io.StringIO()
stream.isatty = lambda: True # type: ignore[method-assign]
with patch("esphome.helpers.sys.stderr", stream):
progress = BatchDownloadProgress("Downloading", 10)
progress = _BatchDownloadProgress("Downloading", 10)
progress.tracker()(5)
progress.done()
assert stream.getvalue().endswith("50% \n")
@@ -1237,14 +1235,14 @@ class TestBatchDownloadProgress:
stream = io.StringIO()
stream.isatty = lambda: True # type: ignore[method-assign]
with patch("esphome.helpers.sys.stderr", stream):
BatchDownloadProgress("Downloading", 10).done()
_BatchDownloadProgress("Downloading", 10).done()
assert stream.getvalue() == ""
def test_done_after_full_bar_adds_nothing(self) -> None:
stream = io.StringIO()
stream.isatty = lambda: True # type: ignore[method-assign]
with patch("esphome.helpers.sys.stderr", stream):
progress = BatchDownloadProgress("Downloading", 10)
progress = _BatchDownloadProgress("Downloading", 10)
progress.tracker()(10)
progress.done()
assert stream.getvalue().endswith("100% Done...\r\n")
@@ -1261,6 +1259,22 @@ class TestDownloadFromMirrors:
assert url == "https://example.com/f"
assert target.read_bytes() == b"filedata"
def test_file_object_target_reports_progress(self) -> None:
"""The library prefetch's production path: a file-object target
streams through the mirror fallback and ticks the tracker."""
buf = io.BytesIO()
ticks: list[int] = []
with patch(
"requests.get",
return_value=_mock_response(b"filedata"),
):
url = download_from_mirrors(
["https://example.com/f"], {}, buf, progress=ticks.append
)
assert url == "https://example.com/f"
assert buf.getvalue() == b"filedata"
assert ticks and ticks[-1] == len(b"filedata")
def test_substitutions_applied_to_url(self, tmp_path: Path) -> None:
with patch(
"requests.get",
+43 -67
View File
@@ -8,7 +8,6 @@ from contextlib import contextmanager
import json
import logging
from pathlib import Path
from types import SimpleNamespace
import pytest
@@ -236,6 +235,7 @@ def _patch_registry_resolve(monkeypatch: pytest.MonkeyPatch) -> None:
pkgname,
"1.0.0",
f"http://x/{pkgname}.tar.gz",
None,
),
)
@@ -276,8 +276,7 @@ def test_wave_requirement_growth_defers_the_superseded_download(tmp_path, monkey
(self.path / "library.json").write_text(json.dumps(manifests[self.name]))
monkeypatch.setattr(ConvertedLibrary, "download", fake_download)
# Hermetic: unknown sizes take the sequential path instead of real HEADs
monkeypatch.setattr(lib, "_content_lengths", lambda urls: [None] * len(urls))
# Hermetic: the stubbed registry reports no size, so no batch prefetch
_patch_registry_resolve(monkeypatch)
top = convert_libraries(
[Library("esphome/A", "1.0.0", None), Library("esphome/B", None, None)],
@@ -640,22 +639,23 @@ def test_prefetch_wave_downloads_registry_archives_in_parallel(
git/local sources and failures are left to the sequential call."""
calls: list[str] = []
def fake_download(self, force=False, salt="", namespace="", progress=None):
calls.append(self.source.url)
def fake_download(
self, dir_suffix, force=False, salt="", namespace="", progress=None
):
calls.append(self.url)
if progress is not None:
progress(0)
if "boom" in self.source.url:
if "boom" in self.url:
raise RuntimeError("boom")
monkeypatch.setattr(ConvertedLibrary, "download", fake_download)
monkeypatch.setattr(lib, "_content_lengths", lambda urls: [1] * len(urls))
monkeypatch.setattr(URLSource, "download", fake_download)
wave = [
("a", ConvertedLibrary("a", "1.0", URLSource("https://x/a.tar.gz"))),
("b", ConvertedLibrary("b", "1.0", URLSource("https://x/b.tar.gz"))),
("a", ConvertedLibrary("a", "1.0", URLSource("https://x/a.tar.gz", 1))),
("b", ConvertedLibrary("b", "1.0", URLSource("https://x/b.tar.gz", 1))),
# Duplicate URL must prefetch once (two threads must never extract
# into the same cache directory)
("b2", ConvertedLibrary("b2", "1.0", URLSource("https://x/b.tar.gz"))),
("c", ConvertedLibrary("c", "1.0", URLSource("https://x/boom.tar.gz"))),
("b2", ConvertedLibrary("b2", "1.0", URLSource("https://x/b.tar.gz", 1))),
("c", ConvertedLibrary("c", "1.0", URLSource("https://x/boom.tar.gz", 1))),
("g", ConvertedLibrary("g", "*", lib.GitSource("https://x/g.git", None))),
]
lib._prefetch_wave(wave, "", "idf")
@@ -668,25 +668,27 @@ def test_prefetch_wave_downloads_registry_archives_in_parallel(
assert "Prefetch of c failed (retrying sequentially)" in caplog.text
def test_prefetch_wave_unknown_size_falls_back_to_sequential(
setup_core, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
def test_prefetch_wave_unknown_size_left_to_sequential(
setup_core, monkeypatch: pytest.MonkeyPatch
) -> None:
"""Any unknown HEAD size skips the parallel prefetch entirely so the
sequential downloads keep their per-file bars."""
def fail_download(self, force=False, salt="", namespace="", progress=None):
raise AssertionError("prefetched despite unknown size")
monkeypatch.setattr(ConvertedLibrary, "download", fail_download)
monkeypatch.setattr(lib, "_content_lengths", lambda urls: [1, None])
"""Archives without a registry-reported size skip the batch (their
sequential per-file bars don't interleave); the known subset still
prefetches."""
calls: list[str] = []
monkeypatch.setattr(
URLSource,
"download",
lambda self, dir_suffix, force=False, salt="", namespace="", progress=None: (
calls.append(self.url)
),
)
wave = [
("a", ConvertedLibrary("a", "1.0", URLSource("https://x/a.tar.gz"))),
("b", ConvertedLibrary("b", "1.0", URLSource("https://x/b.tar.gz"))),
("a", ConvertedLibrary("a", "1.0", URLSource("https://x/a.tar.gz", 1))),
("b", ConvertedLibrary("b", "1.0", URLSource("https://x/b.tar.gz", 1))),
("u", ConvertedLibrary("u", "1.0", URLSource("https://x/u.tar.gz"))),
]
caplog.set_level("DEBUG")
lib._prefetch_wave(wave, "", "idf")
# The culprit URL is named so the fallback is traceable
assert "No Content-Length for https://x/b.tar.gz" in caplog.text
assert sorted(calls) == ["https://x/a.tar.gz", "https://x/b.tar.gz"]
def test_join_flag_args_empty_argument_warns_and_drops(
@@ -697,54 +699,24 @@ def test_join_flag_args_empty_argument_warns_and_drops(
assert "Ignoring '-D' with empty argument in build_flags" in caplog.text
def test_content_lengths_head_requests(monkeypatch: pytest.MonkeyPatch) -> None:
"""Sizes come from HEAD Content-Length; a failing HEAD reads as 0 so
the combined bar is skipped rather than wrong."""
import requests
def fake_head(url, timeout, allow_redirects):
if "bad" in url:
raise requests.ConnectionError("down")
if "gone" in url:
return SimpleNamespace(ok=False, status_code=404, headers={})
if "garbage" in url:
# A proxy/CDN doubling the header ("123, 123") or emitting junk
# must degrade to unknown, not ValueError the build
return SimpleNamespace(ok=True, headers={"content-length": "123, 123"})
return SimpleNamespace(ok=True, headers={"content-length": "123"})
monkeypatch.setattr(requests, "head", fake_head)
# None marks an unknown size (probe failure or non-2xx), distinct
# from a genuine zero
assert lib._content_lengths(
["https://x/a", "https://x/bad", "https://x/gone", "https://x/garbage"]
) == [
123,
None,
None,
None,
]
def test_prefetch_wave_cache_probe_failure_still_prefetches(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""The cache probe is best-effort; a failing probe prefetches anyway."""
calls: list[str] = []
monkeypatch.setattr(
ConvertedLibrary,
URLSource,
"download",
lambda self, **kw: calls.append(self.source.url),
lambda self, dir_suffix, **kw: calls.append(self.url),
)
monkeypatch.setattr(
URLSource,
"is_cached",
lambda self, *a, **kw: (_ for _ in ()).throw(RuntimeError("no core")),
)
monkeypatch.setattr(lib, "_content_lengths", lambda urls: [1] * len(urls))
wave = [
("a", ConvertedLibrary("a", "1.0", URLSource("https://x/a.tar.gz"))),
("b", ConvertedLibrary("b", "1.0", URLSource("https://x/b.tar.gz"))),
("a", ConvertedLibrary("a", "1.0", URLSource("https://x/a.tar.gz", 1))),
("b", ConvertedLibrary("b", "1.0", URLSource("https://x/b.tar.gz", 1))),
]
lib._prefetch_wave(wave, "", "idf")
assert sorted(calls) == ["https://x/a.tar.gz", "https://x/b.tar.gz"]
@@ -756,13 +728,15 @@ def test_prefetch_wave_warm_cache_is_silent(
"""Already-extracted archives download nothing; a warm build must not
print a Downloading line or draw a bar."""
monkeypatch.setattr(
ConvertedLibrary,
URLSource,
"download",
lambda self, **kw: (_ for _ in ()).throw(AssertionError("downloaded")),
lambda self, dir_suffix, **kw: (_ for _ in ()).throw(
AssertionError("downloaded")
),
)
wave = []
for name in ("a", "b", "c"):
comp = ConvertedLibrary(name, "1.0", URLSource(f"https://x/{name}.tar.gz"))
comp = ConvertedLibrary(name, "1.0", URLSource(f"https://x/{name}.tar.gz", 1))
marker_dir = comp.source._cache_dir(comp.get_sanitized_name(), "", "idf")
marker_dir.mkdir(parents=True)
(marker_dir / ".esphome_extracted").touch()
@@ -777,12 +751,14 @@ def test_prefetch_wave_single_archive_skips_the_pool(
"""One archive gains nothing from a pool; the sequential call keeps its
progress bar."""
monkeypatch.setattr(
ConvertedLibrary,
URLSource,
"download",
lambda self, **kw: (_ for _ in ()).throw(AssertionError("prefetched")),
lambda self, dir_suffix, **kw: (_ for _ in ()).throw(
AssertionError("prefetched")
),
)
lib._prefetch_wave(
[("a", ConvertedLibrary("a", "1.0", URLSource("https://x/a.tar.gz")))],
[("a", ConvertedLibrary("a", "1.0", URLSource("https://x/a.tar.gz", 1)))],
"",
"idf",
)