Size the prefetch from the registry, fold the batch API, guard the prefetch

The registry response already carries each package file's size, so the
combined bar is sized from URLSource instead of a HEAD sweep; the probe,
its all-or-nothing fallback, and its duplicate Content-Length parsing are
gone, and archives without a reported size simply stay on the sequential
path. run_batch_downloads now takes (name, size, fetch) jobs and builds
the (now private) progress bar itself, collecting failures from future
results instead of a shared list; both call sites use functools.partial
instead of lambda factories. The prefetch touches only URLSource, so the
progress parameter comes off Source/Git/Local/ConvertedLibrary, and the
wave prefetch is wrapped best-effort like its ESP-IDF twin, which also
dedupes by dest only after the verifiability check. ProgressBar.done()
itself now skips the newline when no frame was drawn or the 100% frame
already ended the line, collapsing the batch wrapper's guard and fixing
the stray blank line after OTA uploads.
This commit is contained in:
J. Nick Koston
2026-08-23 13:08:32 -05:00
parent caff71c17a
commit 9d761102e5
9 changed files with 228 additions and 279 deletions
+35 -24
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
@@ -15,7 +16,6 @@ import platformdirs
from esphome.core import CORE, Version
from esphome.framework_helpers import (
BatchDownloadProgress,
PathType,
archive_extract_all,
create_venv,
@@ -692,6 +692,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,
@@ -734,21 +746,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"])
# Never download unverified: an entry without sha256/size is
# left to the installer, which fails loudly on a bad archive
if entry.get("sha256") and entry.get("size"):
entries.append(entry)
else:
# 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(
@@ -758,26 +772,23 @@ def _prefetch_idf_tool_archives(
)
# No sequential fallback here: skipping the prefetch would lose the
# resume workaround for #17703, and every entry has a size (above)
def _download(entry: dict):
return lambda tracker: download_with_resume(
entry["url"],
dist_path / entry["dest"],
sha256=entry["sha256"],
size=entry["size"],
progress=tracker,
)
# 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.
+20 -22
View File
@@ -743,21 +743,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 ``(name, fetch)`` download jobs concurrently under one combined bar.
"""Run ``(name, size, fetch)`` download jobs concurrently under one bar.
Each ``fetch(tracker)`` reports absolute byte counts. 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;
``.part`` files keep the fetched bytes. ``jobs`` must be non-empty.
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:
@@ -770,13 +774,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
@@ -785,14 +790,13 @@ 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:
class _BatchDownloadProgress:
"""One bar across several concurrent downloads, summing tracker bytes.
The lock also serialises stderr writes so workers never interleave
@@ -821,13 +825,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()
+2 -1
View File
@@ -729,7 +729,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()
+75 -117
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,
@@ -84,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
@@ -103,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
@@ -166,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:
@@ -206,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():
@@ -301,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
@@ -316,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)
@@ -578,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
@@ -610,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]:
@@ -896,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(
@@ -927,64 +892,57 @@ def _prefetch_wave(
"""Best-effort parallel download of a wave's registry archives.
The walk's own ``download()`` stays authoritative; duplicate URLs
prefetch once so two threads never share a cache directory.
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
# Combined bar sized by HEAD requests; unknown sizes fall back to
# sequential per-file bars rather than a silent multi-MB download
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(
@@ -1102,11 +1060,11 @@ 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)
@@ -1114,7 +1072,7 @@ def convert_libraries(
node = nodes[key]
if frozenset(node.requirements) != resolved_requirements[key]:
# Requirements grew mid-wave: skip parsing a manifest the
# next wave will replace (its archive is already fetched)
# 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",
)