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

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

Browse Source
This commit is contained in:
J. Nick Koston
2026-08-20 15:29:04 -05:00
parent 8043767460 ea429ffdb6
commit d736b0d1eb
15 changed files with 618 additions and 519 deletions
Download Patch File Download Diff File Expand all files Collapse all files
esphome/arduino/__init__.py
View File
esphome/arduino8266/component.py → esphome/arduino/library.py
+18 -13
View File
@@ -1,10 +1,12 @@
"""Arduino ESP8266 backend for the shared PlatformIO library converter.
"""Arduino-core backend for the shared PlatformIO library converter.
Turns the libraries registered via ``cg.add_library()`` into build inputs for
the ninja generator. Bare names that exist under the framework's bundled
a native Arduino build. Bare names that exist under the framework's bundled
``libraries/`` directory (ESP8266WiFi, Wire, SPI, ...) are read straight from
the framework tree; everything else goes through the shared
resolution/download pipeline in ``esphome.platformio.library``.
resolution/download pipeline in ``esphome.platformio.library``. Nothing here
is core-specific: the caller names the PlatformIO platform, MCU, and cache
key of the Arduino core it builds.
Mirrors PlatformIO's ``lib_ldf_mode=off`` behavior: each library builds into
its own static archive and every library's include dir joins one global
@@ -40,8 +42,6 @@ from esphome.platformio.library import (
_LOGGER = logging.getLogger(__name__)
ESP8266_PLATFORM = "espressif8266"
@dataclass
class ArduinoLibrary:
@@ -145,8 +145,15 @@ def _bundled_library(framework_path: Path, name: str) -> ArduinoLibrary:
return _library_info(name, lib_dir, {"name": name, **data})
def resolve_libraries(framework_path: Path) -> list[ArduinoLibrary]:
"""Resolve every ``cg.add_library()`` entry into an :class:`ArduinoLibrary`."""
def resolve_libraries(
framework_path: Path, *, pio_platform: str, board_mcu: str, cache_key: str
) -> list[ArduinoLibrary]:
"""Resolve every ``cg.add_library()`` entry into an :class:`ArduinoLibrary`.
``pio_platform``/``board_mcu`` filter manifests the way PlatformIO would
for that core (e.g. ``espressif8266``/``esp8266``); ``cache_key`` keys the
shared converter's download cache.
"""
bundled: list[ArduinoLibrary] = []
external: list[Library] = []
# PlatformIO's lib_ignore covers framework-bundled libraries too; the
@@ -200,7 +207,7 @@ def resolve_libraries(framework_path: Path) -> list[ArduinoLibrary]:
)
continue
try:
check_library_data(dep, ESP8266_PLATFORM, "arduino")
check_library_data(dep, pio_platform, "arduino")
except InvalidLibrary as err:
# check_library_data's only raise is the platform filter, and
# rejecting another platform's dependency of a cross-platform
@@ -212,9 +219,7 @@ def resolve_libraries(framework_path: Path) -> list[ArduinoLibrary]:
bundled.append(_bundled_library(framework_path, name))
def _emit(component: ConvertedLibrary) -> None:
apply_extra_script(
component, board_mcu="esp8266", pio_platform=ESP8266_PLATFORM
)
apply_extra_script(component, board_mcu=board_mcu, pio_platform=pio_platform)
converted.append(
_library_info(
component.get_require_name(), component.source_dir, component.data
@@ -226,10 +231,10 @@ def resolve_libraries(framework_path: Path) -> list[ArduinoLibrary]:
convert_libraries(
external,
LibraryBackend(
platform=ESP8266_PLATFORM,
platform=pio_platform,
framework="arduino",
emit=_emit,
cache_key="arduino8266",
cache_key=cache_key,
),
)
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/build_gen/arduino8266.py
+30 -32
View File
@@ -39,25 +39,19 @@ from esphome.const import KEY_CORE, KEY_FRAMEWORK_VERSION
from esphome.core import CORE, EsphomeError
from esphome.framework_helpers import get_project_cxx_compile_flags
from esphome.helpers import mkdir_p, write_file_if_changed
from esphome.platformio.library import join_flag_args, split_flag_entry
from esphome.platformio.library import (
SOURCE_KIND_FOR_SUFFIX,
join_flag_args,
split_flag_entry,
)
_LOGGER = logging.getLogger(__name__)
# Compile rule per source suffix; keys must cover SRC_FILE_EXTENSIONS so any
# source a library manifest selects has a rule (pinned by a drift test).
# Compile rule per source suffix, derived from the shared suffix -> kind map
# so every source extension a library manifest can select has a rule.
_RULE_FOR_KIND = {"c": "cc", "cxx": "cxx", "asm": "asm"}
_RULE_FOR_SUFFIX = {
".c": "cc",
".cpp": "cxx",
".cc": "cxx",
".cxx": "cxx",
".c++": "cxx",
".S": "asm",
".spp": "asm",
".SPP": "asm",
".sx": "asm",
".s": "asm",
".asm": "asm",
".ASM": "asm",
suffix: _RULE_FOR_KIND[kind] for suffix, kind in SOURCE_KIND_FOR_SUFFIX.items()
}
# Always excluded from the core build: ESPHome uses its own native OTA
@@ -309,16 +303,11 @@ def _quote_arg(tok: str) -> str:
return f'"{quoted}"'
def _q(value) -> str:
"""Force-quote a path for the ninja command line (shell/CreateProcess)."""
return _quote_arg(str(value).replace("$", "$$"))
_NEEDS_QUOTE = re.compile(r'[\s"\']')
def _shell_token(tok: str) -> str:
"""Quote a lexed token only when needed; ``_q`` force-quotes paths.
def _shell_token(tok: str, force: bool = False) -> str:
"""Quote a lexed token only when needed; ``force`` always quotes.
Lexing strips the quoting a user wrote (``-DX="a b"`` becomes the single
token ``-DX=a b``); re-quote on the way out so the compiler receives the
@@ -328,9 +317,14 @@ def _shell_token(tok: str) -> str:
PlatformIO parity.
"""
tok = tok.replace("$", "$$") # ninja would expand a bare $ to nothing
if not _NEEDS_QUOTE.search(tok):
return tok
return _quote_arg(tok)
if force or _NEEDS_QUOTE.search(tok):
return _quote_arg(tok)
return tok
def _q(value) -> str:
"""Force-quote a path for the ninja command line (shell/CreateProcess)."""
return _shell_token(str(value), force=True)
def _defines_flags(
@@ -505,7 +499,7 @@ def write_project(paths: dict[str, Path]) -> bool:
Returns True when ``build.ninja`` changed, so the caller can skip work
derived purely from it (the compile database) on unchanged builds.
"""
from esphome.arduino8266.component import resolve_libraries
from esphome.arduino.library import resolve_libraries
from esphome.arduino8266.framework import ccache_path
framework = paths["framework_path"]
@@ -528,7 +522,12 @@ def write_project(paths: dict[str, Path]) -> bool:
variant_dir = framework / "variants" / board_build["variant"]
src_dir = CORE.relative_src_path()
libraries = resolve_libraries(framework)
libraries = resolve_libraries(
framework,
pio_platform="espressif8266",
board_mcu="esp8266",
cache_key="arduino8266",
)
# A missing install directory would otherwise surface as a wall of
# include errors; failing here names the path instead.
@@ -574,11 +573,10 @@ def write_project(paths: dict[str, Path]) -> bool:
# build_unflags applies to the framework flag sets too (compile and link),
# as under PlatformIO (a silently ignored ``build_unflags: -Os`` would
# diverge between the toolchains).
cflags = [f for f in cflags if f not in unflags]
cxxflags = [f for f in cxxflags if f not in unflags]
asflags = [f for f in asflags if f not in unflags]
link_flags = [f for f in _LINKFLAGS if f not in unflags]
cflags, cxxflags, asflags, link_flags = (
[f for f in flags if f not in unflags]
for flags in (cflags, cxxflags, asflags, _LINKFLAGS)
)
if esp8266_data[KEY_SCANF_FLOAT]:
link_flags += ["-u", "_scanf_float"]
link_flags += project_link_flags
esphome/build_gen/build_tool.py
+3 -1
View File
@@ -25,7 +25,9 @@ def main() -> int:
# treats backslashes in response files as escapes, corrupting Windows
# paths ("sub\a.o" -> "suba.o").
objects = Path(rspfile).read_text(encoding="utf-8").split()
return subprocess.run([ar, "rc", archive, *objects], check=False).returncode
return subprocess.run(
[ar, "rc", archive, *objects], check=False, close_fds=False
).returncode
if mode == "copy":
src, dst = sys.argv[2:4]
shutil.copyfile(src, dst)
esphome/build_helpers/size_summary.py
+9
View File
@@ -17,3 +17,12 @@ def format_bar(used: int, total: int) -> str:
f"[{progress:<{blocks}}] {pct_raw: 6.1%} "
f"(used {used:d} bytes from {total:d} bytes)"
)
def print_size_line(label: str, used: int, total: int) -> None:
"""One PlatformIO-format summary line (``RAM``/``Flash``).
The label padding is part of the format: ``script/ci_memory_impact_extract.py``
matches these lines verbatim.
"""
print(f"{label + ':':<7}{format_bar(used, total)}")
esphome/espidf/size_summary.py
+3 -3
View File
@@ -28,7 +28,7 @@ import json
import logging
from pathlib import Path
from esphome.build_helpers.size_summary import format_bar
from esphome.build_helpers.size_summary import print_size_line
_LOGGER = logging.getLogger(__name__)
_SIZE_SUFFIXES = {"K": 1024, "M": 1024 * 1024}
@@ -89,7 +89,7 @@ def print_summary(size_json: Path, partitions_csv: Path | None) -> None:
ram_used = ram_region.get("used")
ram_total = ram_region.get("size")
if ram_total and ram_used is not None:
print(f"RAM: {format_bar(ram_used, ram_total)}")
print_size_line("RAM", ram_used, ram_total)
image_size = data.get("image_size")
if image_size is None or partitions_csv is None:
@@ -99,4 +99,4 @@ def print_summary(size_json: Path, partitions_csv: Path | None) -> None:
except ValueError as e:
_LOGGER.debug("Skipping Flash summary: %s", e)
return
print(f"Flash: {format_bar(image_size, app_size)}")
print_size_line("Flash", image_size, app_size)
esphome/framework_helpers.py
+1
View File
@@ -1238,6 +1238,7 @@ def _ccache_runs(ccache: str) -> bool:
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
timeout=15,
close_fds=False,
)
except (OSError, subprocess.SubprocessError):
_LOGGER.warning(
esphome/platformio/library.py
+17 -14
View File
@@ -48,20 +48,23 @@ DEFAULT_BUILD_SRC_FILTER = (
DEFAULT_BUILD_SRC_DIRS = "src"
DEFAULT_BUILD_INCLUDE_DIR = "include"
DEFAULT_BUILD_FLAGS = []
SRC_FILE_EXTENSIONS = [
".c",
".cpp",
".cc",
".cxx",
".c++",
".S",
".spp",
".SPP",
".sx",
".s",
".asm",
".ASM",
]
# Source suffix -> compiler kind, PlatformIO's CSUFFIXES/CXXSUFFIXES/ASSUFFIXES
# split. Native build generators map the kind to their compile rules.
SOURCE_KIND_FOR_SUFFIX: dict[str, str] = {
".c": "c",
".cpp": "cxx",
".cc": "cxx",
".cxx": "cxx",
".c++": "cxx",
".S": "asm",
".spp": "asm",
".SPP": "asm",
".sx": "asm",
".s": "asm",
".asm": "asm",
".ASM": "asm",
}
SRC_FILE_EXTENSIONS = list(SOURCE_KIND_FOR_SUFFIX)
DOMAIN = "pio_components"
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/build_gen/test_arduino8266.py
+2 -2
View File
@@ -201,7 +201,7 @@ def _write_ninja(
with (
patch.object(arduino8266, "generate_ld_scripts"),
patch(
"esphome.arduino8266.component.resolve_libraries",
"esphome.arduino.library.resolve_libraries",
return_value=libraries or [],
),
patch("esphome.arduino8266.framework.ccache_path", return_value=ccache),
@@ -460,7 +460,7 @@ def test_generate_ld_scripts_testing_mode(tmp_path: Path) -> None:
def test_write_project_libraries_and_variant(
tmp_path: Path, caplog: pytest.LogCaptureFixture
) -> None:
from esphome.arduino8266.component import ArduinoLibrary
from esphome.arduino.library import ArduinoLibrary
paths = _make_framework(tmp_path)
variant_src = paths["framework_path"] / "variants" / "nodemcu" / "variant.cpp"
tests/unit_tests/build_helpers/test_size_summary.py
+13 -1
View File
@@ -2,9 +2,21 @@
from __future__ import annotations
from esphome.build_helpers.size_summary import format_bar
import pytest
from esphome.build_helpers.size_summary import format_bar, print_size_line
def test_format_bar_zero_total() -> None:
"""A zero total must not divide by zero."""
assert format_bar(0, 0) == "[ ] 0.0% (used 0 bytes from 0 bytes)"
def test_print_size_line_label_padding(capsys: pytest.CaptureFixture[str]) -> None:
"""The label column is exactly what ci_memory_impact_extract.py greps."""
print_size_line("RAM", 47932, 180736)
print_size_line("Flash", 888511, 1835008)
out = capsys.readouterr().out.splitlines()
assert out[0].startswith("RAM: [")
assert out[1].startswith("Flash: [")
assert "26.5% (used 47932 bytes from 180736 bytes)" in out[0]
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_arduino8266_component.py → tests/unit_tests/test_arduino_library.py
+44 -9
View File
@@ -1,4 +1,4 @@
"""Tests for esphome.arduino8266.component (library resolution)."""
"""Tests for esphome.arduino.library (Arduino-core library resolution)."""
from __future__ import annotations
@@ -8,7 +8,7 @@ from unittest.mock import patch
import pytest
from esphome.arduino8266 import component
from esphome.arduino import library as component
from esphome.const import KEY_CORE, KEY_TARGET_PLATFORM, PLATFORM_ESP8266
from esphome.core import CORE, EsphomeError, Library
from esphome.platformio.library import ConvertedLibrary, LibraryBackend
@@ -149,7 +149,12 @@ def test_library_info_no_src_dir(tmp_path: Path) -> None:
def test_resolve_libraries_bundled(tmp_path: Path) -> None:
framework = _make_framework(tmp_path)
_add_library("ESP8266WiFi", None)
libs = component.resolve_libraries(framework)
libs = component.resolve_libraries(
framework,
pio_platform="espressif8266",
board_mcu="esp8266",
cache_key="arduino8266",
)
assert [lib.name for lib in libs] == ["ESP8266WiFi"]
@@ -159,7 +164,12 @@ def test_resolve_libraries_bare_registry_name_is_external(tmp_path: Path) -> Non
framework = _make_framework(tmp_path)
_add_library("pngle", None)
with patch.object(component, "convert_libraries", return_value=[]) as mock_convert:
component.resolve_libraries(framework)
component.resolve_libraries(
framework,
pio_platform="espressif8266",
board_mcu="esp8266",
cache_key="arduino8266",
)
(libraries, _backend), _ = mock_convert.call_args
assert [lib.name for lib in libraries] == ["pngle"]
@@ -197,7 +207,12 @@ def test_resolve_libraries_external_and_bundled_deps(tmp_path: Path) -> None:
)
with _emitting_converter(converted) as mock_extra:
libs = component.resolve_libraries(framework)
libs = component.resolve_libraries(
framework,
pio_platform="espressif8266",
board_mcu="esp8266",
cache_key="arduino8266",
)
mock_extra.assert_called_once_with(
converted, board_mcu="esp8266", pio_platform="espressif8266"
@@ -220,7 +235,12 @@ def test_resolve_libraries_bundled_dep_already_present(tmp_path: Path) -> None:
)
with _emitting_converter(converted):
libs = component.resolve_libraries(framework)
libs = component.resolve_libraries(
framework,
pio_platform="espressif8266",
board_mcu="esp8266",
cache_key="arduino8266",
)
# Wire appears once (from the explicit registration), not twice
assert [lib.name for lib in libs] == ["Wire", "some__External"]
@@ -233,7 +253,12 @@ def test_resolve_libraries_versioned_bare_name_is_external(tmp_path: Path) -> No
_add_library("pngle", "1.1.0")
with patch.object(component, "convert_libraries", return_value=[]) as mock_convert:
component.resolve_libraries(framework)
component.resolve_libraries(
framework,
pio_platform="espressif8266",
board_mcu="esp8266",
cache_key="arduino8266",
)
(libraries, _backend), _ = mock_convert.call_args
assert [lib.name for lib in libraries] == ["pngle"]
@@ -283,7 +308,12 @@ def test_resolve_libraries_lib_ignore_covers_bundled(tmp_path: Path) -> None:
_add_library("ESP8266WiFi", None)
_add_library("Wire", None)
CORE.platformio_options = {"lib_ignore": ["Wire"]}
libs = component.resolve_libraries(framework)
libs = component.resolve_libraries(
framework,
pio_platform="espressif8266",
board_mcu="esp8266",
cache_key="arduino8266",
)
assert [lib.name for lib in libs] == ["ESP8266WiFi"]
@@ -301,7 +331,12 @@ def test_resolve_libraries_lib_ignore_covers_bundled_dependencies(
)
with _emitting_converter(converted):
libs = component.resolve_libraries(framework)
libs = component.resolve_libraries(
framework,
pio_platform="espressif8266",
board_mcu="esp8266",
cache_key="arduino8266",
)
assert [lib.name for lib in libs] == ["some__External"]
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