github/esphome
2
0
Fork 0
mirror of https://github.com/esphome/esphome.git synced 2026-08-22 22:26:21 +00:00
Code Issues Packages Projects Releases Wiki Activity

[core] Improve framework mirror selection, download errors, and version parsing (#17615)

Browse Source
This commit is contained in:
Jonathan Swoboda
2026-07-16 15:22:19 -04:00
committed by GitHub
parent f53394e46f
commit c0e78a5574
7 changed files with 281 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files
esphome/config_validation.py
+4 -2
View File
@@ -422,12 +422,14 @@ class Version:
@classmethod
def parse(cls, value: str) -> Version:
match = re.match(r"^(\d+).(\d+).(\d+)[-.]?(\w*)$", value)
# The patch component is optional and defaults to 0, so "6.0" and
# "6.0-rc1" parse as 6.0.0 and 6.0.0-rc1.
match = re.match(r"^(\d+)\.(\d+)(?:\.(\d+))?[-.]?(\w*)$", value)
if match is None:
raise ValueError(f"Not a valid version number {value}")
major = int(match[1])
minor = int(match[2])
patch = int(match[3])
patch = int(match[3] or 0)
extra = match[4] or ""
return Version(major=major, minor=minor, patch=patch, extra=extra)
esphome/espidf/framework.py
+24 -7
View File
@@ -63,7 +63,7 @@ ESPHOME_IDF_FRAMEWORK_MIRRORS = str_to_lst_of_str(
os.environ.get("ESPHOME_IDF_FRAMEWORK_MIRRORS")
or [
"https://github.com/esphome-libs/esp-idf/releases/download/v{VERSION}/esp-idf-v{VERSION}.tar.xz",
"https://github.com/esphome-libs/esp-idf/releases/download/v{MAJOR}.{MINOR}{EXTRA}/esp-idf-v{MAJOR}.{MINOR}{EXTRA}.tar.xz",
"https://github.com/esphome-libs/esp-idf/releases/download/v{SHORT_VERSION}/esp-idf-v{SHORT_VERSION}.tar.xz",
]
)
@@ -536,10 +536,14 @@ def _check_esphome_idf_framework_install(
env: Optional dictionary of environment variables to set
source_url: Optional override URL for the framework tarball. Supports
the same ``{VERSION}`` / ``{MAJOR}`` / ``{MINOR}`` / ``{PATCH}`` /
``{EXTRA}`` substitutions as ESPHOME_IDF_FRAMEWORK_MIRRORS
(``{EXTRA}`` includes its leading ``-``, e.g. ``-rc1``, or is empty).
When set, it replaces the default mirror list — no implicit fallback,
so a misspelled URL fails loudly.
``{EXTRA}`` / ``{SHORT_VERSION}`` substitutions as
ESPHOME_IDF_FRAMEWORK_MIRRORS (``{EXTRA}`` includes its leading
``-``, e.g. ``-rc1``, or is empty; ``{SHORT_VERSION}`` is ``x.y``
plus any extra and only available for x.y.0 versions — a URL
referencing it is skipped for other versions). When set, it
replaces the default mirror list — no implicit fallback, so a
misspelled or skipped URL fails loudly with an EsphomeError naming
the URL.
Returns:
tuple of (framework_path, install_flag)
@@ -588,7 +592,11 @@ def _check_esphome_idf_framework_install(
with tempfile.NamedTemporaryFile() as tmp:
_LOGGER.info("Downloading ESP-IDF %s framework ...", version)
# Create substitutions for the URLs
# 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)
@@ -596,8 +604,17 @@ def _check_esphome_idf_framework_install(
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:
pass
_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_from_mirrors(mirrors, substitutions, tmp.file)
esphome/framework_helpers.py
+62 -7
View File
@@ -552,6 +552,17 @@ def archive_extract_all(
matched_fct(archive_ref, extract_dir, progress_header=progress_header)
def _failure_reason(e: Exception) -> str:
"""Format a download exception for the aggregated error message.
``requests`` appends " for url: <url>" to HTTP errors; the URL is already
printed on the line above, so strip the suffix to keep lines short. Falls
back to the repr for exceptions with no message (e.g. ``TimeoutError()``)
so the line always names the failure.
"""
return str(e).split(" for url: ", maxsplit=1)[0] or repr(e)
def download_from_mirrors(
mirrors: list[str],
substitutions: dict[str, str],
@@ -570,14 +581,22 @@ def download_from_mirrors(
Returns:
The source URL.
Mirror URL templates that reference a substitution not present in
``substitutions`` are skipped, so callers can offer templates that only
apply to some downloads.
Raises:
ValueError: If mirrors list is empty.
Exception: If all download attempts fail.
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.
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)):
@@ -590,13 +609,31 @@ def download_from_mirrors(
)
# 2. Try each mirror in order
last_exception = None
failures: list[tuple[str, Exception]] = []
skipped: list[tuple[str, str]] = []
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 downloading from %s", url)
_LOGGER.debug("Trying to download from %s", url)
try:
# 4. Reset file pointer and download
@@ -631,9 +668,27 @@ def download_from_mirrors(
except Exception as e: # noqa: BLE001 # pylint: disable=broad-exception-caught
_LOGGER.debug("Failed to download %s: %s", url, str(e))
last_exception = e
failures.append((url, e))
# 7. Raise last exception if all mirrors failed
if last_exception:
raise last_exception
# 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")
tests/component_tests/esp32/test_esp32.py
+25
View File
@@ -740,3 +740,28 @@ def test_signed_ota_keys_invalid_combinations(config: dict, match: str) -> None:
with pytest.raises(cv.Invalid, match=match):
_validate_signed_ota_keys(config)
@pytest.mark.parametrize(
("value", "expected"),
[
# Full x.y.z versions are rewritten into pioarduino release URLs
(
"55.3.30",
"https://github.com/pioarduino/platform-espressif32/releases/download/55.03.30/platform-espressif32.zip",
),
(
"55.3.31-2",
"https://github.com/pioarduino/platform-espressif32/releases/download/55.03.31-2/platform-espressif32.zip",
),
# Non-version values pass through untouched
(
"https://github.com/pioarduino/platform-espressif32.git#develop",
"https://github.com/pioarduino/platform-espressif32.git#develop",
),
],
)
def test_parse_pio_platform_version(value: str, expected: str) -> None:
from esphome.components.esp32 import _parse_pio_platform_version
assert _parse_pio_platform_version(value) == expected
tests/unit_tests/test_config_validation.py
+34 -2
View File
@@ -1436,9 +1436,41 @@ def test_version_parse_with_extra() -> None:
assert version.extra == "dev20240101"
def test_version_parse_invalid() -> None:
def test_version_parse_without_patch() -> None:
"""A two-part version parses with patch defaulting to 0, so framework
shorthands like '6.0' and '6.0-rc1' are accepted."""
version = cv.Version.parse("6.0")
assert (version.major, version.minor, version.patch, version.extra) == (
6,
0,
0,
"",
)
version = cv.Version.parse("6.0-rc1")
assert (version.major, version.minor, version.patch, version.extra) == (
6,
0,
0,
"rc1",
)
def test_version_parse_numeric_extra() -> None:
"""Four-part versions keep the trailing component as extra (pioarduino
packaging revisions, e.g. 5.5.3.1)."""
version = cv.Version.parse("5.5.3.1")
assert (version.major, version.minor, version.patch, version.extra) == (
5,
5,
3,
"1",
)
@pytest.mark.parametrize("value", ["not.a.version", "6", "a.b", ""])
def test_version_parse_invalid(value: str) -> None:
with pytest.raises(ValueError, match="Not a valid version number"):
cv.Version.parse("not.a.version")
cv.Version.parse(value)
def test_version_is_beta() -> None:
tests/unit_tests/test_espidf_framework.py
+23
View File
@@ -489,6 +489,29 @@ def test_check_esp_idf_install_unparseable_version(
espidf_mocks.extract.assert_called_once()
@pytest.mark.parametrize(
("version", "short_version"),
[
("6.0.0", "6.0"),
("6.0.0-rc1", "6.0-rc1"),
("5.5.4", None), # vX.Y tags only exist for X.Y.0 releases
],
)
def test_check_esp_idf_install_short_version_substitution(
espidf_mocks: SimpleNamespace, version: str, short_version: str | None
) -> None:
"""SHORT_VERSION is only offered for x.y.0 releases, so the vX.Y mirror
template is never tried for versions whose tag cannot exist."""
_get_framework_path(version).mkdir(parents=True, exist_ok=True)
check_esp_idf_install(version, force=True)
# First call downloads the framework archive; a later call fetches the
# constraints file with its own substitutions.
substitutions = espidf_mocks.download.call_args_list[0][0][1]
assert substitutions.get("SHORT_VERSION") == short_version
assert substitutions["VERSION"] == version
# ---------------------------------------------------------------------------
# _patch_tools_json_for_linux_arm64 (arm64-only ninja backport)
# ---------------------------------------------------------------------------
tests/unit_tests/test_framework_helpers.py
+108 -3
View File
@@ -16,6 +16,7 @@ import zipfile
import pytest
import requests as req
from esphome.core import EsphomeError
from esphome.framework_helpers import (
_7z_extract_all,
_detect_archive_root,
@@ -546,6 +547,99 @@ class TestDownloadFromMirrors:
)
assert mock_get.call_args[0][0] == "https://example.com/1.2.3.bin"
def test_template_with_missing_substitution_is_skipped(
self, tmp_path: Path
) -> None:
"""A template referencing an unavailable substitution is skipped, not
formatted into a bogus URL (e.g. SHORT_VERSION only exists for x.y.0
framework versions)."""
with patch(
"requests.get",
return_value=_mock_response(b"x"),
) as mock_get:
url = download_from_mirrors(
[
"https://example.com/{SHORT_VERSION}.bin",
"https://example.com/{VERSION}.bin",
],
{"VERSION": "1.2.3"},
tmp_path / "out.bin",
)
assert url == "https://example.com/1.2.3.bin"
assert mock_get.call_count == 1
def test_all_templates_skipped_raises_esphome_error(self, tmp_path: Path) -> None:
with (
patch("requests.get") as mock_get,
pytest.raises(EsphomeError, match="No mirror URL template matched") as ei,
):
download_from_mirrors(
["https://example.com/{MISSING}.bin"],
{"VERSION": "1.2.3"},
tmp_path / "out.bin",
)
mock_get.assert_not_called()
# The skipped template and its missing substitution are named
assert "https://example.com/{MISSING}.bin" in str(ei.value)
assert "MISSING" in str(ei.value)
def test_failure_message_includes_skipped_templates(self, tmp_path: Path) -> None:
"""When downloads fail, templates that were skipped for missing
substitutions are also listed so a typo'd custom mirror is
attributable."""
with (
patch(
"requests.get",
return_value=_mock_response(b"", ok=False),
),
pytest.raises(EsphomeError, match="all mirrors") as ei,
):
download_from_mirrors(
[
"https://example.com/{TYPO}.bin",
"https://example.com/{VERSION}.bin",
],
{"VERSION": "1.2.3"},
tmp_path / "out.bin",
)
message = str(ei.value)
assert "https://example.com/1.2.3.bin" in message
assert (
"https://example.com/{TYPO}.bin\n not applicable (TYPO not available)"
in message
)
def test_malformed_template_warns_and_is_reported(
self, tmp_path: Path, caplog: pytest.LogCaptureFixture
) -> None:
"""A structurally malformed template is an authoring error: warned
about even when another mirror succeeds, and named in the aggregate
error when everything fails."""
with (
patch("requests.get", return_value=_mock_response(b"x")),
caplog.at_level(logging.WARNING, logger="esphome.framework_helpers"),
):
url = download_from_mirrors(
["https://example.com/{oops.bin", "https://example.com/{VERSION}.bin"],
{"VERSION": "1.2.3"},
tmp_path / "out.bin",
)
assert url == "https://example.com/1.2.3.bin"
assert "malformed mirror URL template" in caplog.text
with (
patch("requests.get", return_value=_mock_response(b"", ok=False)),
pytest.raises(EsphomeError, match="all mirrors") as ei,
):
download_from_mirrors(
["https://example.com/{oops.bin", "https://example.com/{VERSION}.bin"],
{"VERSION": "1.2.3"},
tmp_path / "out.bin",
)
assert "https://example.com/{oops.bin\n skipped (ValueError(" in str(
ei.value
)
def test_falls_back_to_second_mirror(self, tmp_path: Path) -> None:
with patch(
"requests.get",
@@ -559,15 +653,26 @@ class TestDownloadFromMirrors:
assert url == "https://mirror2.com/f"
assert (tmp_path / "out.bin").read_bytes() == b"second"
def test_all_mirrors_fail_reraises_last_exception(self, tmp_path: Path) -> None:
def test_all_mirrors_fail_raises_error_listing_every_attempt(
self, tmp_path: Path
) -> None:
with (
patch(
"requests.get",
return_value=_mock_response(b"", ok=False),
),
pytest.raises(req.HTTPError),
pytest.raises(EsphomeError, match="all mirrors") as excinfo,
):
download_from_mirrors(["https://example.com/f"], {}, tmp_path / "out.bin")
download_from_mirrors(
["https://mirror1.com/f", "https://mirror2.com/f"],
{},
tmp_path / "out.bin",
)
# Every attempted URL appears in the message, and the first mirror's
# exception (the primary URL, usually the one that matters) is chained.
assert "https://mirror1.com/f" in str(excinfo.value)
assert "https://mirror2.com/f" in str(excinfo.value)
assert isinstance(excinfo.value.__cause__, req.HTTPError)
def test_empty_mirrors_raises_value_error(self, tmp_path: Path) -> None:
with pytest.raises(ValueError, match="empty mirrors list"):