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

Move the PlatformIO registry installer to esphome.platformio.registry and transliterate get_systype

Browse Source
This commit is contained in:
J. Nick Koston
2026-08-20 15:25:12 -05:00
parent 38b6f222b5
commit 113259275c
4 changed files with 478 additions and 444 deletions
Download Patch File Download Diff File Expand all files Collapse all files
esphome/arduino8266/framework.py
+8 -160
View File
@@ -7,7 +7,8 @@ ESP-IDF install in ``esphome.espidf.framework``):
<cache>/arduino8266/toolchains/<version>/ toolchain-xtensa (gcc 10.3)
ninja itself comes from PATH or the ninja PyPI wheel (a requirements.txt
dependency), so only the two packages above are downloaded here.
dependency), so only the two packages above are downloaded, via the shared
PlatformIO-registry installer in ``esphome.platformio.registry``.
Sources default to the PlatformIO registry (the exact packages the PlatformIO
toolchain has always used, so the bits are identical); the
@@ -17,27 +18,20 @@ toolchain has always used, so the bits are identical); the
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
from esphome.core import EsphomeError, Version
from esphome.framework_helpers import (
archive_extract_all,
ccache_defaults_env,
download_from_mirrors,
download_with_resume,
resolve_ccache_path,
rmdir,
str_to_lst_of_str,
tools_cache_path,
)
from esphome.platformio.registry import install_package
_LOGGER = logging.getLogger(__name__)
@@ -48,10 +42,6 @@ TOOLCHAIN_PACKAGE = "toolchain-xtensa"
# reinstall (the install dir is keyed on the version).
TOOLCHAIN_VERSION = "2.100300.220621"
_REGISTRY_URL = (
"https://api.registry.platformio.org/v3/packages/platformio/tool/{package}"
)
ESPHOME_ARDUINO8266_FRAMEWORK_MIRRORS = str_to_lst_of_str(
os.environ.get("ESPHOME_ARDUINO8266_FRAMEWORK_MIRRORS", "")
)
@@ -88,151 +78,6 @@ def get_toolchain_path() -> Path:
return get_arduino8266_tools_path() / "toolchains" / TOOLCHAIN_VERSION
def _downloads_path() -> Path:
path = get_arduino8266_tools_path() / "downloads"
path.mkdir(parents=True, exist_ok=True)
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.
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 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(
f"No {sysname}/{machine} build of the ESP8266 toolchain exists; "
"use 'toolchain: platformio'"
)
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.
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"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:
continue
for file in ver.get("files", []):
# A bare string would make ``in`` a substring test
systems = file.get("system") or "*"
if isinstance(systems, str):
systems = [systems]
if "*" in systems or system in systems:
sha256 = (file.get("checksum") or {}).get("sha256")
if not sha256:
# Never extract an unverified archive; the registry
# publishes a checksum for every package file.
raise EsphomeError(
f"The package registry returned no sha256 for "
f"{package} {version}; refusing the unverified download"
)
return (file["download_url"], sha256, file.get("size"))
raise EsphomeError(f"No {package} {version} build for this platform ({system})")
raise EsphomeError(f"{package} {version} not found in the package registry")
def _install_package(
name: str,
version: str,
dest: Path,
mirrors: list[str],
expect: Collection[str] = (),
) -> None:
"""Download, verify, and extract one package if not already installed.
The registry path is integrity-checked against the sha256 the registry
publishes; a mirror override is trusted as configured.
"""
marker = dest / ".esphome_extracted"
if marker.is_file():
return
from filelock import FileLock
# The cache is machine-global; serialize concurrent cold builds so one
# process cannot wipe the directory another is extracting into (same
# filelock pattern as platformio/toolchain.py and git.py).
dest.parent.mkdir(parents=True, exist_ok=True)
# fallback_to_soft would silently degrade to an existence lock on a
# flock-less filesystem; a hard-killed run would then hang every later
# build forever (same hazard git.py documents).
with FileLock(f"{dest}.lock", fallback_to_soft=False):
if marker.is_file():
# Another process finished the install while we waited
return
rmdir(dest, msg=f"Clean up incomplete {name} install")
# A persistent download location (not a temp dir) so an interrupted
# download resumes across esphome runs via download_with_resume's
# .part file, mirroring the espidf dist/ convention.
archive = _downloads_path() / f"{name}-{version}"
_LOGGER.info("Downloading %s %s ...", name, version)
if mirrors:
_LOGGER.warning(
"Downloading %s from a mirror override; checksum verification "
"is skipped for mirrors",
name,
)
download_from_mirrors(
mirrors, {"VERSION": version, "SYSTEM": _pio_system()}, archive
)
else:
url, sha256, size = _registry_download(name, version)
download_with_resume(url, archive, sha256=sha256, size=size)
_LOGGER.info("Extracting %s ...", name)
archive_extract_all(archive, dest, progress_header="Extracting")
# Validate the layout before recording success, so an unexpected
# package is never cached as a working install.
for rel in expect:
if not (dest / rel).is_dir():
raise EsphomeError(
f"{name} {version} extracted without the expected {rel} "
"directory; run 'esphome clean-all' and retry"
)
marker.touch()
archive.unlink(missing_ok=True)
def _find_ninja() -> Path:
"""Locate the ninja binary: PATH first, else the ninja PyPI wheel.
@@ -270,19 +115,22 @@ def check_and_install(framework_version: Version) -> dict[str, Path]:
ninja_path = _find_ninja()
package_version = framework_package_version(framework_version)
framework_path = get_framework_path(package_version)
_install_package(
downloads_dir = get_arduino8266_tools_path() / "downloads"
install_package(
FRAMEWORK_PACKAGE,
package_version,
framework_path,
ESPHOME_ARDUINO8266_FRAMEWORK_MIRRORS,
downloads_dir,
expect=("cores/esp8266", "tools/sdk", "libraries"),
)
toolchain_path = get_toolchain_path()
_install_package(
install_package(
TOOLCHAIN_PACKAGE,
TOOLCHAIN_VERSION,
toolchain_path,
ESPHOME_ARDUINO8266_TOOLCHAIN_MIRRORS,
downloads_dir,
expect=("bin",),
)
return {
esphome/platformio/registry.py
+159
View File
@@ -0,0 +1,159 @@
"""Install packages from the PlatformIO registry without PlatformIO.
Native toolchains install the exact registry packages the PlatformIO backend
uses, so the bits are identical, but resolve and verify them with esphome's
own download machinery instead of importing the platformio package.
"""
from __future__ import annotations
from collections.abc import Collection
import io
import json
import logging
import os
from pathlib import Path
import platform
from esphome.core import EsphomeError
from esphome.framework_helpers import (
archive_extract_all,
download_from_mirrors,
download_with_resume,
rmdir,
)
_LOGGER = logging.getLogger(__name__)
_REGISTRY_URL = (
"https://api.registry.platformio.org/v3/packages/platformio/tool/{package}"
)
def get_systype() -> str:
"""The registry system tag for the current host.
A transliteration of ``platformio.util.get_systype()``, honoring the same
``PLATFORMIO_SYSTEM_TYPE`` override, so this module never imports the
platformio package. One deviation: windows-arm64 maps straight to
``windows_amd64``: the registry ships no arm64 toolchains and those hosts
run x86 binaries via emulation, which upstream leaves to the override.
"""
if systype := os.environ.get("PLATFORMIO_SYSTEM_TYPE"):
return systype
system = platform.system().lower()
arch = platform.machine().lower()
if system == "windows":
if not arch: # same fallback as upstream (platformio issue #4353)
arch = "x86_" + platform.architecture()[0]
if "x86" in arch:
arch = "amd64" if "64" in arch else "x86"
elif arch == "arm64":
arch = "amd64"
if arch == "aarch64" and platform.architecture()[0] == "32bit":
# 64-bit kernel with a 32-bit userland (e.g. 32-bit Raspberry Pi OS)
arch = "armv7l"
return f"{system}_{arch}" if arch else system
def registry_download(package: str, version: str) -> tuple[str, str, int | None]:
"""Resolve a package's download URL, sha256, and size via the registry.
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"The package registry returned invalid JSON for {package}: {err}"
) from err
systype = get_systype()
for ver in data.get("versions", []):
if ver.get("name") != version:
continue
for file in ver.get("files", []):
# A bare string would make ``in`` a substring test
systems = file.get("system") or "*"
if isinstance(systems, str):
systems = [systems]
if "*" in systems or systype in systems:
sha256 = (file.get("checksum") or {}).get("sha256")
if not sha256:
# Never extract an unverified archive; the registry
# publishes a checksum for every package file.
raise EsphomeError(
f"The package registry returned no sha256 for "
f"{package} {version}; refusing the unverified download"
)
return (file["download_url"], sha256, file.get("size"))
raise EsphomeError(
f"No {package} {version} build for this platform ({systype})"
)
raise EsphomeError(f"{package} {version} not found in the package registry")
def install_package(
name: str,
version: str,
dest: Path,
mirrors: list[str],
downloads_dir: Path,
expect: Collection[str] = (),
) -> None:
"""Download, verify, and extract one package if not already installed.
The registry path is integrity-checked against the sha256 the registry
publishes; a mirror override (URL templates with ``{VERSION}``/``{SYSTEM}``
substitution) is trusted as configured. ``downloads_dir`` holds the
archive between runs so an interrupted download resumes.
"""
marker = dest / ".esphome_extracted"
if marker.is_file():
return
from filelock import FileLock
# The cache is machine-global; serialize concurrent cold builds so one
# process cannot wipe the directory another is extracting into (same
# filelock pattern as platformio/toolchain.py and git.py).
dest.parent.mkdir(parents=True, exist_ok=True)
# fallback_to_soft would silently degrade to an existence lock on a
# flock-less filesystem; a hard-killed run would then hang every later
# build forever (same hazard git.py documents).
with FileLock(f"{dest}.lock", fallback_to_soft=False):
if marker.is_file():
# Another process finished the install while we waited
return
rmdir(dest, msg=f"Clean up incomplete {name} install")
# A persistent download location (not a temp dir) so an interrupted
# download resumes across esphome runs via download_with_resume's
# .part file, mirroring the espidf dist/ convention.
downloads_dir.mkdir(parents=True, exist_ok=True)
archive = downloads_dir / f"{name}-{version}"
_LOGGER.info("Downloading %s %s ...", name, version)
if mirrors:
_LOGGER.warning(
"Downloading %s from a mirror override; checksum verification "
"is skipped for mirrors",
name,
)
download_from_mirrors(
mirrors, {"VERSION": version, "SYSTEM": get_systype()}, archive
)
else:
url, sha256, size = registry_download(name, version)
download_with_resume(url, archive, sha256=sha256, size=size)
_LOGGER.info("Extracting %s ...", name)
archive_extract_all(archive, dest, progress_header="Extracting")
# Validate the layout before recording success, so an unexpected
# package is never cached as a working install.
for rel in expect:
if not (dest / rel).is_dir():
raise EsphomeError(
f"{name} {version} extracted without the expected {rel} "
"directory; run 'esphome clean-all' and retry"
)
marker.touch()
archive.unlink(missing_ok=True)
tests/unit_tests/test_arduino8266_framework.py
+1 -284
View File
@@ -2,8 +2,6 @@
from __future__ import annotations
from contextlib import contextmanager
import json
import os
from pathlib import Path
import subprocess
@@ -38,226 +36,6 @@ def test_tools_path_default_and_prefix(tmp_path: Path) -> None:
assert path != Path.cwd()
@pytest.mark.parametrize(
("system", "machine", "expected"),
[
("Darwin", "arm64", "darwin_arm64"),
("Darwin", "x86_64", "darwin_x86_64"),
("Windows", "AMD64", "windows_amd64"),
("Windows", "ARM64", "windows_amd64"),
("Windows", "x86", "windows_x86"),
("Linux", "x86_64", "linux_x86_64"),
("Linux", "aarch64", "linux_aarch64"),
("Linux", "i686", "linux_i686"),
("Linux", "armv7l", "linux_armv7l"),
],
)
def test_pio_system(system: str, machine: str, expected: str) -> None:
with (
patch("platform.system", return_value=system),
patch("platform.machine", return_value=machine),
):
assert framework._pio_system() == expected
@pytest.mark.parametrize(
("system", "machine"),
[
("FreeBSD", "amd64"),
("Linux", "ppc64le"),
("Darwin", "ppc"),
("Darwin", ""),
("Windows", "ia64"),
],
)
def test_pio_system_unsupported_host_raises(system: str, machine: str) -> None:
# Fails at resolution rather than installing a toolchain that can't run
with (
patch("platform.system", return_value=system),
patch("platform.machine", return_value=machine),
pytest.raises(EsphomeError, match="use 'toolchain: platformio'"),
):
framework._pio_system()
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_uses_shared_downloader() -> None:
"""The metadata fetch delegates its retries and error reporting to
download_from_mirrors; failures surface unchanged."""
with (
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")
(mirrors, substitutions, _), _ = mock_download.call_args
assert mirrors == [framework._REGISTRY_URL]
assert substitutions == {"package": "pkg"}
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"
with (
patch.object(framework, "download_from_mirrors", side_effect=fake_download),
pytest.raises(EsphomeError, match="invalid JSON"),
):
framework._registry_download("pkg", "1.0.0")
def test_registry_download_matches_system() -> None:
with (
_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") == (
"http://x/linux",
"abc123",
42,
)
def test_registry_download_bare_string_system() -> None:
"""A bare-string system tag is an exact match, not a substring test."""
with (
_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:
with _registry_response(
[
{
"system": "*",
"download_url": "http://x/any",
"checksum": {"sha256": "abc"},
"size": 7,
}
]
):
assert framework._registry_download("pkg", "1.0.0") == (
"http://x/any",
"abc",
7,
)
def test_registry_download_missing_checksum_raises() -> None:
"""An unverifiable archive is refused, never silently extracted."""
with (
_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:
with (
_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"),
):
framework._registry_download("pkg", "1.0.0")
def test_registry_download_version_not_found() -> None:
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.object(framework, "download_from_mirrors", side_effect=fake_download),
pytest.raises(EsphomeError, match="not found"),
):
framework._registry_download("pkg", "1.0.0")
def test_install_package_skips_when_marker_exists(tmp_path: Path) -> None:
dest = tmp_path / "pkg"
dest.mkdir()
(dest / ".esphome_extracted").touch()
with patch.object(framework, "download_from_mirrors") as mock_download:
framework._install_package("pkg", "1.0.0", dest, [])
mock_download.assert_not_called()
def test_install_package_downloads_via_mirrors(tmp_path: Path) -> None:
dest = tmp_path / "pkg"
mirrors = ["http://mirror/{VERSION}/{SYSTEM}.tar.gz"]
with (
patch.object(framework, "download_from_mirrors") as mock_download,
patch.object(framework, "archive_extract_all") as mock_extract,
patch.object(framework, "_pio_system", return_value="linux_x86_64"),
):
# Extraction is expected to create the directory
mock_extract.side_effect = lambda *_a, **_kw: dest.mkdir()
framework._install_package("pkg", "1.0.0", dest, mirrors)
assert mock_download.call_args[0][0] is mirrors
assert mock_download.call_args[0][1] == {
"VERSION": "1.0.0",
"SYSTEM": "linux_x86_64",
}
assert (dest / ".esphome_extracted").is_file()
def test_install_package_downloads_via_registry(tmp_path: Path) -> None:
"""The registry path downloads with the registry's sha256 and size."""
dest = tmp_path / "pkg"
with (
patch.object(framework, "download_with_resume") as mock_download,
patch.object(framework, "archive_extract_all") as mock_extract,
patch.object(
framework,
"_registry_download",
return_value=("http://x/pkg.tar.gz", "abc123", 42),
),
):
mock_extract.side_effect = lambda *_a, **_kw: dest.mkdir()
framework._install_package("pkg", "1.0.0", dest, [])
assert mock_download.call_args[0][0] == "http://x/pkg.tar.gz"
assert mock_download.call_args[1] == {"sha256": "abc123", "size": 42}
def test_find_ninja_prefers_path(tmp_path: Path) -> None:
with patch("shutil.which", return_value=str(tmp_path / "ninja")):
assert framework._find_ninja() == tmp_path / "ninja"
@@ -298,7 +76,7 @@ def test_find_ninja_missing_everywhere(tmp_path: Path) -> None:
def test_check_and_install_returns_paths(tmp_path: Path) -> None:
with (
patch.dict(os.environ, {"ESPHOME_ARDUINO8266_PREFIX": str(tmp_path)}),
patch.object(framework, "_install_package") as mock_install,
patch.object(framework, "install_package") as mock_install,
patch.object(framework, "_find_ninja", return_value=tmp_path / "ninja"),
):
paths = framework.check_and_install(cv.Version(3, 1, 2))
@@ -385,53 +163,6 @@ def test_ccache_env(tmp_path: Path) -> None:
assert env["CCACHE_DIR"].endswith("ccache")
def test_install_package_validates_expected_layout(tmp_path: Path) -> None:
"""The success marker is only written when the extracted tree is usable."""
dest = tmp_path / "pkg"
with (
patch.object(framework, "download_from_mirrors"),
patch.object(framework, "archive_extract_all") as mock_extract,
patch.object(framework, "_pio_system", return_value="linux_x86_64"),
):
mock_extract.side_effect = lambda *_a, **_kw: (dest / "bin").mkdir(parents=True)
framework._install_package("pkg", "1.0.0", dest, ["http://m"], expect=("bin",))
assert (dest / ".esphome_extracted").is_file()
def test_install_package_unexpected_layout_raises(tmp_path: Path) -> None:
dest = tmp_path / "pkg"
with (
patch.object(framework, "download_from_mirrors"),
patch.object(framework, "archive_extract_all") as mock_extract,
patch.object(framework, "_pio_system", return_value="linux_x86_64"),
pytest.raises(EsphomeError, match="without the expected bin"),
):
mock_extract.side_effect = lambda *_a, **_kw: dest.mkdir()
framework._install_package("pkg", "1.0.0", dest, ["http://m"], expect=("bin",))
assert not (dest / ".esphome_extracted").exists()
def test_install_package_marker_rechecked_under_lock(tmp_path: Path) -> None:
"""A concurrent install finishing while we wait for the lock is detected."""
dest = tmp_path / "pkg"
marker = dest / ".esphome_extracted"
@contextmanager
def _fake_lock(*_a, **_kw):
dest.mkdir(parents=True, exist_ok=True)
marker.touch()
yield
with (
patch("filelock.FileLock", _fake_lock),
patch.object(framework, "download_from_mirrors") as mock_download,
patch.object(framework, "rmdir") as mock_rmdir,
):
framework._install_package("pkg", "1.0.0", dest, ["http://m"])
mock_download.assert_not_called()
mock_rmdir.assert_not_called()
def test_ccache_env_requires_build_path() -> None:
"""Building the env before preload set build_path fails loudly."""
CORE.build_path = None
@@ -446,17 +177,3 @@ def test_check_and_install_rejects_old_core(tmp_path: Path) -> None:
"""Calling the installer below the floor fails before any download."""
with pytest.raises(EsphomeError, match=">= 3.1.1"):
framework.check_and_install(cv.Version(3, 0, 2))
def test_install_package_uses_hard_lock(tmp_path: Path) -> None:
"""The install lock must never degrade to a soft (existence) lock."""
dest = tmp_path / "pkg"
with (
patch("filelock.FileLock") as mock_lock,
patch.object(framework, "download_from_mirrors"),
patch.object(framework, "archive_extract_all") as mock_extract,
patch.object(framework, "_pio_system", return_value="linux_x86_64"),
):
mock_extract.side_effect = lambda *_a, **_kw: dest.mkdir(exist_ok=True)
framework._install_package("pkg", "1.0.0", dest, ["http://m"])
assert mock_lock.call_args.kwargs["fallback_to_soft"] is False
tests/unit_tests/test_platformio_registry.py
+310
View File
@@ -0,0 +1,310 @@
"""Tests for esphome.platformio.registry (PIO-registry package installs)."""
from __future__ import annotations
from contextlib import contextmanager
import json
import os
from pathlib import Path
from unittest.mock import patch
import pytest
from esphome.core import EsphomeError
from esphome.platformio import registry
@pytest.mark.parametrize(
("system", "machine", "expected"),
[
("Darwin", "arm64", "darwin_arm64"),
("Darwin", "x86_64", "darwin_x86_64"),
("Windows", "AMD64", "windows_amd64"),
# Deviation from upstream: auto-mapped to the emulated-x86 packages
("Windows", "ARM64", "windows_amd64"),
("Windows", "x86", "windows_x86"),
("Linux", "x86_64", "linux_x86_64"),
("Linux", "aarch64", "linux_aarch64"),
("Linux", "i686", "linux_i686"),
("Linux", "armv7l", "linux_armv7l"),
# Unknown hosts pass through like upstream; the registry lookup
# then fails naming the tag
("FreeBSD", "amd64", "freebsd_amd64"),
],
)
def test_get_systype(system: str, machine: str, expected: str) -> None:
with (
patch("platform.system", return_value=system),
patch("platform.machine", return_value=machine),
patch("platform.architecture", return_value=("64bit", "")),
):
assert registry.get_systype() == expected
def test_get_systype_env_override() -> None:
"""PLATFORMIO_SYSTEM_TYPE wins, exactly as in upstream get_systype()."""
with patch.dict(os.environ, {"PLATFORMIO_SYSTEM_TYPE": "windows_amd64"}):
assert registry.get_systype() == "windows_amd64"
def test_get_systype_aarch64_32bit_userland() -> None:
"""A 32-bit userland on a 64-bit arm kernel gets armv7l binaries."""
with (
patch("platform.system", return_value="Linux"),
patch("platform.machine", return_value="aarch64"),
patch("platform.architecture", return_value=("32bit", "")),
):
assert registry.get_systype() == "linux_armv7l"
def test_get_systype_windows_empty_machine() -> None:
"""An empty machine string falls back to the architecture bits."""
with (
patch("platform.system", return_value="Windows"),
patch("platform.machine", return_value=""),
patch("platform.architecture", return_value=("64bit", "")),
):
assert registry.get_systype() == "windows_amd64"
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(registry, "download_from_mirrors", side_effect=fake_download)
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.object(
registry,
"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"),
):
registry.registry_download("pkg", "1.0.0")
(mirrors, substitutions, _), _ = mock_download.call_args
assert mirrors == [registry._REGISTRY_URL]
assert substitutions == {"package": "pkg"}
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"
with (
patch.object(registry, "download_from_mirrors", side_effect=fake_download),
pytest.raises(EsphomeError, match="invalid JSON"),
):
registry.registry_download("pkg", "1.0.0")
def test_registry_download_matches_system() -> None:
with (
_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(registry, "get_systype", return_value="linux_x86_64"),
):
assert registry.registry_download("pkg", "1.0.0") == (
"http://x/linux",
"abc123",
42,
)
def test_registry_download_bare_string_system() -> None:
"""A bare-string system tag is an exact match, not a substring test."""
with (
_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(registry, "get_systype", return_value="linux_x86_64"),
):
assert registry.registry_download("pkg", "1.0.0")[0] == "http://x/x86_64"
def test_registry_download_wildcard_system() -> None:
with _registry_response(
[
{
"system": "*",
"download_url": "http://x/any",
"checksum": {"sha256": "abc"},
"size": 7,
}
]
):
assert registry.registry_download("pkg", "1.0.0") == (
"http://x/any",
"abc",
7,
)
def test_registry_download_missing_checksum_raises() -> None:
"""An unverifiable archive is refused, never silently extracted."""
with (
_registry_response([{"system": "*", "download_url": "http://x/any"}]),
pytest.raises(EsphomeError, match="no sha256"),
):
registry.registry_download("pkg", "1.0.0")
def test_registry_download_no_system_match() -> None:
with (
_registry_response(
[{"system": ["windows_amd64"], "download_url": "http://x/win"}]
),
patch.object(registry, "get_systype", return_value="linux_x86_64"),
pytest.raises(EsphomeError, match="No pkg 1.0.0 build"),
):
registry.registry_download("pkg", "1.0.0")
def test_registry_download_version_not_found() -> None:
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.object(registry, "download_from_mirrors", side_effect=fake_download),
pytest.raises(EsphomeError, match="not found"),
):
registry.registry_download("pkg", "1.0.0")
def test_install_package_skips_when_marker_exists(tmp_path: Path) -> None:
dest = tmp_path / "pkg"
dest.mkdir()
(dest / ".esphome_extracted").touch()
with patch.object(registry, "download_from_mirrors") as mock_download:
registry.install_package("pkg", "1.0.0", dest, [], tmp_path / "dl")
mock_download.assert_not_called()
def test_install_package_downloads_via_mirrors(tmp_path: Path) -> None:
dest = tmp_path / "pkg"
mirrors = ["http://mirror/{VERSION}/{SYSTEM}.tar.gz"]
with (
patch.object(registry, "download_from_mirrors") as mock_download,
patch.object(registry, "archive_extract_all") as mock_extract,
patch.object(registry, "get_systype", return_value="linux_x86_64"),
):
# Extraction is expected to create the directory
mock_extract.side_effect = lambda *_a, **_kw: dest.mkdir()
registry.install_package("pkg", "1.0.0", dest, mirrors, tmp_path / "dl")
assert mock_download.call_args[0][0] is mirrors
assert mock_download.call_args[0][1] == {
"VERSION": "1.0.0",
"SYSTEM": "linux_x86_64",
}
assert (dest / ".esphome_extracted").is_file()
def test_install_package_downloads_via_registry(tmp_path: Path) -> None:
"""The registry path downloads with the registry's sha256 and size."""
dest = tmp_path / "pkg"
with (
patch.object(registry, "download_with_resume") as mock_download,
patch.object(registry, "archive_extract_all") as mock_extract,
patch.object(
registry,
"registry_download",
return_value=("http://x/pkg.tar.gz", "abc123", 42),
),
):
mock_extract.side_effect = lambda *_a, **_kw: dest.mkdir()
registry.install_package("pkg", "1.0.0", dest, [], tmp_path / "dl")
assert mock_download.call_args[0][0] == "http://x/pkg.tar.gz"
assert mock_download.call_args[1] == {"sha256": "abc123", "size": 42}
def test_install_package_validates_expected_layout(tmp_path: Path) -> None:
"""The success marker is only written when the extracted tree is usable."""
dest = tmp_path / "pkg"
with (
patch.object(registry, "download_from_mirrors"),
patch.object(registry, "archive_extract_all") as mock_extract,
patch.object(registry, "get_systype", return_value="linux_x86_64"),
):
mock_extract.side_effect = lambda *_a, **_kw: (dest / "bin").mkdir(parents=True)
registry.install_package(
"pkg", "1.0.0", dest, ["http://m"], tmp_path / "dl", expect=("bin",)
)
assert (dest / ".esphome_extracted").is_file()
def test_install_package_unexpected_layout_raises(tmp_path: Path) -> None:
dest = tmp_path / "pkg"
with (
patch.object(registry, "download_from_mirrors"),
patch.object(registry, "archive_extract_all") as mock_extract,
patch.object(registry, "get_systype", return_value="linux_x86_64"),
pytest.raises(EsphomeError, match="without the expected bin"),
):
mock_extract.side_effect = lambda *_a, **_kw: dest.mkdir()
registry.install_package(
"pkg", "1.0.0", dest, ["http://m"], tmp_path / "dl", expect=("bin",)
)
assert not (dest / ".esphome_extracted").exists()
def test_install_package_marker_rechecked_under_lock(tmp_path: Path) -> None:
"""A concurrent install finishing while we wait for the lock is detected."""
dest = tmp_path / "pkg"
marker = dest / ".esphome_extracted"
@contextmanager
def _fake_lock(*_a, **_kw):
dest.mkdir(parents=True, exist_ok=True)
marker.touch()
yield
with (
patch("filelock.FileLock", _fake_lock),
patch.object(registry, "download_from_mirrors") as mock_download,
patch.object(registry, "rmdir") as mock_rmdir,
):
registry.install_package("pkg", "1.0.0", dest, ["http://m"], tmp_path / "dl")
mock_download.assert_not_called()
mock_rmdir.assert_not_called()
def test_install_package_uses_hard_lock(tmp_path: Path) -> None:
"""The install lock must never degrade to a soft (existence) lock."""
dest = tmp_path / "pkg"
with (
patch("filelock.FileLock") as mock_lock,
patch.object(registry, "download_from_mirrors"),
patch.object(registry, "archive_extract_all") as mock_extract,
patch.object(registry, "get_systype", return_value="linux_x86_64"),
):
mock_extract.side_effect = lambda *_a, **_kw: dest.mkdir(exist_ok=True)
registry.install_package("pkg", "1.0.0", dest, ["http://m"], tmp_path / "dl")
assert mock_lock.call_args.kwargs["fallback_to_soft"] is False