diff --git a/esphome/arduino8266/framework.py b/esphome/arduino8266/framework.py index 5c2d9f4821..cf7506412f 100644 --- a/esphome/arduino8266/framework.py +++ b/esphome/arduino8266/framework.py @@ -19,12 +19,13 @@ from __future__ import annotations from collections.abc import Collection import functools +import io +import json import logging import os from pathlib import Path import platform import shutil -import time from esphome.core import EsphomeError, Version from esphome.framework_helpers import ( @@ -93,35 +94,40 @@ def _downloads_path() -> Path: return path +# (system, machine) -> registry tag, both lowercased. The windows-arm64 and +# darwin-arm64 mappings are deliberate: the toolchain packages ship x86_64 +# binaries for those hosts (Rosetta / x86 emulation). +_SYSTEM_TAGS: dict[tuple[str, str], str] = { + ("darwin", "arm64"): "darwin_arm64", + ("darwin", "x86_64"): "darwin_x86_64", + ("windows", "amd64"): "windows_amd64", + ("windows", "arm64"): "windows_amd64", + ("windows", "x86"): "windows_x86", + ("windows", "i686"): "windows_x86", + ("windows", "i386"): "windows_x86", + ("linux", "x86_64"): "linux_x86_64", + ("linux", "amd64"): "linux_x86_64", + ("linux", "aarch64"): "linux_aarch64", + ("linux", "arm64"): "linux_aarch64", + ("linux", "i686"): "linux_i686", + ("linux", "i386"): "linux_i686", + ("linux", "x86"): "linux_i686", +} + + def _pio_system() -> str: """The PlatformIO registry system tag for the current host. - Hand-rolled instead of ``platformio.util.get_systype()`` so this backend - never imports the PlatformIO package. The windows-arm64 and darwin-arm64 - mappings are deliberate: the toolchain packages ship x86_64 binaries for - those hosts (Rosetta / x86 emulation). + A local table instead of ``platformio.util.get_systype()`` so this + backend never imports the PlatformIO package. """ sysname = platform.system().lower() machine = platform.machine().lower() - if sysname == "darwin": - if machine == "arm64": - return "darwin_arm64" - if machine == "x86_64": - return "darwin_x86_64" - if sysname == "windows": - if machine in ("amd64", "arm64"): - return "windows_amd64" - if machine in ("x86", "i686", "i386"): - return "windows_x86" - if sysname == "linux": - if machine in ("arm64", "aarch64"): - return "linux_aarch64" - if machine in ("i686", "i386", "x86"): - return "linux_i686" - if machine.startswith("arm"): - return f"linux_{machine}" - if machine in ("x86_64", "amd64"): - return "linux_x86_64" + if tag := _SYSTEM_TAGS.get((sysname, machine)): + return tag + if sysname == "linux" and machine.startswith("arm"): + # 32-bit arm tags carry the exact machine name (armv6l, armv7l, ...) + return f"linux_{machine}" # Fail here, near the cause, rather than installing a toolchain whose # binaries cannot execute on this host. raise EsphomeError( @@ -131,26 +137,19 @@ def _pio_system() -> str: def _registry_download(package: str, version: str) -> tuple[str, str, int | None]: - """Resolve a package's download URL, sha256, and size via the PIO registry.""" - import requests + """Resolve a package's download URL, sha256, and size via the PIO registry. - url = _REGISTRY_URL.format(package=package) - last_err: Exception | None = None - for attempt in range(3): - try: - resp = requests.get(url, timeout=30) - resp.raise_for_status() - data = resp.json() - break - except requests.RequestException as err: - last_err = err - # Back off so the retries are not one burst against a hiccup - time.sleep(2**attempt) - else: - # A clean, retried error like the other download paths in the tree + The metadata fetch goes through ``download_from_mirrors`` so it shares + the retry, backoff, and error reporting of every other download here. + """ + buf = io.BytesIO() + download_from_mirrors([_REGISTRY_URL], {"package": package}, buf) + try: + data = json.loads(buf.getvalue()) + except ValueError as err: raise EsphomeError( - f"Could not query the package registry for {package}: {last_err}" - ) from last_err + f"The package registry returned invalid JSON for {package}: {err}" + ) from err system = _pio_system() for ver in data.get("versions", []): if ver.get("name") != version: diff --git a/tests/unit_tests/test_arduino8266_framework.py b/tests/unit_tests/test_arduino8266_framework.py index 3ca9e70ee2..357afc5fbb 100644 --- a/tests/unit_tests/test_arduino8266_framework.py +++ b/tests/unit_tests/test_arduino8266_framework.py @@ -3,6 +3,7 @@ from __future__ import annotations from contextlib import contextmanager +import json import os from pathlib import Path import subprocess @@ -79,66 +80,59 @@ def test_pio_system_unsupported_host_raises(system: str, machine: str) -> None: framework._pio_system() -def _registry_response(files: list[dict]) -> MagicMock: - resp = MagicMock() - resp.json.return_value = {"versions": [{"name": "1.0.0", "files": files}]} - return resp +def _registry_response(files: list[dict]): + """Patch the shared downloader to serve a canned registry response.""" + payload = {"versions": [{"name": "1.0.0", "files": files}]} + + def fake_download(mirrors: list[str], substitutions: dict, target) -> str: + target.write(json.dumps(payload).encode()) + return mirrors[0].format(**substitutions) + + return patch.object(framework, "download_from_mirrors", side_effect=fake_download) -def test_registry_download_network_error_is_clean_and_retried() -> None: - """Registry failures raise EsphomeError after retries, not a traceback.""" - import requests - +def test_registry_download_uses_shared_downloader() -> None: + """The metadata fetch delegates its retries and error reporting to + download_from_mirrors; failures surface unchanged.""" with ( - patch("requests.get", side_effect=requests.ConnectionError("boom")) as mock_get, - patch.object(framework.time, "sleep") as mock_sleep, - pytest.raises(EsphomeError, match="Could not query the package registry"), + patch.object( + framework, + "download_from_mirrors", + side_effect=EsphomeError("Failed to download from all mirrors"), + ) as mock_download, + pytest.raises(EsphomeError, match="Failed to download from all mirrors"), ): framework._registry_download("pkg", "1.0.0") - assert mock_get.call_count == 3 - # Backed-off retries, not one burst - assert mock_sleep.call_count == 3 + (mirrors, substitutions, _), _ = mock_download.call_args + assert mirrors == [framework._REGISTRY_URL] + assert substitutions == {"package": "pkg"} -def test_registry_download_retries_transient_error() -> None: - import requests +def test_registry_download_invalid_json_is_clean() -> None: + def fake_download(mirrors: list[str], substitutions: dict, target) -> str: + target.write(b"not json") + return "http://x" - resp = _registry_response( - [ - { - "system": ["linux_x86_64"], - "download_url": "http://x/linux", - "checksum": {"sha256": "abc123"}, - "size": 42, - } - ] - ) with ( - patch("requests.get", side_effect=[requests.ConnectionError("boom"), resp]), - patch.object(framework.time, "sleep"), - patch.object(framework, "_pio_system", return_value="linux_x86_64"), + patch.object(framework, "download_from_mirrors", side_effect=fake_download), + pytest.raises(EsphomeError, match="invalid JSON"), ): - assert framework._registry_download("pkg", "1.0.0") == ( - "http://x/linux", - "abc123", - 42, - ) + framework._registry_download("pkg", "1.0.0") def test_registry_download_matches_system() -> None: - resp = _registry_response( - [ - {"system": ["windows_amd64"], "download_url": "http://x/win"}, - { - "system": ["linux_x86_64"], - "download_url": "http://x/linux", - "checksum": {"sha256": "abc123"}, - "size": 42, - }, - ] - ) with ( - patch("requests.get", return_value=resp), + _registry_response( + [ + {"system": ["windows_amd64"], "download_url": "http://x/win"}, + { + "system": ["linux_x86_64"], + "download_url": "http://x/linux", + "checksum": {"sha256": "abc123"}, + "size": 42, + }, + ] + ), patch.object(framework, "_pio_system", return_value="linux_x86_64"), ): assert framework._registry_download("pkg", "1.0.0") == ( @@ -150,25 +144,24 @@ def test_registry_download_matches_system() -> None: def test_registry_download_bare_string_system() -> None: """A bare-string system tag is an exact match, not a substring test.""" - resp = _registry_response( - [ - {"system": "linux_x86", "download_url": "http://x/x86"}, - { - "system": "linux_x86_64", - "download_url": "http://x/x86_64", - "checksum": {"sha256": "abc"}, - }, - ] - ) with ( - patch("requests.get", return_value=resp), + _registry_response( + [ + {"system": "linux_x86", "download_url": "http://x/x86"}, + { + "system": "linux_x86_64", + "download_url": "http://x/x86_64", + "checksum": {"sha256": "abc"}, + }, + ] + ), patch.object(framework, "_pio_system", return_value="linux_x86_64"), ): assert framework._registry_download("pkg", "1.0.0")[0] == "http://x/x86_64" def test_registry_download_wildcard_system() -> None: - resp = _registry_response( + with _registry_response( [ { "system": "*", @@ -177,8 +170,7 @@ def test_registry_download_wildcard_system() -> None: "size": 7, } ] - ) - with patch("requests.get", return_value=resp): + ): assert framework._registry_download("pkg", "1.0.0") == ( "http://x/any", "abc", @@ -188,20 +180,18 @@ def test_registry_download_wildcard_system() -> None: def test_registry_download_missing_checksum_raises() -> None: """An unverifiable archive is refused, never silently extracted.""" - resp = _registry_response([{"system": "*", "download_url": "http://x/any"}]) with ( - patch("requests.get", return_value=resp), + _registry_response([{"system": "*", "download_url": "http://x/any"}]), pytest.raises(EsphomeError, match="no sha256"), ): framework._registry_download("pkg", "1.0.0") def test_registry_download_no_system_match() -> None: - resp = _registry_response( - [{"system": ["windows_amd64"], "download_url": "http://x/win"}] - ) with ( - patch("requests.get", return_value=resp), + _registry_response( + [{"system": ["windows_amd64"], "download_url": "http://x/win"}] + ), patch.object(framework, "_pio_system", return_value="linux_x86_64"), pytest.raises(EsphomeError, match="No pkg 1.0.0 build"), ): @@ -209,10 +199,14 @@ def test_registry_download_no_system_match() -> None: def test_registry_download_version_not_found() -> None: - resp = MagicMock() - resp.json.return_value = {"versions": [{"name": "2.0.0", "files": []}]} + def fake_download(mirrors: list[str], substitutions: dict, target) -> str: + target.write( + json.dumps({"versions": [{"name": "2.0.0", "files": []}]}).encode() + ) + return "http://x" + with ( - patch("requests.get", return_value=resp), + patch.object(framework, "download_from_mirrors", side_effect=fake_download), pytest.raises(EsphomeError, match="not found"), ): framework._registry_download("pkg", "1.0.0")