github/esphome
2
0
Fork 0
mirror of https://github.com/esphome/esphome.git synced 2026-08-29 09:13:28 +00:00
Code Issues Packages Projects Releases Wiki Activity

[espidf] Resume interrupted toolchain downloads instead of restarting (#17706)

Browse Source
This commit is contained in:
J. Nick Koston
2026-07-21 08:12:17 +12:00
committed by GitHub
parent 9a0e8606cd
commit 271c1b409a
6 changed files with 1844 additions and 159 deletions
Download Patch File Download Diff File Expand all files Collapse all files
esphome/espidf/framework.py
+148 -65
View File
@@ -9,7 +9,7 @@ from pathlib import Path
import platform
import re
import shutil
import tempfile
from typing import NoReturn
import platformdirs
@@ -20,6 +20,7 @@ from esphome.framework_helpers import (
archive_extract_all,
create_venv,
download_from_mirrors,
download_with_resume,
get_python_env_executable_path,
get_system_python_path,
rmdir,
@@ -231,6 +232,40 @@ def _write_stamp(file: PathType, data: dict[str, str]):
json.dump(data, fp)
def _run_idf_tools_script(
idf_framework_root: PathType,
script_name: str,
msg: str,
args: list[str] | None = None,
env: dict[str, str] | None = None,
) -> tuple[bool, str | None, str | None]:
"""Run one of the sibling idf_tools-backed helper scripts.
The script is executed with the framework's ``tools`` directory on
PYTHONPATH so it imports the framework's own ``idf_tools`` module.
"""
cmd = [
get_system_python_path(),
str(_SCRIPTS_DIR / script_name),
str(idf_framework_root),
*(args or []),
]
return run_command(
cmd,
msg=msg,
env=(env or os.environ)
| {"PYTHONPATH": str(Path(idf_framework_root) / "tools")},
)
def _raise_script_failure(what: str, root: PathType, stderr: str | None) -> NoReturn:
"""Raise RuntimeError for a failed helper script, appending stderr detail."""
detail = (stderr or "").strip()
raise RuntimeError(
f"Can't get {what} of {root}" + (f": {detail}" if detail else "")
)
def _get_idf_version(
idf_framework_root: PathType, env: dict[str, str] | None = None
) -> str:
@@ -248,26 +283,13 @@ def _get_idf_version(
RuntimeError: If ESP-IDF version cannot be determined
"""
cmd = [
get_system_python_path(),
str(_SCRIPTS_DIR / "get_idf_version.py"),
str(idf_framework_root),
]
success, stdout, stderr = run_command(
cmd,
msg="ESP-IDF version",
env=(env or os.environ)
| {"PYTHONPATH": str(Path(idf_framework_root) / "tools")},
success, stdout, stderr = _run_idf_tools_script(
idf_framework_root, "get_idf_version.py", "ESP-IDF version", env=env
)
if stdout:
stdout = stdout.strip()
if not success or not stdout:
detail = (stderr or "").strip()
raise RuntimeError(
f"Can't get ESP-IDF version of {idf_framework_root}"
+ (f": {detail}" if detail else "")
)
_raise_script_failure("ESP-IDF version", idf_framework_root, stderr)
return stdout
@@ -288,24 +310,11 @@ def _get_idf_tool_paths(
RuntimeError: If ESP-IDF tool paths cannot be determined
"""
cmd = [
get_system_python_path(),
str(_SCRIPTS_DIR / "get_idf_tool_paths.py"),
str(idf_framework_root),
]
success, stdout, stderr = run_command(
cmd,
msg="ESP-IDF tool paths",
env=(env or os.environ)
| {"PYTHONPATH": str(Path(idf_framework_root) / "tools")},
success, stdout, stderr = _run_idf_tools_script(
idf_framework_root, "get_idf_tool_paths.py", "ESP-IDF tool paths", env=env
)
if not success or not stdout:
detail = (stderr or "").strip()
raise RuntimeError(
f"Can't get ESP-IDF tool paths of {idf_framework_root}"
+ (f": {detail}" if detail else "")
)
_raise_script_failure("ESP-IDF tool paths", idf_framework_root, stderr)
# Extract json values
try:
@@ -579,6 +588,69 @@ def _patch_tools_json_demote_openocd(framework_path: Path) -> None:
)
def _prefetch_idf_tool_archives(
framework_path: Path,
targets_str: str,
tools: list[str],
env: dict[str, str] | None,
) -> None:
"""Pre-download the tool archives ``idf_tools.py install`` would fetch.
``idf_tools.py``'s own downloader restarts from byte zero on every retry,
which makes large archives effectively impossible to fetch on unstable
connections (#17703). This asks the framework's idf_tools (via
``get_tool_downloads.py``) which archives the coming install needs, then
downloads each into ``<IDF_TOOLS_PATH>/dist`` with
``download_with_resume``. The installer then finds the verified archives
already in place ("file ... is already downloaded") and never touches the
network.
Strictly best-effort: any failure here just logs and returns, leaving
``idf_tools.py install`` to download whatever is missing exactly as
before. Leftover ``.part`` files live in ``dist/`` and are removed by the
post-install cache prune.
"""
try:
success, stdout, stderr = _run_idf_tools_script(
framework_path,
"get_tool_downloads.py",
"ESP-IDF tool download list",
args=[targets_str, *tools],
env=env,
)
if not success or not stdout:
_LOGGER.warning(
"Could not determine ESP-IDF tool downloads: %s",
(stderr or "").strip(),
)
return
dist_path = get_idf_tools_path() / "dist"
entries = [
entry
for entry in json.loads(stdout)
if not (dist_path / entry["dest"]).is_file()
]
for index, entry in enumerate(entries, start=1):
_LOGGER.info(
"Downloading %s (%d/%d) ...", entry["name"], index, len(entries)
)
try:
download_with_resume(
entry["url"],
dist_path / entry["dest"],
sha256=entry["sha256"],
size=entry["size"],
)
except Exception as e: # noqa: BLE001 # pylint: disable=broad-exception-caught
# Keep prefetching the remaining archives; the installer
# will retry this one itself (without resume).
_LOGGER.warning("Could not prefetch %s: %s", entry["name"], 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.
_LOGGER.warning("ESP-IDF tool prefetch failed: %s", e)
def _check_esphome_idf_framework_install(
version: str,
targets: list[str],
@@ -650,41 +722,51 @@ def _check_esphome_idf_framework_install(
git_url, ref = git_source
_clone_idf_with_submodules(framework_path, git_url, ref)
else:
# Download in temporary file
with tempfile.NamedTemporaryFile() as tmp:
_LOGGER.info("Downloading ESP-IDF %s framework ...", version)
_LOGGER.info("Downloading ESP-IDF %s framework ...", version)
# Create substitutions for the URLs. SHORT_VERSION (x.y with
# optional -extra) is only provided for x.y.0 releases, since
# the vX.Y release tags only exist for those; templates that
# reference it are skipped for other versions by
# download_from_mirrors.
substitutions = {"VERSION": version}
try:
ver = Version.parse(version)
substitutions["MAJOR"] = str(ver.major)
substitutions["MINOR"] = str(ver.minor)
substitutions["PATCH"] = str(ver.patch)
substitutions["EXTRA"] = f"-{ver.extra}" if ver.extra else ""
if ver.patch == 0:
substitutions["SHORT_VERSION"] = (
f"{ver.major}.{ver.minor}{substitutions['EXTRA']}"
)
except ValueError:
_LOGGER.warning(
"ESP-IDF version '%s' is not a valid version number; "
"only the {VERSION} substitution is available for "
"mirror URLs",
version,
# Create substitutions for the URLs. SHORT_VERSION (x.y with
# optional -extra) is only provided for x.y.0 releases, since
# the vX.Y release tags only exist for those; templates that
# reference it are skipped for other versions by
# download_from_mirrors.
substitutions = {"VERSION": version}
try:
ver = Version.parse(version)
substitutions["MAJOR"] = str(ver.major)
substitutions["MINOR"] = str(ver.minor)
substitutions["PATCH"] = str(ver.patch)
substitutions["EXTRA"] = f"-{ver.extra}" if ver.extra else ""
if ver.patch == 0:
substitutions["SHORT_VERSION"] = (
f"{ver.major}.{ver.minor}{substitutions['EXTRA']}"
)
mirrors = [source_url] if source_url else ESPHOME_IDF_FRAMEWORK_MIRRORS
download_from_mirrors(mirrors, substitutions, tmp.file)
_LOGGER.info("Extracting ESP-IDF %s framework ...", version)
archive_extract_all(
tmp.file, framework_path, progress_header="Extracting"
except ValueError:
_LOGGER.warning(
"ESP-IDF version '%s' is not a valid version number; "
"only the {VERSION} substitution is available for "
"mirror URLs",
version,
)
mirrors = [source_url] if source_url else ESPHOME_IDF_FRAMEWORK_MIRRORS
# Download to a persistent file in the tool download cache (not
# a temp file) so an interrupted download resumes on the next
# run; the cache is pruned after a successful install anyway.
tarball_path = get_idf_tools_path() / "dist" / f"esp-idf-{version}.tar.xz"
download_from_mirrors(mirrors, substitutions, tarball_path)
_LOGGER.info("Extracting ESP-IDF %s framework ...", version)
try:
with tarball_path.open("rb") as tarball:
archive_extract_all(
tarball, framework_path, progress_header="Extracting"
)
finally:
# Success: drop the archive rather than caching ~70MB twice.
# Failure: a corrupt archive (e.g. torn by an unclean
# shutdown) must not be reused — without a checksum only a
# failed extraction can expose it, so force a re-download.
tarball_path.unlink(missing_ok=True)
extracted_marker.touch()
# Idempotent post-extract patch: written every invocation so a build
@@ -722,6 +804,7 @@ def _check_esphome_idf_framework_install(
if install:
_LOGGER.info("Installing ESP-IDF %s framework ...", version)
targets_str = ",".join(targets)
_prefetch_idf_tool_archives(framework_path, targets_str, tools, env)
cmd = [
get_system_python_path(),
str(idf_tools_path),
esphome/espidf/get_tool_downloads.py
+88
View File
@@ -0,0 +1,88 @@
"""Print JSON download info for the ESP-IDF tools an install would fetch.
Run via ``python <this file> <idf_framework_root> <targets-csv> <tool-spec>...``.
PYTHONPATH must include ``<idf_framework_root>/tools`` so ``idf_tools`` is
importable, and IDF_TOOLS_PATH must be set. Prints a JSON list of
``{name, url, size, sha256, dest}`` for every tool version that is not yet
installed, where ``dest`` is the archive filename ``idf_tools.py install``
expects to find in ``<IDF_TOOLS_PATH>/dist``. Tools with no download for the
current platform are skipped; already-installed versions are skipped so a
pruned download cache is not re-fetched.
The target/tool expansion mirrors ``idf_tools.py install`` (targets passed to
``add_and_check_targets`` accumulate with idf-env.json) but nothing is saved
or written — this script only reports what the install would download.
"""
# pylint: disable=import-error # idf_tools is on PYTHONPATH at runtime only
from contextlib import redirect_stdout
import json
import os
from pathlib import Path
import sys
from idf_tools import (
CURRENT_PLATFORM,
TOOLS_FILE,
IDFEnv,
ToolBinaryError,
add_and_check_targets,
expand_tools_arg,
g,
get_idf_download_url_apply_mirrors,
load_tools_info,
)
def collect_downloads() -> list[dict]:
g.idf_path = sys.argv[1]
g.idf_tools_path = os.environ.get("IDF_TOOLS_PATH")
g.tools_json = str(Path(g.idf_path) / TOOLS_FILE)
targets = add_and_check_targets(IDFEnv.get_idf_env(), sys.argv[2])
tools_info = load_tools_info()
downloads: list[dict] = []
for name in expand_tools_arg(sys.argv[3:], tools_info, targets):
if "@" in name:
name, version = name.split("@", 1)
else:
version = None
tool = tools_info.get(name)
if tool is None or not tool.compatible_with_platform():
continue
version = version or tool.get_recommended_version()
if version is None:
continue
try:
tool.find_installed_versions()
except ToolBinaryError as e:
# A broken installed binary is idf_tools' problem to repair on
# install; note it and treat the version as not installed.
print(f"tool {name} failed its binary check: {e}", file=sys.stderr)
if version in tool.versions_installed or version not in tool.versions:
continue
download = tool.versions[version].get_download_for_platform(CURRENT_PLATFORM)
if download is None:
continue
downloads.append(
{
"name": f"{name}@{version}",
# Apply the same IDF_MIRROR_PREFIX_MAP / IDF_GITHUB_ASSETS
# rewriting the installer's own downloader applies, so users
# behind a mirror prefetch from the mirror too.
"url": get_idf_download_url_apply_mirrors(None, download.url),
"size": download.size,
"sha256": download.sha256,
"dest": download.rename_dist or Path(download.url).name,
}
)
return downloads
# idf_tools prints informational lines (e.g. mirror URL rewrites) to stdout;
# route them to stderr so stdout carries only the JSON result.
with redirect_stdout(sys.stderr):
result = collect_downloads()
print(json.dumps(result))
esphome/framework_helpers.py
+462 -85
View File
@@ -2,21 +2,31 @@
from collections.abc import Iterable
from contextlib import ExitStack
import hashlib
import io
import json
import logging
import os
from pathlib import Path
import subprocess
import sys
import time
from typing import IO
from typing import IO, TYPE_CHECKING
from esphome.helpers import ProgressBar, rmtree
if TYPE_CHECKING:
import requests
PathType = str | os.PathLike
_LOGGER = logging.getLogger(__name__)
# Attempts per mirror URL before falling through to the next mirror; only
# mid-stream drops retry (resuming when the server gave a validator),
# connect errors move on immediately.
_MIRROR_ATTEMPTS = 3
def get_project_link_flags() -> list[str]:
"""Return the sorted -Wl, linker flags from the current build."""
@@ -394,17 +404,23 @@ def _zip_extract_all(
progress.update(1)
def _rename_with_retry(src: Path, dst: Path, attempts: int = 5) -> None:
def _rename_with_retry(
src: Path, dst: Path, attempts: int = 5, overwrite: bool = False
) -> None:
"""Rename ``src`` to ``dst`` with backoff retries on Windows sharing violations.
Antivirus/indexer handles on freshly-written files can briefly block
``os.rename`` with ERROR_SHARING_VIOLATION / ERROR_ACCESS_DENIED. The
handle is released within tens of ms in practice, so exponential backoff
works.
works. With ``overwrite`` an existing ``dst`` is replaced instead of
failing.
"""
for i in range(attempts):
try:
src.rename(dst)
if overwrite:
src.replace(dst)
else:
src.rename(dst)
return
except PermissionError:
if i == attempts - 1:
@@ -525,8 +541,8 @@ def archive_extract_all(
ValueError: If archive format is unsupported
"""
# 1. Handle different archive input types
with ExitStack() as stack:
# 1. Handle different archive input types
archive_ref: io.BufferedIOBase
if isinstance(archive, (str, os.PathLike)):
archive_ref = stack.enter_context(Path(archive).open("rb"))
@@ -552,6 +568,311 @@ def archive_extract_all(
matched_fct(archive_ref, extract_dir, progress_header=progress_header)
def _open_ranged(
url: str, offset: int, timeout: int, validator: str | None = None
) -> tuple["requests.Response | None", int]:
"""Open a streaming GET, asking the server to resume at ``offset``.
``validator`` is an ETag or Last-Modified value from the interrupted
response; it is sent as ``If-Range`` so the server only honors the Range
when the content is unchanged, replying 200 (full body, restart) if the
file was replaced between requests — the resumed bytes can then never be
stitched onto a different file's prefix.
Returns ``(response, effective_offset)``. The response is None when the
server answered 416 Range Not Satisfiable: the file holds every byte the
server has (a previous attempt was interrupted after the last byte), so
there is nothing to stream and the caller's verification decides whether
the file is good. The offset drops to 0 when the server ignored the
``Range`` header (no 206), meaning the caller must restart the file.
Raises on connect errors and HTTP error statuses; the response is closed
on failure.
"""
import requests
headers = {"Range": f"bytes={offset}-"} if offset else {}
if offset and validator:
headers["If-Range"] = validator
resp = requests.get(url, stream=True, timeout=timeout, headers=headers)
if offset and resp.status_code == 416:
resp.close()
return None, offset
if offset and resp.status_code != 206:
_LOGGER.debug(
"Server did not resume %s (HTTP %s), restarting", url, resp.status_code
)
offset = 0
if not resp.ok:
resp.close()
resp.raise_for_status()
if offset:
_LOGGER.info("Resuming download at %d bytes ...", offset)
return resp, offset
def _verify_file(path: Path, sha256: str | None, size: int | None) -> None:
"""Raise EsphomeError when ``path`` fails an available sha256/size check."""
from esphome.core import EsphomeError
if size is not None and path.stat().st_size != size:
raise EsphomeError(f"size mismatch: expected {size}, got {path.stat().st_size}")
if sha256 is not None:
with path.open("rb") as f:
digest = hashlib.file_digest(f, "sha256").hexdigest()
if digest != sha256:
raise EsphomeError(f"sha256 mismatch: got {digest}")
def _load_download_meta(meta: Path, url: str) -> tuple[str | None, int]:
"""Return the ``(validator, total)`` a previous run recorded for ``url``.
``(None, 0)`` when there is no sidecar, it is unreadable, or it belongs
to a different URL (e.g. a different mirror was tried last time).
"""
try:
with meta.open(encoding="utf-8") as f:
data = json.load(f)
except (OSError, json.JSONDecodeError):
return None, 0
if not isinstance(data, dict) or data.get("url") != url:
return None, 0
validator = data.get("validator")
total = data.get("total")
return (
validator if isinstance(validator, str) else None,
total if isinstance(total, int) else 0,
)
def _write_download_meta(
meta: Path, url: str, validator: str | None, total: int
) -> None:
"""Persist resume metadata next to the part file; best-effort.
Without a validator there is nothing a later run could resume against,
so any stale sidecar is removed instead.
"""
try:
if validator is None:
meta.unlink(missing_ok=True)
else:
meta.write_text(
json.dumps({"url": url, "validator": validator, "total": total}),
encoding="utf-8",
)
except OSError as e:
_LOGGER.debug("Could not update download metadata %s: %s", meta, e)
def _content_length(resp: "requests.Response") -> int:
"""Return the response's Content-Length, or 0 when absent or malformed.
0 means "unknown", which downstream disables the progress bar and the
resume/completeness logic — a garbage header from a broken proxy must
degrade to a plain single-stream download, not crash the attempt.
"""
try:
return int(resp.headers.get("content-length", 0))
except ValueError:
return 0
def _response_validator(resp: "requests.Response") -> str | None:
"""Return the response's strong validator for ``If-Range`` resumes.
Weak ETags (``W/...``) are not usable for byte-range conditionals, so
fall back to Last-Modified, or None when the server offers neither.
"""
etag = resp.headers.get("ETag")
if etag and not etag.startswith("W/"):
return etag
return resp.headers.get("Last-Modified")
def _stream_response_to_file(
resp: "requests.Response", f: IO[bytes], offset: int, size: int | None = None
) -> None:
"""Stream an open ``_open_ranged`` response body into ``f`` at ``offset``.
Truncates ``f`` to ``offset`` first, so a server-rejected resume
(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.
"""
f.seek(offset)
f.truncate(offset)
total_size = size or offset + _content_length(resp)
downloaded = offset
progress = ProgressBar("Downloading") if total_size > 0 else None
for chunk in resp.iter_content(chunk_size=256 * 1024):
if chunk:
f.write(chunk)
downloaded += len(chunk)
if progress is not None:
progress.update(downloaded / total_size)
if progress is not None:
progress.update(1)
def download_with_resume(
url: str,
dest: PathType,
sha256: str | None = None,
size: int | None = None,
# More attempts than _MIRROR_ATTEMPTS: a single-URL download has no
# mirror fallback, and each retry only re-fetches the remainder.
attempts: int = 5,
timeout: int = 30,
retry_connect_errors: bool = True,
) -> None:
"""Download ``url`` to ``dest``, resuming partial downloads.
The body streams into ``<dest>.part``, which persists across attempts and
esphome runs: a mid-stream connection drop only costs one attempt and the
next continues from where it stopped, so an unstable connection converges
on a complete file instead of restarting from zero each retry (#17703).
When ``size`` / ``sha256`` are given the completed file is verified and a
mismatch restarts from scratch; success renames the part file into place.
An already-present ``dest`` that passes verification is kept as-is.
Resuming a part file from an earlier run needs proof the content is
unchanged: ``sha256`` when the caller has one, or otherwise the server's
If-Range validator recorded in a ``<dest>.part.meta`` sidecar by the run
that started the download — a size alone cannot detect a same-length
content change on the server.
With ``retry_connect_errors`` disabled, a failure before any body bytes
flow (connect error, HTTP error status) propagates immediately instead
of consuming attempts — for callers with their own fallback, like
``download_from_mirrors``.
Raises EsphomeError when all attempts are exhausted.
"""
# Imported lazily: requests is a heavy import (~85ms) and is only needed
# when actually downloading a toolchain, never during config validation.
import requests
from esphome.core import EsphomeError
dest = Path(dest)
part = dest.with_name(dest.name + ".part")
meta = part.with_name(part.name + ".meta")
dest.parent.mkdir(parents=True, exist_ok=True)
last_error: Exception | None = None
# An earlier run already completed this download. Only trust it when
# there is something to verify it against; without sha/size the remote
# content may have changed (e.g. a refreshed constraints file), so
# re-download and atomically replace it.
if dest.is_file() and (sha256 is not None or size is not None):
try:
_verify_file(dest, sha256, size)
return
except EsphomeError:
dest.unlink()
# Adopt the validator/total the run that started this part file recorded,
# so an unfinished download resumes across runs even without a sha256.
validator, expected_total = _load_download_meta(meta, url)
for _ in range(attempts):
streamed = False
try:
offset = part.stat().st_size if part.is_file() else 0
# A stitched resume needs two proofs: content identity (the
# bytes being appended belong to the same file as the prefix)
# and completeness. sha256 provides both, across runs. Without
# it, identity needs this run's If-Range validator — a size
# alone cannot detect a same-length content change, so a
# leftover part file from an earlier run must restart — and
# completeness needs a known total length.
if (
offset
and sha256 is None
and (validator is None or not (size or expected_total))
):
_LOGGER.debug(
"Restarting %s from zero: cannot prove a resumed "
"file correct (no sha256, validator=%s, total=%s)",
url,
validator is not None,
size or expected_total,
)
offset = 0
if size is None or offset < size:
resp, offset = _open_ranged(url, offset, timeout, validator)
# A None response means HTTP 416: the part file already holds
# every byte the server has; fall through to verification.
if resp is not None:
with resp, part.open("ab") as f:
streamed = True
if offset == 0:
validator = _response_validator(resp)
expected_total = _content_length(resp)
# Recorded so a later run can prove an If-Range
# resume of this part file safe.
_write_download_meta(meta, url, validator, expected_total)
_stream_response_to_file(resp, f, offset, size)
# else: a previous run already wrote every byte (or more) but
# was killed before the rename below. Skip the network entirely
# — a Range request past EOF would draw HTTP 416 — and let
# verification decide whether to promote the file or discard it
# and start over.
expected_size = size if size is not None else expected_total
_verify_file(part, sha256, expected_size or None)
if not expected_size and sha256 is None:
# No sha, no size, and the server sent no usable
# content-length: nothing can prove the download complete
# (urllib3 still errors on most short bodies, but not on a
# cleanly closed chunked stream). Promote with a debug
# note rather than fail or warn: some servers (e.g. the
# Espressif constraints host) never send a length, the user
# can do nothing about it, and every current caller
# extracts or parses the file afterwards, where corruption
# fails loudly.
_LOGGER.debug(
"Downloaded %s without any way to verify completeness",
dest.name,
)
# Retry on Windows sharing violations: an antivirus handle on the
# freshly-written file must not get the verified download deleted
# as corrupt by the except clause below. If even the backoff
# retries fail, keep the verified part so the next attempt (or
# run) only has to redo the rename, not the download.
try:
_rename_with_retry(part, dest, overwrite=True)
except PermissionError as e:
_LOGGER.debug("Could not move %s into place: %s", part, e)
last_error = e
continue
meta.unlink(missing_ok=True)
return
except requests.RequestException as e:
# Network failures — including connect errors, since a single
# URL has no mirror-list fallback — keep the part file for the
# next attempt (or the next esphome run) to resume from. Checked
# before OSError: RequestException subclasses IOError.
if not retry_connect_errors and not streamed:
# The caller falls back to another URL on pre-body failures.
raise
_LOGGER.debug("Download of %s interrupted: %s", url, e)
last_error = e
except (OSError, EsphomeError) as e:
# A completed-but-corrupt file (or local disk error) can't be
# trusted for resume; start over.
_LOGGER.debug("Discarding %s: %s", part, e)
part.unlink(missing_ok=True)
meta.unlink(missing_ok=True)
last_error = e
raise EsphomeError(
f"Failed to download {url} after {attempts} attempts: "
f"{_failure_reason(last_error)}"
) from last_error
def _failure_reason(e: Exception) -> str:
"""Format a download exception for the aggregated error message.
@@ -585,110 +906,166 @@ def download_from_mirrors(
``substitutions`` are skipped, so callers can offer templates that only
apply to some downloads.
A path target downloads through ``download_with_resume``, so an
interrupted download resumes on the next esphome run; a file-like target
only resumes mid-stream drops within this call.
Raises:
ValueError: If mirrors list is empty.
EsphomeError: If all download attempts fail; the message lists every
attempted URL with its individual failure reason. Also raised if
no template matched the provided substitutions.
"""
# Imported lazily: requests is a heavy import (~85ms) and is only needed
# when actually downloading a toolchain, never during config validation.
# Imported lazily: requests is a heavy import (~85ms) and is only
# needed when actually downloading, never during config validation.
import requests
from esphome.core import EsphomeError
# 1. Open target file for writing if path given
with ExitStack() as stack:
if isinstance(target, (str, os.PathLike)):
f = stack.enter_context(Path(target).open("wb"))
elif isinstance(target, (io.RawIOBase, io.IOBase)):
f = target
else:
raise TypeError(
f"target must be str, Path, or file-like object: {type(target)}"
)
# 1. Classify the target: filesystem path or open file object
path_target: Path | None = None
f: IO[bytes] | None = None
if isinstance(target, (str, os.PathLike)):
path_target = Path(target)
elif isinstance(target, (io.RawIOBase, io.IOBase)):
f = target
else:
raise TypeError(
f"target must be str, Path, or file-like object: {type(target)}"
)
# 2. Try each mirror in order
failures: list[tuple[str, Exception]] = []
skipped: list[tuple[str, str]] = []
# 2. Try each mirror in order
failures: list[tuple[str, Exception]] = []
skipped: list[tuple[str, str]] = []
for mirror in mirrors:
# 3. Apply substitutions to URL
for mirror in mirrors:
# 3. Apply substitutions to URL
try:
url = mirror.format(**substitutions)
except KeyError as e:
# The template references a substitution not provided for
# this download (e.g. SHORT_VERSION only exists for x.y.0
# versions) - expected, the template just doesn't apply.
_LOGGER.debug("Skipping mirror %s: %s not available", mirror, e)
skipped.append((mirror, f"not applicable ({e.args[0]} not available)"))
continue
except (IndexError, ValueError) as e:
# A malformed template (unbalanced braces, bad format spec)
# is an authoring error, not an expected fallthrough - warn
# even if a later mirror succeeds.
_LOGGER.warning("Skipping malformed mirror URL template %s: %r", mirror, e)
skipped.append((mirror, f"skipped ({e!r})"))
continue
_LOGGER.debug("Trying to download from %s", url)
# Path targets delegate to download_with_resume so a partial
# download persists (and resumes) across esphome runs.
if path_target is not None:
try:
url = mirror.format(**substitutions)
except KeyError as e:
# The template references a substitution not provided for
# this download (e.g. SHORT_VERSION only exists for x.y.0
# versions) - expected, the template just doesn't apply.
_LOGGER.debug("Skipping mirror %s: %s not available", mirror, e)
skipped.append((mirror, f"not applicable ({e.args[0]} not available)"))
continue
except (IndexError, ValueError) as e:
# A malformed template (unbalanced braces, bad format spec)
# is an authoring error, not an expected fallthrough - warn
# even if a later mirror succeeds.
_LOGGER.warning(
"Skipping malformed mirror URL template %s: %r", mirror, e
download_with_resume(
url,
path_target,
attempts=_MIRROR_ATTEMPTS,
timeout=timeout,
# Pre-body failures (connect/HTTP errors) fall to the
# next mirror immediately; only mid-stream drops
# retry-with-resume on the same URL.
retry_connect_errors=False,
)
skipped.append((mirror, f"skipped ({e!r})"))
return url
except (requests.RequestException, OSError, EsphomeError) as e:
# Everything download_with_resume classifies as a download
# failure; programming errors propagate.
_LOGGER.debug("Failed to download %s: %s", url, str(e))
failures.append((url, e))
continue
_LOGGER.debug("Trying to download from %s", url)
# 4. Download; mid-stream failures retry the same mirror with
# resume (see download_with_resume) instead of starting over.
# There is no checksum to verify a resumed file against, so a
# stitch is only trusted when the server proves consistency: the
# If-Range validator guarantees 206 only for unchanged content,
# and the expected total length (when the first response carried
# one) guards against short or shifted bodies. Without a
# validator the retry restarts from zero.
offset = 0
expected_total = 0
validator = None
for attempt in range(_MIRROR_ATTEMPTS):
try:
resp, offset = _open_ranged(url, offset, timeout, validator)
except (requests.RequestException, OSError) as e:
# Connect/HTTP error, no bytes flowed — next mirror.
_LOGGER.debug("Failed to download %s: %s", url, str(e))
failures.append((url, e))
break
try:
# 4. Reset file pointer and download
f.seek(0)
f.truncate(0)
# A None response means HTTP 416: the file already holds
# every byte the server has (a drop after the last byte);
# only the length check below remains.
if resp is not None:
with resp:
if offset == 0:
validator = _response_validator(resp)
expected_total = _content_length(resp)
_stream_response_to_file(resp, f, offset)
with requests.get(url, stream=True, timeout=timeout) as r:
r.raise_for_status()
total_size = int(r.headers.get("content-length", 0))
downloaded = 0
progress = ProgressBar("Downloading") if total_size > 0 else None
for chunk in r.iter_content(chunk_size=8192):
if chunk:
f.write(chunk)
downloaded += len(chunk)
if progress is not None:
progress.update(downloaded / total_size)
if progress is not None:
progress.update(1)
if expected_total and f.tell() != expected_total:
raise EsphomeError(
f"size mismatch: expected {expected_total}, got {f.tell()}"
)
if not expected_total:
# Same trust decision as download_with_resume's
# unverifiable promotion; surface it at the same level.
_LOGGER.debug(
"Downloaded %s without any way to verify completeness",
url,
)
_LOGGER.debug("Downloaded successfully from: %s", url)
# 6. Reset file pointer and return
# 5. Reset file pointer and return
f.seek(0)
return url
except Exception as e: # noqa: BLE001 # pylint: disable=broad-exception-caught
except (requests.RequestException, OSError, EsphomeError) as e:
# Mid-stream drop: keep the received bytes and retry this
# mirror from the current position — but only when the
# server gave a validator to resume against safely AND a
# total length to prove the stitched file complete (the
# length check above is the only verification here).
_LOGGER.debug("Failed to download %s: %s", url, str(e))
failures.append((url, e))
if validator and expected_total:
offset = f.tell()
else:
_LOGGER.debug(
"Restarting %s from zero: cannot prove a "
"resumed file complete (validator=%s, total=%s)",
url,
validator is not None,
expected_total,
)
offset = 0
if attempt == _MIRROR_ATTEMPTS - 1:
failures.append((url, e))
# 7. Report every attempted URL if all mirrors failed. Falling back
# past an early mirror is normal (e.g. only one of the framework URL
# templates matches a given version's tag), so raising only the last
# error would hide the failure that actually matters.
if failures:
attempts = "".join(
f"\n {url}\n {_failure_reason(e)}" for url, e in failures
)
attempts += "".join(
f"\n {mirror}\n {reason}" for mirror, reason in skipped
)
raise EsphomeError(
f"Failed to download from all mirrors:{attempts}"
) from failures[0][1]
if skipped:
details = "".join(
f"\n {mirror}\n {reason}" for mirror, reason in skipped
)
raise EsphomeError(
f"No mirror URL template matched the provided substitutions:{details}"
)
raise ValueError("download_from_mirrors called with an empty mirrors list")
# 6. Report every attempted URL if all mirrors failed. Falling back
# past an early mirror is normal (e.g. only one of the framework URL
# templates matches a given version's tag), so raising only the last
# error would hide the failure that actually matters.
if failures:
attempts = "".join(
f"\n {url}\n {_failure_reason(e)}" for url, e in failures
)
attempts += "".join(f"\n {mirror}\n {reason}" for mirror, reason in skipped)
raise EsphomeError(
f"Failed to download from all mirrors:{attempts}"
) from failures[0][1]
if skipped:
details = "".join(f"\n {mirror}\n {reason}" for mirror, reason in skipped)
raise EsphomeError(
f"No mirror URL template matched the provided substitutions:{details}"
)
raise ValueError("download_from_mirrors called with an empty mirrors list")
tests/unit_tests/fixtures/idf_tools_stub/idf_tools.py
+120
View File
@@ -0,0 +1,120 @@
"""Minimal idf_tools stand-in for get_tool_downloads.py tests."""
from collections.abc import Iterable
import os
CURRENT_PLATFORM = "linux-amd64"
TOOLS_FILE = "tools/tools.json"
class ToolBinaryError(RuntimeError):
pass
class _G:
idf_path: str | None = None
idf_tools_path: str | None = None
tools_json: str | None = None
g = _G()
class IDFEnv:
@classmethod
def get_idf_env(cls) -> "IDFEnv":
return cls()
def add_and_check_targets(idf_env_obj: IDFEnv, targets_str: str) -> list[str]:
return targets_str.split(",")
class _Download:
def __init__(self, url: str, size: int, sha256: str, rename_dist: str = "") -> None:
self.url = url
self.size = size
self.sha256 = sha256
self.rename_dist = rename_dist
class _Version:
def __init__(self, download: _Download | None) -> None:
self._download = download
def get_download_for_platform(self, platform_name: str) -> _Download | None:
return self._download
class _Tool:
def __init__(
self,
versions: dict[str, _Version],
recommended: str | None,
installed: Iterable[str] = (),
broken: bool = False,
) -> None:
self.versions = versions
self._recommended = recommended
self.versions_installed = list(installed)
self._broken = broken
def compatible_with_platform(self) -> bool:
return True
def get_recommended_version(self) -> str | None:
return self._recommended
def find_installed_versions(self) -> None:
if self._broken:
raise ToolBinaryError("broken binary")
_TOOLS = {
"cmake": _Tool(
{"3.30.2": _Version(_Download("https://gh.test/cmake.tar.gz", 11, "aa"))},
"3.30.2",
),
"ninja": _Tool(
{
"1.12.1": _Version(
_Download("https://gh.test/ninja-mac.zip", 22, "bb", "ninja-v1.zip")
)
},
"1.12.1",
),
"installed-tool": _Tool(
{"1.0": _Version(_Download("https://gh.test/x.tar.gz", 33, "cc"))},
"1.0",
installed=["1.0"],
),
"broken-tool": _Tool(
{"2.0": _Version(_Download("https://gh.test/y.tar.gz", 44, "dd"))},
"2.0",
broken=True,
),
"no-recommended-tool": _Tool({"3.0": _Version(None)}, None),
"no-download-tool": _Tool({"4.0": _Version(None)}, "4.0"),
}
def load_tools_info() -> dict[str, _Tool]:
return _TOOLS
def expand_tools_arg(
tools_spec: list[str], overall_tools: dict[str, _Tool], targets: list[str]
) -> list[str]:
if "required" in tools_spec:
return list(overall_tools)
return [t for t in tools_spec if "@" not in t] + [t for t in tools_spec if "@" in t]
def get_idf_download_url_apply_mirrors(
args: object = None, download_url: str = ""
) -> str:
print(f"Changed download URL: {download_url}") # noise on stdout, like idf_tools
prefix = os.environ.get("TEST_MIRROR_PREFIX")
if prefix:
return prefix + download_url
return download_url
tests/unit_tests/test_espidf_framework.py
+316 -1
View File
@@ -3,10 +3,14 @@
# pylint: disable=protected-access
from contextlib import contextmanager
import importlib.util
import io
import json
import logging
import os
from pathlib import Path
import runpy
import subprocess
import sys
import tarfile
from types import SimpleNamespace
@@ -27,6 +31,7 @@ from esphome.espidf.framework import (
_parse_git_source,
_patch_tools_json_demote_openocd,
_patch_tools_json_for_linux_arm64,
_prefetch_idf_tool_archives,
_windows_long_paths_enabled,
_write_idf_version_txt,
_write_stamp,
@@ -311,6 +316,21 @@ class TestTarExtractHardLinkPrefixStripping:
_IDF_VERSION = "5.1.2"
def _fake_download_from_mirrors(
mirrors: list[str],
substitutions: dict[str, str],
target: object,
**kwargs: object,
) -> str:
"""Stand-in for download_from_mirrors that creates path targets, since
the framework code opens the downloaded tarball afterwards."""
if isinstance(target, (str, os.PathLike)):
path = Path(target)
path.parent.mkdir(parents=True, exist_ok=True)
path.touch()
return "https://example.com/idf.tar.xz"
@pytest.fixture
def espidf_mocks(setup_core: Path):
"""Patch the heavy I/O of check_esp_idf_install and pre-create the framework dir."""
@@ -321,7 +341,7 @@ def espidf_mocks(setup_core: Path):
patch("esphome.espidf.framework.rmdir") as rmdir_mock,
patch(
"esphome.espidf.framework.download_from_mirrors",
return_value="https://example.com/idf.tar.xz",
side_effect=_fake_download_from_mirrors,
) as download,
patch("esphome.espidf.framework.archive_extract_all") as extract,
patch("esphome.espidf.framework.create_venv") as venv,
@@ -333,6 +353,7 @@ def espidf_mocks(setup_core: Path):
patch("esphome.espidf.framework._write_idf_version_txt"),
patch("esphome.espidf.framework._patch_tools_json_for_linux_arm64"),
patch("esphome.espidf.framework._patch_tools_json_demote_openocd"),
patch("esphome.espidf.framework._prefetch_idf_tool_archives"),
patch("esphome.espidf.framework._write_stamp"),
patch("esphome.espidf.framework._check_stamp", return_value=True),
patch("esphome.espidf.framework._get_idf_version", return_value=_IDF_VERSION),
@@ -413,6 +434,20 @@ def test_check_esp_idf_install_already_installed(espidf_mocks: SimpleNamespace)
espidf_mocks.venv.assert_not_called()
def test_corrupt_tarball_removed_when_extraction_fails(
espidf_mocks: SimpleNamespace,
) -> None:
"""A tarball that fails to extract (e.g. torn by an unclean shutdown) is
deleted so the next run re-downloads instead of failing forever."""
espidf_mocks.extract.side_effect = RuntimeError("xz: unexpected end of input")
tarball = get_idf_tools_path() / "dist" / f"esp-idf-{_IDF_VERSION}.tar.xz"
with pytest.raises(RuntimeError, match="unexpected end of input"):
check_esp_idf_install(_IDF_VERSION, force=True)
assert not tarball.exists()
def test_check_esp_idf_install_framework_failure(espidf_mocks: SimpleNamespace) -> None:
"""A failing idf_tools install raises."""
espidf_mocks.run_ok.side_effect = [False]
@@ -636,6 +671,286 @@ def test_patch_tools_json_already_patched_is_noop(tmp_path: Path) -> None:
assert tools_json.read_text(encoding="utf-8") == before
# ---------------------------------------------------------------------------
# _prefetch_idf_tool_archives
# ---------------------------------------------------------------------------
_PREFETCH_JSON = json.dumps(
[
{
"name": "cmake@3.30.2",
"url": "https://example.com/cmake.tar.gz",
"size": 123,
"sha256": "ab" * 32,
"dest": "cmake-3.30.2.tar.gz",
},
{
"name": "ninja@1.12.1",
"url": "https://example.com/ninja.zip",
"size": 45,
"sha256": "cd" * 32,
"dest": "ninja.zip",
},
]
)
def test_prefetch_downloads_each_archive_with_resume(tmp_path: Path) -> None:
with (
patch(
"esphome.espidf.framework.run_command",
return_value=(True, _PREFETCH_JSON, ""),
),
patch("esphome.espidf.framework.download_with_resume") as download,
patch("esphome.espidf.framework.get_system_python_path", return_value="python"),
):
_prefetch_idf_tool_archives(tmp_path, "esp32", ["required"], None)
dist = get_idf_tools_path() / "dist"
assert download.call_count == 2
assert download.call_args_list[0][0] == (
"https://example.com/cmake.tar.gz",
dist / "cmake-3.30.2.tar.gz",
)
assert download.call_args_list[0][1] == {"sha256": "ab" * 32, "size": 123}
def test_prefetch_skips_already_downloaded_archives(tmp_path: Path) -> None:
dist = get_idf_tools_path() / "dist"
dist.mkdir(parents=True)
(dist / "cmake-3.30.2.tar.gz").write_bytes(b"cached")
with (
patch(
"esphome.espidf.framework.run_command",
return_value=(True, _PREFETCH_JSON, ""),
),
patch("esphome.espidf.framework.download_with_resume") as download,
patch("esphome.espidf.framework.get_system_python_path", return_value="python"),
):
_prefetch_idf_tool_archives(tmp_path, "esp32", ["required"], None)
# only the missing archive is downloaded
assert download.call_count == 1
assert download.call_args[0][1] == dist / "ninja.zip"
@pytest.mark.parametrize(
("run_result", "download_error", "expected_log"),
[
((False, "", "script exploded"), None, "tool downloads"), # script failure
((True, "{ not json", ""), None, "prefetch failed"), # unparsable output
(
(True, _PREFETCH_JSON, ""),
OSError("network down"),
"Could not prefetch",
), # download failure
],
)
def test_prefetch_failures_never_raise(
tmp_path: Path,
caplog: pytest.LogCaptureFixture,
run_result: tuple[bool, str, str],
download_error: Exception | None,
expected_log: str,
) -> None:
"""The prefetch is best-effort; idf_tools downloads whatever is missing."""
with (
patch("esphome.espidf.framework.run_command", return_value=run_result),
patch(
"esphome.espidf.framework.download_with_resume",
side_effect=download_error,
),
patch("esphome.espidf.framework.get_system_python_path", return_value="python"),
):
_prefetch_idf_tool_archives(tmp_path, "esp32", ["required"], None)
assert expected_log in caplog.text
def test_prefetch_one_failed_archive_does_not_stop_the_rest(
tmp_path: Path, caplog: pytest.LogCaptureFixture
) -> None:
"""A single archive failing its download must not abort the prefetch of
the remaining archives."""
with (
patch(
"esphome.espidf.framework.run_command",
return_value=(True, _PREFETCH_JSON, ""),
),
patch(
"esphome.espidf.framework.download_with_resume",
side_effect=[OSError("network down"), None],
) as download,
patch("esphome.espidf.framework.get_system_python_path", return_value="python"),
):
_prefetch_idf_tool_archives(tmp_path, "esp32", ["required"], None)
assert download.call_count == 2
assert "Could not prefetch cmake@3.30.2" in caplog.text
def test_prefetch_passes_targets_and_tools_to_script(tmp_path: Path) -> None:
with (
patch(
"esphome.espidf.framework.run_command", return_value=(True, "[]", "")
) as run,
patch("esphome.espidf.framework.get_system_python_path", return_value="python"),
):
_prefetch_idf_tool_archives(
tmp_path, "esp32,esp32c3", ["required", "cmake"], {"IDF_TOOLS_PATH": "/x"}
)
cmd = run.call_args[0][0]
assert cmd[-3:] == ["esp32,esp32c3", "required", "cmake"]
assert cmd[1].endswith("get_tool_downloads.py")
# the script inherits the caller's env plus the framework tools PYTHONPATH
env = run.call_args[1]["env"]
assert env["IDF_TOOLS_PATH"] == "/x"
assert env["PYTHONPATH"] == str(tmp_path / "tools")
def test_framework_install_prefetches_before_installer(
espidf_mocks: SimpleNamespace,
) -> None:
"""The prefetch runs before idf_tools.py install so the installer finds
the archives already in dist/."""
calls: list[str] = []
with (
patch(
"esphome.espidf.framework._prefetch_idf_tool_archives",
side_effect=lambda *a, **k: calls.append("prefetch"),
),
):
espidf_mocks.run_ok.side_effect = lambda *a, **k: (
calls.append("install") or True
)
check_esp_idf_install(_IDF_VERSION, force=True)
assert calls.index("prefetch") < calls.index("install")
# ---------------------------------------------------------------------------
# get_tool_downloads.py (against the stub idf_tools module in fixtures/)
# ---------------------------------------------------------------------------
_IDF_TOOLS_STUB_DIR = Path(__file__).parent / "fixtures" / "idf_tools_stub"
def _run_downloads_script(
tmp_path: Path, *args: str, env_extra: dict[str, str] | None = None
) -> subprocess.CompletedProcess[str]:
"""Run the real get_tool_downloads.py against the stub idf_tools module."""
script = Path(__file__).parents[2] / "esphome" / "espidf" / "get_tool_downloads.py"
env = os.environ | {
"PYTHONPATH": str(_IDF_TOOLS_STUB_DIR),
"IDF_TOOLS_PATH": str(tmp_path / "tp"),
}
if env_extra:
env |= env_extra
return subprocess.run(
[sys.executable, str(script), str(tmp_path / "fw"), *args],
capture_output=True,
text=True,
env=env,
check=False,
)
def test_get_tool_downloads_lists_missing_tools(tmp_path: Path) -> None:
"""Installed versions are skipped, tools that fail their binary check are
still listed, rename_dist decides the dist filename, and idf_tools' stdout
chatter stays off the JSON channel."""
result = _run_downloads_script(tmp_path, "esp32", "required")
assert result.returncode == 0, result.stderr
downloads = {d["name"]: d for d in json.loads(result.stdout)}
# installed-tool@1.0 is already installed and must not be listed
assert set(downloads) == {"cmake@3.30.2", "ninja@1.12.1", "broken-tool@2.0"}
assert downloads["cmake@3.30.2"]["dest"] == "cmake.tar.gz"
assert downloads["cmake@3.30.2"]["size"] == 11
assert downloads["cmake@3.30.2"]["sha256"] == "aa"
# rename_dist overrides the URL basename
assert downloads["ninja@1.12.1"]["dest"] == "ninja-v1.zip"
# the stub prints informational lines; they must be on stderr
assert "Changed download URL" in result.stderr
def test_get_tool_downloads_applies_mirror_rewrite(tmp_path: Path) -> None:
result = _run_downloads_script(
tmp_path,
"esp32",
"required",
env_extra={"TEST_MIRROR_PREFIX": "https://mirror.test/"},
)
assert result.returncode == 0, result.stderr
downloads = json.loads(result.stdout)
assert all(d["url"].startswith("https://mirror.test/") for d in downloads)
def _run_downloads_inprocess(
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
capsys: pytest.CaptureFixture[str],
*args: str,
) -> list[dict]:
"""Execute get_tool_downloads.py in-process against the stub idf_tools.
Unlike the subprocess variant this runs under coverage, exercising the
script's own lines.
"""
spec = importlib.util.spec_from_file_location(
"idf_tools", _IDF_TOOLS_STUB_DIR / "idf_tools.py"
)
stub = importlib.util.module_from_spec(spec)
spec.loader.exec_module(stub)
monkeypatch.setitem(sys.modules, "idf_tools", stub)
monkeypatch.setenv("IDF_TOOLS_PATH", str(tmp_path / "tp"))
script = Path(__file__).parents[2] / "esphome" / "espidf" / "get_tool_downloads.py"
monkeypatch.setattr(sys, "argv", [str(script), str(tmp_path / "fw"), *args])
runpy.run_path(str(script))
return json.loads(capsys.readouterr().out)
def test_get_tool_downloads_inprocess_full_flow(
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
capsys: pytest.CaptureFixture[str],
) -> None:
"""In-process run covering the whole script: required expansion,
installed/broken tools, rename_dist, and version pinning via tool@version."""
downloads = {
d["name"]: d
for d in _run_downloads_inprocess(
tmp_path, monkeypatch, capsys, "esp32", "required"
)
}
assert set(downloads) == {"cmake@3.30.2", "ninja@1.12.1", "broken-tool@2.0"}
assert downloads["ninja@1.12.1"]["dest"] == "ninja-v1.zip"
assert downloads["cmake@3.30.2"]["url"] == "https://gh.test/cmake.tar.gz"
def test_get_tool_downloads_inprocess_explicit_tool_specs(
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
capsys: pytest.CaptureFixture[str],
) -> None:
"""Explicit tool names and tool@version specs resolve; unknown tools and
unknown versions are skipped."""
downloads = _run_downloads_inprocess(
tmp_path,
monkeypatch,
capsys,
"esp32",
"cmake@3.30.2",
"no-such-tool",
"cmake@9.9.9",
)
assert [d["name"] for d in downloads] == ["cmake@3.30.2"]
# ---------------------------------------------------------------------------
# _patch_tools_json_demote_openocd (openocd-esp32 made optional)
# ---------------------------------------------------------------------------
tests/unit_tests/test_framework_helpers.py
+710 -8
View File
@@ -2,8 +2,10 @@
# pylint: disable=protected-access
import hashlib
import importlib.util
import io
import json
import logging
import os
from pathlib import Path
@@ -16,6 +18,7 @@ import zipfile
import pytest
import requests as req
from esphome import framework_helpers
from esphome.core import EsphomeError
from esphome.framework_helpers import (
_7z_extract_all,
@@ -26,6 +29,7 @@ from esphome.framework_helpers import (
archive_extract_all,
create_venv,
download_from_mirrors,
download_with_resume,
get_project_compile_flags,
get_project_cxx_compile_flags,
get_project_link_flags,
@@ -507,7 +511,7 @@ class TestArchiveExtractAll:
# ---------------------------------------------------------------------------
# download_from_mirrors
# download_from_mirrors / download_with_resume
# ---------------------------------------------------------------------------
@@ -515,6 +519,8 @@ def _mock_response(content: bytes, ok: bool = True) -> MagicMock:
r = MagicMock()
r.__enter__.return_value = r
r.__exit__.return_value = False
r.status_code = 200
r.ok = ok
if ok:
r.raise_for_status.return_value = None
else:
@@ -524,6 +530,563 @@ def _mock_response(content: bytes, ok: bool = True) -> MagicMock:
return r
def _interrupted_response(content: bytes, etag: str | None = None) -> MagicMock:
"""A response whose body yields ``content`` and then drops mid-stream.
``etag`` makes the response resumable: without a validator the retry
logic restarts from zero rather than stitching unverified bytes.
"""
def body(chunk_size):
yield content
raise req.exceptions.ChunkedEncodingError("connection dropped")
r = _mock_response(b"")
if etag is not None:
r.headers = {**r.headers, "ETag": etag}
r.iter_content.side_effect = body
return r
def _resumed_response(content: bytes) -> MagicMock:
"""An HTTP 206 response continuing an interrupted download."""
r = _mock_response(content)
r.status_code = 206
return r
class TestOpenRanged:
def test_fresh_download_sends_no_range(self) -> None:
with patch("requests.get", return_value=_mock_response(b"x")) as mock_get:
resp, offset = framework_helpers._open_ranged("https://e.com/f", 0, 30)
assert offset == 0
assert mock_get.call_args[1]["headers"] == {}
assert resp is mock_get.return_value
def test_resume_kept_on_206(self) -> None:
with patch("requests.get", return_value=_resumed_response(b"x")):
_, offset = framework_helpers._open_ranged("https://e.com/f", 7, 30)
assert offset == 7
def test_resume_downgraded_on_200(self) -> None:
"""A server that ignores the Range header forces a restart."""
with patch("requests.get", return_value=_mock_response(b"x")):
_, offset = framework_helpers._open_ranged("https://e.com/f", 7, 30)
assert offset == 0
def test_http_error_closes_response_and_raises(self) -> None:
r = _mock_response(b"", ok=False)
with (
patch("requests.get", return_value=r),
pytest.raises(req.HTTPError),
):
framework_helpers._open_ranged("https://e.com/f", 0, 30)
r.close.assert_called_once()
def test_connect_error_propagates(self) -> None:
with (
patch("requests.get", side_effect=req.ConnectionError("refused")),
pytest.raises(req.ConnectionError),
):
framework_helpers._open_ranged("https://e.com/f", 0, 30)
class TestDownloadWithResume:
def test_downloads_and_renames(self, tmp_path: Path) -> None:
dest = tmp_path / "tool.tar.gz"
with patch("requests.get", return_value=_mock_response(b"data")) as mock_get:
download_with_resume("https://example.com/t", dest)
assert dest.read_bytes() == b"data"
assert not (tmp_path / "tool.tar.gz.part").exists()
# a fresh download must not send a Range header
assert "Range" not in mock_get.call_args[1]["headers"]
def test_mid_stream_drop_resumes_with_range(self, tmp_path: Path) -> None:
dest = tmp_path / "tool.tar.gz"
first = _interrupted_response(b"1234", etag='"v1"')
first.headers = {**first.headers, "content-length": "8"}
with patch(
"requests.get",
side_effect=[first, _resumed_response(b"5678")],
) as mock_get:
download_with_resume("https://example.com/t", dest)
# earlier bytes were kept, remainder appended conditionally
assert dest.read_bytes() == b"12345678"
assert mock_get.call_args_list[1][1]["headers"] == {
"Range": "bytes=4-",
"If-Range": '"v1"',
}
def test_unverifiable_drop_without_length_restarts(self, tmp_path: Path) -> None:
"""A validator alone is not enough to stitch when nothing can prove
the stitched file complete (no sha/size and no content-length)."""
dest = tmp_path / "tool.tar.gz"
with patch(
"requests.get",
side_effect=[
_interrupted_response(b"1234", etag='"v1"'),
_mock_response(b"full"),
],
) as mock_get:
download_with_resume("https://example.com/t", dest)
assert dest.read_bytes() == b"full"
assert "Range" not in mock_get.call_args_list[1][1]["headers"]
def test_resumed_clean_but_short_body_discarded(self, tmp_path: Path) -> None:
"""A resumed stream that ends cleanly but short of the advertised
total is rejected and re-downloaded, not promoted."""
dest = tmp_path / "tool.tar.gz"
first = _interrupted_response(b"abcd", etag='"v1"')
first.headers = {**first.headers, "content-length": "8"}
# resume ends cleanly after only 2 of the 4 missing bytes
short = _resumed_response(b"ef")
full = _mock_response(b"abcdefgh")
full.headers = {**full.headers, "content-length": "8"}
with patch("requests.get", side_effect=[first, short, full]) as mock_get:
download_with_resume("https://example.com/t", dest)
assert dest.read_bytes() == b"abcdefgh"
# the short stitch was discarded; the final attempt started fresh
assert "Range" not in mock_get.call_args_list[2][1]["headers"]
def test_unverifiable_drop_without_validator_restarts(self, tmp_path: Path) -> None:
"""No sha/size and no server validator: the retry must not stitch."""
dest = tmp_path / "tool.tar.gz"
with patch(
"requests.get",
side_effect=[_interrupted_response(b"1234"), _mock_response(b"full")],
) as mock_get:
download_with_resume("https://example.com/t", dest)
assert dest.read_bytes() == b"full"
assert "Range" not in mock_get.call_args_list[1][1]["headers"]
def test_resume_across_invocations_from_part_file(self, tmp_path: Path) -> None:
"""A .part file left by a previous run is resumed, not restarted,
when sha/size verification will vouch for the stitched result."""
dest = tmp_path / "tool.tar.gz"
(tmp_path / "tool.tar.gz.part").write_bytes(b"12345")
good = hashlib.sha256(b"12345678").hexdigest()
with patch("requests.get", return_value=_resumed_response(b"678")) as mock_get:
download_with_resume("https://example.com/t", dest, sha256=good, size=8)
assert dest.read_bytes() == b"12345678"
assert mock_get.call_args[1]["headers"] == {"Range": "bytes=5-"}
def test_unverifiable_leftover_part_file_ignored(self, tmp_path: Path) -> None:
"""Without sha/size there is no way to vouch for a cross-run stitch,
so a leftover part file starts over."""
dest = tmp_path / "tool.tar.gz"
(tmp_path / "tool.tar.gz.part").write_bytes(b"12345")
with patch("requests.get", return_value=_mock_response(b"fresh")) as mock_get:
download_with_resume("https://example.com/t", dest)
assert dest.read_bytes() == b"fresh"
assert "Range" not in mock_get.call_args[1]["headers"]
def test_server_without_range_support_restarts(self, tmp_path: Path) -> None:
"""HTTP 200 in response to a Range request truncates and restarts."""
dest = tmp_path / "tool.tar.gz"
(tmp_path / "tool.tar.gz.part").write_bytes(b"sta")
good = hashlib.sha256(b"fresh").hexdigest()
with patch("requests.get", return_value=_mock_response(b"fresh")) as mock_get:
download_with_resume("https://example.com/t", dest, sha256=good, size=5)
# the Range request was sent (verifiable resume) and downgraded
assert mock_get.call_args[1]["headers"] == {"Range": "bytes=3-"}
assert dest.read_bytes() == b"fresh"
def test_size_only_leftover_part_restarts(self, tmp_path: Path) -> None:
"""A size alone cannot detect a same-length content change on the
server, so a cross-run part without sha256 restarts from zero."""
dest = tmp_path / "tool.tar.gz"
(tmp_path / "tool.tar.gz.part").write_bytes(b"12")
with patch("requests.get", return_value=_mock_response(b"1234")) as mock_get:
download_with_resume("https://example.com/t", dest, size=4)
assert "Range" not in mock_get.call_args[1]["headers"]
assert dest.read_bytes() == b"1234"
def test_size_only_in_run_drop_resumes_with_validator(self, tmp_path: Path) -> None:
"""Within a run the If-Range validator proves identity, so size-only
callers still resume mid-stream drops."""
dest = tmp_path / "tool.tar.gz"
with patch(
"requests.get",
side_effect=[
_interrupted_response(b"12", etag='"v1"'),
_resumed_response(b"34"),
],
) as mock_get:
download_with_resume("https://example.com/t", dest, size=4)
assert dest.read_bytes() == b"1234"
assert mock_get.call_args_list[1][1]["headers"] == {
"Range": "bytes=2-",
"If-Range": '"v1"',
}
def test_unverifiable_download_logged(
self, tmp_path: Path, caplog: pytest.LogCaptureFixture
) -> None:
"""No sha, no size, no content-length: the download is promoted with
a debug note (routine for e.g. the constraints host, so not a
warning) that completeness could not be verified."""
dest = tmp_path / "tool.tar.gz"
with (
caplog.at_level(logging.DEBUG, logger="esphome.framework_helpers"),
patch("requests.get", return_value=_mock_response(b"data")),
):
download_with_resume("https://example.com/t", dest)
assert dest.read_bytes() == b"data"
assert "without any way to verify completeness" in caplog.text
def test_416_promotes_complete_part_when_size_unknown(self, tmp_path: Path) -> None:
"""sha256-only caller with a byte-complete part file: the server's
416 confirms nothing is missing, verification promotes in place, and
the 416 must not loop as a retryable error."""
dest = tmp_path / "tool.tar.gz"
(tmp_path / "tool.tar.gz.part").write_bytes(b"data")
good = hashlib.sha256(b"data").hexdigest()
r416 = _mock_response(b"", ok=False)
r416.status_code = 416
with patch("requests.get", return_value=r416) as mock_get:
download_with_resume("https://example.com/t", dest, sha256=good)
assert mock_get.call_count == 1
r416.close.assert_called_once()
assert dest.read_bytes() == b"data"
def test_416_with_corrupt_part_discards_and_redownloads(
self, tmp_path: Path
) -> None:
dest = tmp_path / "tool.tar.gz"
(tmp_path / "tool.tar.gz.part").write_bytes(b"bad!")
good = hashlib.sha256(b"data").hexdigest()
r416 = _mock_response(b"", ok=False)
r416.status_code = 416
with patch(
"requests.get", side_effect=[r416, _mock_response(b"data")]
) as mock_get:
download_with_resume("https://example.com/t", dest, sha256=good)
assert "Range" not in mock_get.call_args_list[1][1]["headers"]
assert dest.read_bytes() == b"data"
def test_hash_mismatch_discards_and_retries(self, tmp_path: Path) -> None:
dest = tmp_path / "tool.tar.gz"
good = hashlib.sha256(b"good").hexdigest()
with patch(
"requests.get",
side_effect=[_mock_response(b"bad!"), _mock_response(b"good")],
) as mock_get:
download_with_resume("https://example.com/t", dest, sha256=good, size=4)
assert dest.read_bytes() == b"good"
# the corrupt part file was discarded, so the retry starts fresh
assert "Range" not in mock_get.call_args_list[1][1]["headers"]
def test_size_mismatch_discards_part(self, tmp_path: Path) -> None:
dest = tmp_path / "tool.tar.gz"
with (
patch("requests.get", return_value=_mock_response(b"xx")),
pytest.raises(EsphomeError, match="after 2 attempts"),
):
download_with_resume("https://example.com/t", dest, size=99, attempts=2)
assert not (tmp_path / "tool.tar.gz.part").exists()
assert not dest.exists()
def test_attempts_exhausted_keeps_part_file(self, tmp_path: Path) -> None:
"""Mid-stream failures keep the partial file so a later run resumes."""
dest = tmp_path / "tool.tar.gz"
first = _interrupted_response(b"12", etag='"v1"')
first.headers = {**first.headers, "content-length": "4"}
second = _interrupted_response(b"34")
second.status_code = 206
with (
patch("requests.get", side_effect=[first, second]),
pytest.raises(EsphomeError, match="after 2 attempts"),
):
download_with_resume("https://example.com/t", dest, attempts=2)
assert (tmp_path / "tool.tar.gz.part").read_bytes() == b"1234"
def test_multiple_drops_accumulate_across_attempts(self, tmp_path: Path) -> None:
"""Each attempt appends its bytes; three partial responses complete
the file."""
dest = tmp_path / "tool.tar.gz"
first = _interrupted_response(b"ab", etag='"v1"')
first.headers = {**first.headers, "content-length": "6"}
second = _interrupted_response(b"cd")
second.status_code = 206
third = _resumed_response(b"ef")
with patch(
"requests.get",
side_effect=[first, second, third],
) as mock_get:
download_with_resume("https://example.com/t", dest)
assert dest.read_bytes() == b"abcdef"
expected = {"Range": "bytes=2-", "If-Range": '"v1"'}
assert mock_get.call_args_list[1][1]["headers"] == expected
expected = {"Range": "bytes=4-", "If-Range": '"v1"'}
assert mock_get.call_args_list[2][1]["headers"] == expected
def test_connect_error_then_success(self, tmp_path: Path) -> None:
"""A connect error (no response at all) consumes an attempt and the
next attempt succeeds."""
dest = tmp_path / "tool.tar.gz"
with patch(
"requests.get",
side_effect=[req.ConnectionError("refused"), _mock_response(b"data")],
):
download_with_resume("https://example.com/t", dest)
assert dest.read_bytes() == b"data"
def test_http_error_keeps_part_file(self, tmp_path: Path) -> None:
"""A transient HTTP error (e.g. 503) must not discard resume state."""
dest = tmp_path / "tool.tar.gz"
(tmp_path / "tool.tar.gz.part").write_bytes(b"keep")
error = _mock_response(b"", ok=False)
error.status_code = 503
with (
patch("requests.get", return_value=error),
pytest.raises(EsphomeError, match="after 1 attempts"),
):
download_with_resume("https://example.com/t", dest, attempts=1)
assert (tmp_path / "tool.tar.gz.part").read_bytes() == b"keep"
def test_creates_missing_parent_directories(self, tmp_path: Path) -> None:
dest = tmp_path / "dist" / "nested" / "tool.tar.gz"
with patch("requests.get", return_value=_mock_response(b"data")):
download_with_resume("https://example.com/t", dest)
assert dest.read_bytes() == b"data"
def test_verifies_both_size_and_sha(self, tmp_path: Path) -> None:
dest = tmp_path / "tool.tar.gz"
good = hashlib.sha256(b"data").hexdigest()
with patch("requests.get", return_value=_mock_response(b"data")):
download_with_resume("https://example.com/t", dest, sha256=good, size=4)
assert dest.read_bytes() == b"data"
def test_corrupt_partial_resumed_then_discarded_then_redownloaded(
self, tmp_path: Path
) -> None:
"""The full recovery cycle for a corrupted partial download: the
resume completes it, verification fails, the poisoned part file is
discarded, and the next attempt re-downloads from scratch."""
dest = tmp_path / "tool.tar.gz"
# a previous run left a corrupted 4-byte prefix behind
(tmp_path / "tool.tar.gz.part").write_bytes(b"BAD!")
good = hashlib.sha256(b"data66").hexdigest()
with patch(
"requests.get",
side_effect=[
_resumed_response(b"66"), # resume "completes" the bad part
_mock_response(b"data66"), # clean retry from zero
],
) as mock_get:
download_with_resume("https://example.com/t", dest, sha256=good, size=6)
# first attempt resumed at the corrupt offset, failed verification;
# second attempt started fresh (no Range header) and succeeded
assert mock_get.call_args_list[0][1]["headers"] == {"Range": "bytes=4-"}
assert "Range" not in mock_get.call_args_list[1][1]["headers"]
assert dest.read_bytes() == b"data66"
assert not (tmp_path / "tool.tar.gz.part").exists()
def test_existing_dest_passing_verification_kept(self, tmp_path: Path) -> None:
"""A dest completed by an earlier run is reused without any request."""
dest = tmp_path / "tool.tar.gz"
dest.write_bytes(b"data")
good = hashlib.sha256(b"data").hexdigest()
with patch("requests.get") as mock_get:
download_with_resume("https://example.com/t", dest, sha256=good, size=4)
mock_get.assert_not_called()
assert dest.read_bytes() == b"data"
@pytest.mark.parametrize(
"stale",
[
pytest.param(b"corrupt!", id="wrong-size"),
pytest.param(b"bad!", id="right-size-wrong-hash"),
],
)
def test_existing_dest_failing_verification_redownloaded(
self, tmp_path: Path, stale: bytes
) -> None:
dest = tmp_path / "tool.tar.gz"
dest.write_bytes(stale)
good = hashlib.sha256(b"data").hexdigest()
with patch("requests.get", return_value=_mock_response(b"data")):
download_with_resume("https://example.com/t", dest, sha256=good, size=4)
assert dest.read_bytes() == b"data"
def test_existing_dest_with_size_only_kept(self, tmp_path: Path) -> None:
dest = tmp_path / "tool.tar.gz"
dest.write_bytes(b"data")
with patch("requests.get") as mock_get:
download_with_resume("https://example.com/t", dest, size=4)
mock_get.assert_not_called()
def test_existing_dest_with_sha_only_kept(self, tmp_path: Path) -> None:
"""sha-only verification also authorizes reusing a completed dest."""
dest = tmp_path / "tool.tar.gz"
dest.write_bytes(b"data")
good = hashlib.sha256(b"data").hexdigest()
with patch("requests.get") as mock_get:
download_with_resume("https://example.com/t", dest, sha256=good)
mock_get.assert_not_called()
def test_meta_write_failure_is_best_effort(self, tmp_path: Path) -> None:
"""A failure to persist the resume sidecar must not fail the
download itself."""
dest = tmp_path / "f.tar.xz"
first = _mock_response(b"data")
first.headers = {**first.headers, "ETag": '"v1"', "content-length": "4"}
with (
patch("requests.get", return_value=first),
patch.object(Path, "write_text", side_effect=OSError("read-only")),
):
download_with_resume("https://example.com/f", dest)
assert dest.read_bytes() == b"data"
def test_meta_sidecar_written_and_removed(self, tmp_path: Path) -> None:
"""The validator sidecar appears while downloading and is cleaned up
with the promotion."""
dest = tmp_path / "f.tar.xz"
meta = tmp_path / "f.tar.xz.part.meta"
seen: list[bool] = []
first = _interrupted_response(b"1234", etag='"v1"')
first.headers = {**first.headers, "content-length": "8"}
responses = [first]
def get(*args: object, **kwargs: object) -> MagicMock:
if responses:
return responses.pop(0)
# the resume request: the sidecar written by the first response
# must already be on disk at this point
seen.append(meta.is_file())
return _resumed_response(b"5678")
with patch("requests.get", side_effect=get):
download_with_resume("https://example.com/f", dest)
assert dest.read_bytes() == b"12345678"
assert seen == [True] # sidecar existed during the resume attempt
assert not meta.exists() # cleaned up on success
def test_locked_promotion_keeps_verified_part(self, tmp_path: Path) -> None:
"""A rename that stays blocked (e.g. a long-lived Windows file lock)
must not delete the verified download; the next attempt retries just
the rename without touching the network."""
dest = tmp_path / "tool.tar.gz"
good = hashlib.sha256(b"data").hexdigest()
with (
patch("requests.get", return_value=_mock_response(b"data")) as mock_get,
patch(
"esphome.framework_helpers._rename_with_retry",
side_effect=[PermissionError("locked"), None],
) as rename,
):
download_with_resume("https://example.com/t", dest, sha256=good, size=4)
# one download; the second attempt only redid the rename
assert mock_get.call_count == 1
assert rename.call_count == 2
def test_locked_promotion_exhausted_keeps_part_for_next_run(
self, tmp_path: Path
) -> None:
dest = tmp_path / "tool.tar.gz"
good = hashlib.sha256(b"data").hexdigest()
with (
patch("requests.get", return_value=_mock_response(b"data")),
patch(
"esphome.framework_helpers._rename_with_retry",
side_effect=PermissionError("locked"),
),
pytest.raises(EsphomeError, match="after 1 attempts"),
):
download_with_resume(
"https://example.com/t", dest, sha256=good, size=4, attempts=1
)
# the verified bytes survive for the next run
assert (tmp_path / "tool.tar.gz.part").read_bytes() == b"data"
def test_meta_sidecar_resumes_across_runs_without_sha(self, tmp_path: Path) -> None:
"""A later run resumes an unfinished download using the validator the
first run stored — the cross-run fix for the framework tarball."""
dest = tmp_path / "f.tar.xz"
(tmp_path / "f.tar.xz.part").write_bytes(b"1234")
(tmp_path / "f.tar.xz.part.meta").write_text(
json.dumps(
{"url": "https://example.com/f", "validator": '"v1"', "total": 8}
)
)
with patch("requests.get", return_value=_resumed_response(b"5678")) as mock_get:
download_with_resume("https://example.com/f", dest)
assert dest.read_bytes() == b"12345678"
assert mock_get.call_args[1]["headers"] == {
"Range": "bytes=4-",
"If-Range": '"v1"',
}
def test_meta_sidecar_for_other_url_ignored(self, tmp_path: Path) -> None:
"""Metadata from a different mirror URL must not authorize a stitch."""
dest = tmp_path / "f.tar.xz"
(tmp_path / "f.tar.xz.part").write_bytes(b"1234")
(tmp_path / "f.tar.xz.part.meta").write_text(
json.dumps({"url": "https://other.com/f", "validator": '"v1"', "total": 8})
)
full = _mock_response(b"12345678")
with patch("requests.get", return_value=full) as mock_get:
download_with_resume("https://example.com/f", dest)
assert "Range" not in mock_get.call_args[1]["headers"]
assert dest.read_bytes() == b"12345678"
def test_complete_part_file_promoted_without_network(self, tmp_path: Path) -> None:
"""A .part holding every byte (killed between write and rename) is
verified in place and promoted; no request is made, so no 416 loop."""
dest = tmp_path / "tool.tar.gz"
(tmp_path / "tool.tar.gz.part").write_bytes(b"data")
good = hashlib.sha256(b"data").hexdigest()
with patch("requests.get") as mock_get:
download_with_resume("https://example.com/t", dest, sha256=good, size=4)
mock_get.assert_not_called()
assert dest.read_bytes() == b"data"
def test_complete_but_corrupt_part_file_redownloaded(self, tmp_path: Path) -> None:
"""A full-size .part with a wrong hash is discarded and re-downloaded
from scratch."""
dest = tmp_path / "tool.tar.gz"
(tmp_path / "tool.tar.gz.part").write_bytes(b"bad!")
good = hashlib.sha256(b"data").hexdigest()
with patch("requests.get", return_value=_mock_response(b"data")) as mock_get:
download_with_resume("https://example.com/t", dest, sha256=good, size=4)
assert "Range" not in mock_get.call_args[1]["headers"]
assert dest.read_bytes() == b"data"
def test_oversized_part_file_discarded(self, tmp_path: Path) -> None:
"""A .part larger than the expected size fails verification and is
replaced by a fresh download."""
dest = tmp_path / "tool.tar.gz"
(tmp_path / "tool.tar.gz.part").write_bytes(b"toolong")
good = hashlib.sha256(b"data").hexdigest()
with patch("requests.get", return_value=_mock_response(b"data")):
download_with_resume("https://example.com/t", dest, sha256=good, size=4)
assert dest.read_bytes() == b"data"
def test_malformed_content_length_degrades_gracefully(self, tmp_path: Path) -> None:
"""A garbage Content-Length must not crash the attempt; it means
"unknown", so a drop restarts instead of stitching and a clean
download still succeeds."""
dest = tmp_path / "tool.tar.gz"
first = _interrupted_response(b"1234", etag='"v1"')
first.headers = {**first.headers, "content-length": "explode"}
retry = _mock_response(b"full")
retry.headers = {**retry.headers, "content-length": "explode"}
with patch("requests.get", side_effect=[first, retry]) as mock_get:
download_with_resume("https://example.com/t", dest)
assert dest.read_bytes() == b"full"
# unknown length -> completeness unprovable -> no resume attempted
assert "Range" not in mock_get.call_args_list[1][1]["headers"]
def test_zero_byte_part_file_sends_no_range(self, tmp_path: Path) -> None:
"""An empty leftover part file is a fresh download, not a resume."""
dest = tmp_path / "tool.tar.gz"
(tmp_path / "tool.tar.gz.part").write_bytes(b"")
with patch("requests.get", return_value=_mock_response(b"data")) as mock_get:
download_with_resume("https://example.com/t", dest)
assert mock_get.call_args[1]["headers"] == {}
assert dest.read_bytes() == b"data"
class TestDownloadFromMirrors:
def test_success_returns_url_and_writes_content(self, tmp_path: Path) -> None:
target = tmp_path / "out.bin"
@@ -640,7 +1203,8 @@ class TestDownloadFromMirrors:
ei.value
)
def test_falls_back_to_second_mirror(self, tmp_path: Path) -> None:
def test_falls_back_to_second_mirror(self) -> None:
buf = io.BytesIO()
with patch(
"requests.get",
side_effect=[_mock_response(b"", ok=False), _mock_response(b"second")],
@@ -648,14 +1212,152 @@ class TestDownloadFromMirrors:
url = download_from_mirrors(
["https://mirror1.com/f", "https://mirror2.com/f"],
{},
tmp_path / "out.bin",
buf,
)
assert url == "https://mirror2.com/f"
assert (tmp_path / "out.bin").read_bytes() == b"second"
assert buf.getvalue() == b"second"
def test_all_mirrors_fail_raises_error_listing_every_attempt(
self, tmp_path: Path
) -> None:
def test_mid_stream_drop_resumes_same_mirror(self) -> None:
"""A mid-stream failure retries the same mirror with Range and
If-Range headers, keeping the bytes already received, before falling
to the next."""
first = _interrupted_response(b"1234", etag='"v1"')
first.headers = {**first.headers, "content-length": "8"}
buf = io.BytesIO()
with patch(
"requests.get",
side_effect=[first, _resumed_response(b"5678")],
) as mock_get:
url = download_from_mirrors(
["https://mirror1.com/f", "https://mirror2.com/f"],
{},
buf,
)
assert url == "https://mirror1.com/f"
assert buf.getvalue() == b"12345678"
assert mock_get.call_count == 2
assert mock_get.call_args_list[1][0][0] == "https://mirror1.com/f"
# the resume is conditional on the content being unchanged
assert mock_get.call_args_list[1][1]["headers"] == {
"Range": "bytes=4-",
"If-Range": '"v1"',
}
def test_mid_stream_drop_without_validator_restarts(self) -> None:
"""A server offering no ETag/Last-Modified cannot be resumed safely;
the retry restarts from zero instead of stitching unverified bytes."""
buf = io.BytesIO()
with patch(
"requests.get",
side_effect=[_interrupted_response(b"1234"), _mock_response(b"full")],
) as mock_get:
download_from_mirrors(["https://mirror1.com/f"], {}, buf)
assert buf.getvalue() == b"full"
assert "Range" not in mock_get.call_args_list[1][1]["headers"]
def test_drop_after_last_byte_recovers_via_416(self) -> None:
"""A connection drop after the final body byte leaves a complete file;
the retry's 416 answer plus the length check turn it into success
instead of a wasted refetch."""
first = _interrupted_response(b"1234", etag='"v1"')
first.headers = {**first.headers, "content-length": "4"}
r416 = _mock_response(b"", ok=False)
r416.status_code = 416
buf = io.BytesIO()
with patch("requests.get", side_effect=[first, r416]) as mock_get:
url = download_from_mirrors(["https://mirror1.com/f"], {}, buf)
assert url == "https://mirror1.com/f"
assert buf.getvalue() == b"1234"
assert mock_get.call_count == 2
def test_mirror_drop_without_length_restarts(self) -> None:
"""With no content-length there is no way to prove a stitched file
complete, so the retry restarts even though a validator exists."""
buf = io.BytesIO()
with patch(
"requests.get",
side_effect=[
_interrupted_response(b"1234", etag='"v1"'),
_mock_response(b"full"),
],
) as mock_get:
download_from_mirrors(["https://mirror1.com/f"], {}, buf)
assert buf.getvalue() == b"full"
assert "Range" not in mock_get.call_args_list[1][1]["headers"]
def test_path_target_resumes_across_runs(self, tmp_path: Path) -> None:
"""A path target routes through download_with_resume: a part file and
metadata from a previous run resume instead of restarting."""
dest = tmp_path / "idf.tar.xz"
(tmp_path / "idf.tar.xz.part").write_bytes(b"1234")
(tmp_path / "idf.tar.xz.part.meta").write_text(
json.dumps(
{"url": "https://mirror1.com/f", "validator": '"v1"', "total": 8}
)
)
with patch("requests.get", return_value=_resumed_response(b"5678")) as mock_get:
url = download_from_mirrors(["https://mirror1.com/f"], {}, dest)
assert url == "https://mirror1.com/f"
assert dest.read_bytes() == b"12345678"
assert mock_get.call_args[1]["headers"] == {
"Range": "bytes=4-",
"If-Range": '"v1"',
}
def test_path_target_falls_back_to_next_mirror(self, tmp_path: Path) -> None:
dest = tmp_path / "idf.tar.xz"
with patch(
"requests.get",
side_effect=[req.ConnectionError("down"), _mock_response(b"data")],
):
url = download_from_mirrors(
["https://mirror1.com/f", "https://mirror2.com/f"], {}, dest
)
assert url == "https://mirror2.com/f"
assert dest.read_bytes() == b"data"
def test_resumed_short_body_fails_length_check(self) -> None:
"""A stitched file whose final length disagrees with the advertised
total is rejected instead of reported as success."""
first = _interrupted_response(b"1234", etag='"v1"')
first.headers = {**first.headers, "content-length": "8"}
# the resume ends early (5 of 8 bytes); the poisoned part is then
# discarded and the fresh retry also delivers a short body
short_resume = _resumed_response(b"5")
short_fresh = _mock_response(b"56")
short_fresh.headers = {**short_fresh.headers, "content-length": "8"}
buf = io.BytesIO()
with (
patch("requests.get", side_effect=[first, short_resume, short_fresh]),
pytest.raises(EsphomeError, match="all mirrors"),
):
download_from_mirrors(["https://mirror1.com/f"], {}, buf)
def test_failed_mirror_leftovers_not_kept_for_next_mirror(self) -> None:
"""Bytes from a mirror that failed all attempts must not leak into the
next mirror's download (no bogus Range request, fresh content)."""
exhausted = [_interrupted_response(b"AAAA", etag='"a1"')]
for _ in range(2):
r = _interrupted_response(b"BB")
r.status_code = 206
exhausted.append(r)
buf = io.BytesIO()
with patch(
"requests.get",
side_effect=exhausted + [_mock_response(b"clean")],
) as mock_get:
url = download_from_mirrors(
["https://mirror1.com/f", "https://mirror2.com/f"],
{},
buf,
)
assert url == "https://mirror2.com/f"
assert buf.getvalue() == b"clean"
# the second mirror starts fresh, without a Range header
assert mock_get.call_args_list[3][0][0] == "https://mirror2.com/f"
assert "Range" not in mock_get.call_args_list[3][1]["headers"]
def test_all_mirrors_fail_raises_error_listing_every_attempt(self) -> None:
with (
patch(
"requests.get",
@@ -666,7 +1368,7 @@ class TestDownloadFromMirrors:
download_from_mirrors(
["https://mirror1.com/f", "https://mirror2.com/f"],
{},
tmp_path / "out.bin",
io.BytesIO(),
)
# Every attempted URL appears in the message, and the first mirror's
# exception (the primary URL, usually the one that matters) is chained.