Merge branch 'esp8266-native-ninja-emission' into esp8266-arduino-toolchain

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:

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"<html>not json</html>")
+        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")