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

This commit is contained in:
Jonathan Swoboda
2026-07-21 08:18:14 +12:00
committed by Jesse Hills
parent ea01c909b7
commit 95a01ac2ab
7 changed files with 281 additions and 22 deletions
+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"):