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

Merge remote-tracking branch 'origin/esp8266-arduino-toolchain' into esp8266-native-pch

Browse Source
This commit is contained in:
J. Nick Koston
2026-08-25 13:16:51 -05:00
parent 2665a4fa12 d77be85afa
commit aaf76710a9
25 changed files with 1242 additions and 536 deletions
Download Patch File Download Diff File Expand all files Collapse all files
esphome/build_gen/espidf.py
+38 -33
View File
@@ -1,6 +1,7 @@
"""ESP-IDF direct build generator for ESPHome."""
import json
import logging
from pathlib import Path
from esphome.components.esp32 import (
@@ -11,6 +12,7 @@ from esphome.components.esp32 import (
)
import esphome.config_validation as cv
from esphome.core import CORE
from esphome.espidf import variant_to_idf_target
from esphome.framework_helpers import (
get_project_compile_flags,
get_project_cxx_compile_flags,
@@ -18,6 +20,8 @@ from esphome.framework_helpers import (
)
from esphome.helpers import mkdir_p, write_file_if_changed
_LOGGER = logging.getLogger(__name__)
# Replaces the IDF default C++ standard (-std=gnu++2b appended to
# CXX_COMPILE_OPTIONS by project.cmake's __build_init) with the one set via
# cg.set_cpp_standard(). Emitted between include(project.cmake) and project(),
@@ -31,11 +35,12 @@ idf_build_set_property(CXX_COMPILE_OPTIONS "${{esphome_cxx_compile_options}}")""
def get_available_components() -> list[str] | None:
"""Get list of built-in ESP-IDF components from project_description.json.
"""List the built-in ESP-IDF components from ``project_description.json``.
Excludes ``src``, IDF-managed components (``managed_components/``), and
converted PIO libs (``pio_components/``). Returns ``None`` if the build
dir or ``project_description.json`` isn't ready yet.
Only components below its ``idf_path/components`` count, which leaves out
``src``, IDF-managed components, converted PIO libs and project local
ones such as the Arduino ``component_stubs``. Returns ``None`` if the
build dir or ``project_description.json`` isn't ready yet.
"""
if CORE.build_path is None:
return None
@@ -46,30 +51,24 @@ def get_available_components() -> list[str] | None:
try:
with project_desc.open(encoding="utf-8") as f:
data = json.load(f)
component_info = data.get("build_component_info", {})
result = []
for name, info in component_info.items():
# Exclude our own src component
if name == "src":
continue
# Exclude IDF-managed and converted-PIO components (external).
comp_dir = info.get("dir", "")
if "managed_components" in comp_dir or "pio_components" in comp_dir:
continue
result.append(name)
return result
except (json.JSONDecodeError, OSError):
root = (Path(data["idf_path"]) / "components").resolve()
result = [
name
for name, info in data.get("build_component_info", {}).items()
if (comp_dir := info.get("dir"))
and Path(comp_dir).resolve().is_relative_to(root)
]
except (json.JSONDecodeError, KeyError, OSError) as err:
_LOGGER.debug("Could not read %s: %s", project_desc, err)
return None
if not result:
_LOGGER.warning("No ESP-IDF components found under %s", root)
return result
def has_discovered_components() -> bool:
"""Check if we have discovered components from a previous configure."""
return get_available_components() is not None
"""Check if a previous configure discovered any built-in components."""
return bool(get_available_components())
def _cmake_quote(value: str) -> str:
@@ -79,15 +78,17 @@ def _cmake_quote(value: str) -> str:
return f'"{escaped}"'
def get_project_cmakelists(minimal: bool = False) -> str:
def get_project_cmakelists(
minimal: bool = False, builtin_components: list[str] | None = None
) -> str:
"""Generate the top-level CMakeLists.txt for ESP-IDF project.
When ``minimal`` is true, omit ``ESPHOME_PROJECT_BUILTIN_COMPONENTS``
since ``project_description.json`` may be stale on the first write.
``builtin_components`` supplies the discovered list (from the cache)
instead of reading it from ``project_description.json``.
"""
# Get IDF target from ESP32 variant (e.g., ESP32S3 -> esp32s3)
variant = get_esp32_variant()
idf_target = variant.lower().replace("-", "")
idf_target = variant_to_idf_target(get_esp32_variant())
# esp_idf_size 2.x (bundled with IDF >=6.0) made NG the default and
# removed the --ng flag; on 1.x (IDF 5.5) --ng is required to get
@@ -162,9 +163,11 @@ def get_project_cmakelists(minimal: bool = False) -> str:
else "\n".join(
f"idf_build_set_property(ESPHOME_PROJECT_BUILTIN_COMPONENTS {name} APPEND)"
for name in sorted(
set(get_available_components() or []).difference(
CORE.cmake_args.get("EXCLUDE_COMPONENTS", "").split(";")
)
set(
builtin_components
if builtin_components is not None
else get_available_components() or []
).difference(CORE.cmake_args.get("EXCLUDE_COMPONENTS", "").split(";"))
)
)
)
@@ -279,7 +282,9 @@ target_link_options(${{COMPONENT_LIB}} PUBLIC
"""
def write_project(minimal: bool = False) -> None:
def write_project(
minimal: bool = False, builtin_components: list[str] | None = None
) -> None:
"""Write ESP-IDF project files."""
mkdir_p(CORE.build_path)
mkdir_p(CORE.relative_src_path())
@@ -287,7 +292,7 @@ def write_project(minimal: bool = False) -> None:
# Write top-level CMakeLists.txt
write_file_if_changed(
CORE.relative_build_path("CMakeLists.txt"),
get_project_cmakelists(minimal=minimal),
get_project_cmakelists(minimal=minimal, builtin_components=builtin_components),
)
# Write component CMakeLists.txt in src/
esphome/components/dashboard_import/__init__.py
+12 -7
View File
@@ -4,7 +4,6 @@ import re
import secrets
from typing import Any
import requests
from ruamel.yaml import YAML
from esphome import git
@@ -13,7 +12,7 @@ from esphome.components.packages import validate_source_shorthand
import esphome.config_validation as cv
from esphome.const import CONF_ESPHOME, CONF_PROJECT, CONF_REF, CONF_WIFI
import esphome.final_validate as fv
from esphome.happy_eyeballs import ensure_happy_eyeballs
from esphome.net_retry import fetch_with_retry, http_request
from esphome.types import ConfigType
from esphome.yaml_util import dump
@@ -111,14 +110,20 @@ def import_config(
if git_file.query and "full_config" in git_file.query:
url = git_file.raw_url
try:
ensure_happy_eyeballs()
req = requests.get(url, timeout=30)
# Deferred so config-time imports of this component stay light;
# http_request does the lazy import for the request itself.
import requests
def _fetch() -> str:
req = http_request("GET", url, timeout=30)
req.raise_for_status()
return req.text
try:
contents = fetch_with_retry(url, _fetch, what="Import")
except requests.exceptions.RequestException as e:
raise ValueError(f"Error while fetching {url}: {e}") from e
contents = req.text
yaml = YAML()
loaded_yaml = yaml.load(contents)
if (
esphome/components/emontx/sensor/__init__.py
+83 -15
View File
@@ -7,8 +7,10 @@ from esphome.const import (
CONF_ID,
CONF_STATE_CLASS,
CONF_UNIT_OF_MEASUREMENT,
DEVICE_CLASS_APPARENT_POWER,
DEVICE_CLASS_CURRENT,
DEVICE_CLASS_ENERGY,
DEVICE_CLASS_FREQUENCY,
DEVICE_CLASS_POWER,
DEVICE_CLASS_POWER_FACTOR,
DEVICE_CLASS_TEMPERATURE,
@@ -18,8 +20,10 @@ from esphome.const import (
UNIT_AMPERE,
UNIT_CELSIUS,
UNIT_EMPTY,
UNIT_HERTZ,
UNIT_PULSES,
UNIT_VOLT,
UNIT_VOLT_AMPS,
UNIT_WATT,
UNIT_WATT_HOURS,
)
@@ -29,6 +33,32 @@ from .. import CONF_EMONTX_ID, CONF_TAG_NAME, EmonTx, emontx_ns
EmonTxSensor = emontx_ns.class_("EmonTxSensor", sensor.Sensor, cg.Component)
# Known emonTx/avrdb JSON tag conventions, gathered from real firmware
# (see https://github.com/openenergymonitor/avrdb_firmware), used to decide
# whether each tag below requires a numeric index or may also appear bare:
#
# Tag family Bare (no index) Numeric-indexed
# ----------- ----------------------- ----------------------------------
# P (power) no P1, P2, ... (multi-channel boards)
# E (energy) no E1, E2, ...
# V (voltage) Vrms (NOT matched here, V1, V2, V3 (per-phase boards)
# doesn't fit "V"+digits)
# I (current) no I1, I2, ...
# T (temp.) no T1, T2, ...
# F (frequency) F (single mains freq.) not seen indexed
# PULSE pulse (single-CT boards) PULSE1, PULSE2, ... (other variants)
# PF (power not seen bare PF1, PF2, ... (currently unused/
# factor) commented out in avrdb firmware)
# AP (apparent not seen bare AP1, AP2, ... (not an avrdb tag at
# power) all; avrdb uses "VA"+index instead,
# itself currently unused/commented
# out; "AP" is kept here for other
# firmware/integrations using it)
#
# This is why a bare "PULSE" resolves to proper defaults below, but bare
# "PF"/"AP" fall back to generic defaults instead: only PULSE has a
# confirmed bare-tag use in real, currently-shipping firmware.
# Define sensor type configurations by prefix
SENSOR_CONFIGS = {
"P": {
@@ -63,7 +93,25 @@ SENSOR_CONFIGS = {
},
}
# Pattern-based configurations
# Tags reported once, without a numeric index (e.g. "F"), matched exactly
# rather than by prefix.
EXACT_TAG_CONFIGS = {
"F": {
CONF_UNIT_OF_MEASUREMENT: UNIT_HERTZ,
CONF_DEVICE_CLASS: DEVICE_CLASS_FREQUENCY,
CONF_STATE_CLASS: STATE_CLASS_MEASUREMENT,
CONF_ACCURACY_DECIMALS: 2,
},
}
# Pattern-based configurations. The remainder after the prefix must be a
# non-empty numeric index (like V1/I1/E1), so e.g. "APPLE" doesn't collide
# with the "AP" prefix and a bare "PF"/"AP" (no index) doesn't match.
# "PULSE" is the exception: some emonTx firmware (e.g. avrdb-based single-CT
# variants) reports a single pulse counter as a bare "pulse" tag with no
# numeric index at all, so that pattern also accepts an empty suffix.
PATTERNS_ALLOWING_BARE_TAG = {"PULSE"}
PATTERN_CONFIGS = {
"PULSE": {
CONF_UNIT_OF_MEASUREMENT: UNIT_PULSES,
@@ -77,14 +125,21 @@ PATTERN_CONFIGS = {
CONF_STATE_CLASS: STATE_CLASS_MEASUREMENT,
CONF_ACCURACY_DECIMALS: 2,
},
"AP": {
CONF_UNIT_OF_MEASUREMENT: UNIT_VOLT_AMPS,
CONF_DEVICE_CLASS: DEVICE_CLASS_APPARENT_POWER,
CONF_STATE_CLASS: STATE_CLASS_MEASUREMENT,
CONF_ACCURACY_DECIMALS: 2,
},
}
# BASE_SCHEMA intentionally omits state_class and accuracy_decimals defaults.
# Passing them to sensor_schema() would register them via cv.Optional(key, default=...),
# making them always present in the validated config dict and preventing
# apply_tag_defaults from overriding them with the correct per-prefix values.
# They are injected by apply_tag_defaults below, after running through
# sensor.validate_state_class() so the value is code-generation-ready.
# They are injected by apply_tag_defaults below, after running through the
# same validators sensor_schema() would use (see _DEFAULT_VALIDATORS) so the
# values are code-generation-ready.
BASE_SCHEMA = sensor.sensor_schema(EmonTxSensor).extend(
{
cv.GenerateID(CONF_EMONTX_ID): cv.use_id(EmonTx),
@@ -93,30 +148,43 @@ BASE_SCHEMA = sensor.sensor_schema(EmonTxSensor).extend(
)
_DEFAULT_VALIDATORS = {
CONF_STATE_CLASS: sensor.validate_state_class,
CONF_DEVICE_CLASS: sensor.validate_device_class,
CONF_UNIT_OF_MEASUREMENT: sensor.validate_unit_of_measurement,
}
def _apply_defaults(config: ConfigType, defaults: dict) -> None:
"""Inject defaults into config, skipping keys already set by the user.
state_class values are run through validate_state_class so they are
code-generation-ready, matching what sensor_schema() would normally do."""
Values are run through the same validators sensor_schema() would use, so
they are code-generation-ready and a typo'd constant fails validation
instead of shipping silently."""
for key, value in defaults.items():
if key not in config:
if key == CONF_STATE_CLASS:
value = sensor.validate_state_class(value)
if key in _DEFAULT_VALIDATORS:
value = _DEFAULT_VALIDATORS[key](value)
config[key] = value
def apply_tag_defaults(config: ConfigType) -> ConfigType:
"""Apply defaults based on tag prefix if applicable, but don't restrict any tags."""
tag = config[CONF_TAG_NAME]
tag_upper = tag.upper()
if (exact_config := EXACT_TAG_CONFIGS.get(tag_upper)) is not None:
_apply_defaults(config, exact_config)
return config
for pattern, pattern_config in PATTERN_CONFIGS.items():
suffix = tag_upper[len(pattern) :]
bare_ok = not suffix and pattern in PATTERNS_ALLOWING_BARE_TAG
if tag_upper.startswith(pattern) and (suffix.isdigit() or bare_ok):
_apply_defaults(config, pattern_config)
return config
# Only apply defaults for known prefixes with numeric indices (e.g. E1, V2, T3)
if len(tag) >= 2:
tag_upper = tag.upper()
for pattern, pattern_config in PATTERN_CONFIGS.items():
if tag_upper.startswith(pattern):
_apply_defaults(config, pattern_config)
return config
# Only apply defaults for known prefixes with numeric indices (e.g. E1, V2, T3)
prefix = tag_upper[0]
if prefix in SENSOR_CONFIGS and tag[1:].isdigit():
_apply_defaults(config, SENSOR_CONFIGS[prefix])
esphome/components/esp8266/__init__.py
+9 -5
View File
@@ -661,8 +661,10 @@ def _decode_pc(config: ConfigType, addr: str, *, bulk: bool = False) -> None:
_warn_decode_problem(
str(path), "Cannot decode crash addresses: %s missing", path
)
# The detailed warning names no address, so mark each one
_LOGGER.warning("Not decoded %s (toolchain file missing)", addr)
# The detailed warning names no address; mark named
# registers, but bulk stack words at debug (~150 per dump)
log = _LOGGER.debug if bulk else _LOGGER.warning
log("Not decoded %s (toolchain file missing)", addr)
return
addr2line, elf = str(addr2line), str(elf)
else:
@@ -674,7 +676,8 @@ def _decode_pc(config: ConfigType, addr: str, *, bulk: bool = False) -> None:
"no-addr2line",
"Cannot decode crash addresses: no addr2line or ELF in idedata",
)
_LOGGER.warning("Not decoded %s (no addr2line or ELF)", addr)
log = _LOGGER.debug if bulk else _LOGGER.warning
log("Not decoded %s (no addr2line or ELF)", addr)
return
addr2line, elf = idedata.addr2line_path, idedata.firmware_elf_path
command = [addr2line, "-pfiaC", "-e", elf, addr]
@@ -688,8 +691,9 @@ def _decode_pc(config: ConfigType, addr: str, *, bulk: bool = False) -> None:
"addr2line-failed", "Could not decode crash address %s (%s)", addr, err
):
# The detailed warning already named this address; mark only
# the addresses whose warning was rate-limited away
_LOGGER.warning("Not decoded %s (addr2line failed)", addr)
# the rate-limited ones, and bulk stack words at debug
log = _LOGGER.debug if bulk else _LOGGER.warning
log("Not decoded %s (addr2line failed)", addr)
_LOGGER.debug("Caught exception for command %s", command, exc_info=1)
return
esphome/components/nrf52/framework.py
+30 -29
View File
@@ -5,16 +5,14 @@ from pathlib import Path
import platform
import shutil
import sys
import tempfile
from esphome.build_helpers.tools_cache import SDK_NRF_TOOLS_CACHE, tools_cache_path
import esphome.config_validation as cv
from esphome.const import KEY_CORE, KEY_FRAMEWORK_VERSION
from esphome.core import CORE, EsphomeError
from esphome.framework_helpers import (
archive_extract_all,
create_venv,
download_from_mirrors,
download_and_extract,
get_python_env_executable_path,
rmdir,
run_command_ok,
@@ -338,34 +336,37 @@ def check_and_install() -> None:
if not sentinel.exists():
rmdir(toolchains_dir, msg=f"Clean up {TOOLCHAIN_VERSION} toolchain environment")
sysname, machine, extension = _get_toolchain_platform_info()
with tempfile.NamedTemporaryFile() as tmp:
_LOGGER.info("Downloading Zephyr SDK %s minimal ...", TOOLCHAIN_VERSION)
download_from_mirrors(
SDK_NG_MINIMAL_MIRRORS,
{
"VERSION": TOOLCHAIN_VERSION,
"sysname": sysname,
"machine": machine,
"extension": extension,
},
tmp.file,
)
archive_extract_all(tmp.file, toolchains_dir, progress_header="Extracting")
with tempfile.NamedTemporaryFile() as tmp:
_LOGGER.info("Downloading %s toolchain ...", TOOLCHAIN_VERSION)
download_from_mirrors(
substitutions = {
"VERSION": TOOLCHAIN_VERSION,
"sysname": sysname,
"machine": machine,
"extension": extension,
}
# Downloaded next to the destination (not a temp file) so an
# interrupted download's .part file resumes on the next run.
for mirrors, extract_dir, what, slug in (
(SDK_NG_MINIMAL_MIRRORS, toolchains_dir, "Zephyr SDK minimal", "minimal"),
(
SDK_NG_TOOLCHAIN_MIRRORS,
{
"VERSION": TOOLCHAIN_VERSION,
"sysname": sysname,
"machine": machine,
"extension": extension,
},
tmp.file,
)
archive_extract_all(
tmp.file,
toolchains_dir / "arm-zephyr-eabi",
"toolchain",
"toolchain",
),
):
_LOGGER.info("Downloading %s %s ...", TOOLCHAIN_VERSION, what)
download_and_extract(
mirrors,
substitutions,
toolchains_dir.with_name(f"{toolchains_dir.name}.{slug}.archive"),
extract_dir,
progress_header="Extracting",
)
# Best-effort prune of resume leftovers, including a previous
# TOOLCHAIN_VERSION's orphans; the SDK archives are hundreds of MB.
# A locked file must not discard the just-completed install.
for leftover in toolchains_dir.parent.glob("*.archive.part*"):
try:
leftover.unlink()
except OSError as err:
_LOGGER.debug("Could not remove %s: %s", leftover, err)
sentinel.touch()
esphome/espidf/framework.py
+8 -15
View File
@@ -21,8 +21,8 @@ from esphome.build_helpers.tools_cache import IDF_TOOLS_CACHE, tools_cache_path
from esphome.core import Version
from esphome.framework_helpers import (
PathType,
archive_extract_all,
create_venv,
download_and_extract,
download_from_mirrors,
download_with_resume,
failure_reason,
@@ -901,20 +901,13 @@ def _check_esphome_idf_framework_install(
# a temp file) so an interrupted download resumes on the next
# run; the cache is pruned after a successful install anyway.
tarball_path = get_idf_tools_path() / "dist" / f"esp-idf-{version}.tar.xz"
download_from_mirrors(mirrors, substitutions, tarball_path)
_LOGGER.info("Extracting ESP-IDF %s framework ...", version)
try:
with tarball_path.open("rb") as tarball:
archive_extract_all(
tarball, framework_path, progress_header="Extracting"
)
finally:
# Success: drop the archive rather than caching ~70MB twice.
# Failure: a corrupt archive (e.g. torn by an unclean
# shutdown) must not be reused — without a checksum only a
# failed extraction can expose it, so force a re-download.
tarball_path.unlink(missing_ok=True)
download_and_extract(
mirrors,
substitutions,
tarball_path,
framework_path,
progress_header="Extracting",
)
extracted_marker.touch()
# Idempotent post-extract patch: written every invocation so a build
esphome/espidf/toolchain.py
+109 -20
View File
@@ -1,6 +1,7 @@
"""ESP-IDF direct build API for ESPHome."""
from dataclasses import dataclass, field
import hashlib
import json
import logging
import os
@@ -23,7 +24,7 @@ from esphome.core import CORE, EsphomeError
from esphome.espidf import variant_to_idf_target
from esphome.espidf.framework import check_esp_idf_install, get_framework_env
from esphome.espidf.size_summary import print_summary
from esphome.helpers import add_git_ceiling_directory
from esphome.helpers import add_git_ceiling_directory, write_file
_LOGGER = logging.getLogger(__name__)
@@ -256,6 +257,106 @@ def run_reconfigure() -> int:
return run_idf_py(*_get_sdkconfig_args(), "reconfigure")
def _builtin_component_cache_path() -> Path | None:
"""Cache file for this build's built-in component list.
The file lives inside the extracted framework directory so it is
discarded together with that exact checkout (re-extract, source
override, clean-all); the target and the EXCLUDE_COMPONENTS set name it.
The sdkconfig is not part of the key: IDF components register regardless
of CONFIG_* options and only gate their sources on them. A checkout
supplied through IDF_PATH is not managed by ESPHome and is never cached.
"""
if "IDF_PATH" in os.environ:
return None
target = variant_to_idf_target(CORE.data[KEY_ESP32][KEY_VARIANT])
excluded = CORE.cmake_args.get("EXCLUDE_COMPONENTS", "")
excluded_key = hashlib.sha256(excluded.encode()).hexdigest()[:12]
return (
_get_idf_path() / ".esphome_component_lists" / f"{target}-{excluded_key}.json"
)
def load_cached_builtin_components() -> list[str] | None:
"""Return the cached built-in component list for this build, if valid.
Every name must still exist under ``$IDF_PATH/components`` so a stale
entry is treated as a miss instead of failing the configure.
"""
if (path := _builtin_component_cache_path()) is None:
return None
try:
components = json.loads(path.read_text(encoding="utf-8"))
present = {
entry.name
for entry in (path.parents[1] / "components").iterdir()
if entry.is_dir()
}
except (OSError, ValueError):
return None
if (
isinstance(components, list)
and all(isinstance(c, str) for c in components)
and present.issuperset(components)
):
return components
return None
def save_cached_builtin_components(components: list[str]) -> None:
"""Store a built-in component list that just configured successfully."""
if not components or (path := _builtin_component_cache_path()) is None:
return
try:
write_file(path, json.dumps(components, separators=(",", ":")))
except EsphomeError as err:
_LOGGER.warning("Could not write component list cache %s: %s", path, err)
def _write_project_and_reconfigure(builtin_components: list[str] | None) -> int:
"""Write the full CMakeLists.txt and run the configure for it."""
from esphome.build_gen.espidf import write_project
_LOGGER.info("Writing CMakeLists.txt with the built-in component list...")
write_project(minimal=False, builtin_components=builtin_components)
# Explicit reconfigure: ninja only re-runs cmake when CMakeLists.txt
# is strictly newer than build.ninja, which fails on coarse-mtime
# filesystems (#18682). Also keeps idf.py from regenerating memory.ld
# in testing mode.
return run_reconfigure()
def _configure_project() -> int:
"""Configure the project, discovering the built-in components if needed.
A cached component list skips the discovery configure. If the configure
with a cached list fails the entry is dropped and discovery runs once; a
list is only cached after it configured successfully.
"""
from esphome.build_gen.espidf import get_available_components, write_project
if (cached := load_cached_builtin_components()) is not None:
_LOGGER.info("Using cached ESP-IDF component list")
if _write_project_and_reconfigure(cached) == 0:
return 0
_LOGGER.warning("Cached component list failed; rediscovering")
_builtin_component_cache_path().unlink(missing_ok=True)
_LOGGER.info("Discovering available ESP-IDF components...")
write_project(minimal=True)
if (rc := run_reconfigure()) != 0:
_LOGGER.error("Component discovery failed")
return rc
discovered = get_available_components()
if not discovered:
_LOGGER.error("Component discovery found no built-in ESP-IDF components")
return 1
if (rc := _write_project_and_reconfigure(discovered)) != 0:
_LOGGER.error("Reconfigure with discovered components failed")
return rc
save_cached_builtin_components(discovered)
return 0
def has_outdated_files():
"""Check if the build configuration is stale.
@@ -382,29 +483,17 @@ def run_compile(config, verbose: bool) -> int:
"""Compile the ESP-IDF project.
Uses two-phase configure to auto-discover available components:
1. If no previous build, configure with minimal REQUIRES to discover components
1. If no previous build, configure with minimal REQUIRES to discover
components (skipped when a cached list for this IDF/target/exclusion
set exists)
2. Regenerate CMakeLists.txt with discovered components
3. Run full build
"""
from esphome.build_gen.espidf import write_project
# Check if we need to do discovery phase
if need_reconfigure():
_LOGGER.info("Discovering available ESP-IDF components...")
write_project(minimal=True)
rc = run_reconfigure()
if rc != 0:
_LOGGER.error("Component discovery failed")
return rc
_LOGGER.info("Regenerating CMakeLists.txt with discovered components...")
write_project(minimal=False)
# Explicit reconfigure: ninja only re-runs cmake when CMakeLists.txt
# is strictly newer than build.ninja, which fails on coarse-mtime
# filesystems (#18682). Also keeps idf.py from regenerating memory.ld
# in testing mode.
rc = run_reconfigure()
if rc != 0:
_LOGGER.error("Reconfigure with discovered components failed")
if not need_reconfigure():
_LOGGER.info("Build configuration is up to date")
else:
if (rc := _configure_project()) != 0:
return rc
# cmake does not rewrite CMakeCache.txt when only properties change,
# so restamp it or every build repeats discovery. Only after success,
esphome/external_files.py
+5 -9
View File
@@ -5,6 +5,7 @@ from concurrent.futures import ThreadPoolExecutor
import contextlib
from dataclasses import dataclass, field
from datetime import UTC, datetime
from functools import partial
import hashlib
import logging
import os
@@ -14,9 +15,8 @@ import time
import esphome.config_validation as cv
from esphome.const import CONF_FILE, CONF_TYPE, CONF_URL, __version__
from esphome.core import CORE, EsphomeError, TimePeriodSeconds
from esphome.happy_eyeballs import ensure_happy_eyeballs
from esphome.helpers import write_file
from esphome.net_retry import fetch_with_retry
from esphome.net_retry import fetch_with_retry, http_request
from esphome.types import ConfigType
_LOGGER = logging.getLogger(__name__)
@@ -143,7 +143,6 @@ def has_remote_file_changed(
# Deferred so configs with no remote files skip the heavy import.
import requests
ensure_happy_eyeballs()
if local_file_path.exists():
_LOGGER.debug("has_remote_file_changed: File exists at %s", local_file_path)
try:
@@ -165,9 +164,7 @@ def has_remote_file_changed(
# the GET's own retry.
response = fetch_with_retry(
url,
lambda: requests.head(
url, headers=headers, timeout=timeout, allow_redirects=True
),
partial(http_request, "HEAD", url, headers=headers, timeout=timeout),
what="Revalidation",
)
@@ -282,7 +279,6 @@ def download_content(
) from failure.cause
# The file appeared since the failure; revalidate normally.
del run_data.failed_paths[path]
ensure_happy_eyeballs()
if CORE.skip_external_update and path.exists():
_LOGGER.debug("Skipping update for %s (refresh disabled)", url)
run_data.unchecked_paths.add(path)
@@ -304,7 +300,8 @@ def download_content(
_LOGGER.debug("Saving to %s", path)
def _fetch() -> tuple[requests.Response, bytes]:
req = requests.get(
req = http_request(
"GET",
url,
timeout=timeout,
headers={"User-agent": f"ESPHome/{__version__} (https://esphome.io)"},
@@ -371,7 +368,6 @@ def download_content_many(
unique = list(seen.values())
if not unique:
return
ensure_happy_eyeballs()
_LOGGER.info("Checking %d %s for updates", len(unique), description)
def _download_one(file: RemoteFile) -> None:
esphome/framework_helpers.py
+79 -141
View File
@@ -15,9 +15,12 @@ import threading
import time
from typing import IO, TYPE_CHECKING
from esphome.happy_eyeballs import ensure_happy_eyeballs
from esphome.helpers import ProgressBar, rmtree
from esphome.net_retry import NETWORK_MAX_ATTEMPTS, is_transient_download_error
from esphome.net_retry import (
NETWORK_MAX_ATTEMPTS,
http_request,
is_transient_download_error,
)
if TYPE_CHECKING:
import requests
@@ -624,12 +627,10 @@ def _open_ranged(
Raises on connect errors and HTTP error statuses; the response is closed
on failure.
"""
import requests
headers = {"Range": f"bytes={offset}-"} if offset else {}
if offset and validator:
headers["If-Range"] = validator
resp = requests.get(url, stream=True, timeout=timeout, headers=headers)
resp = http_request("GET", url, stream=True, timeout=timeout, headers=headers)
if offset and resp.status_code == 416:
resp.close()
return None, offset
@@ -965,8 +966,6 @@ def download_with_resume(
from esphome.core import EsphomeError
ensure_happy_eyeballs()
dest = Path(dest)
part = _part_path(dest)
meta = part.with_name(part.name + ".meta")
@@ -1102,20 +1101,9 @@ def failure_reason(e: BaseException) -> str:
return str(e).split(" for url: ", maxsplit=1)[0] or repr(e)
def _spent_attempts_error(e: Exception, attempts: int) -> Exception:
"""Wrap a failure whose mirror already consumed download attempts, so
the sweep classifies it as permanent."""
from esphome.core import EsphomeError
err = EsphomeError(f"failed after {attempts} attempts: {failure_reason(e)}")
err.__cause__ = e
return err
def _try_mirrors_once(
urls: list[str],
path_target: Path | None,
f: IO[bytes] | None,
path_target: Path,
timeout: int,
failures: list[tuple[str, Exception]],
progress: Callable[[int], None] | None = None,
@@ -1134,109 +1122,73 @@ def _try_mirrors_once(
for url in urls:
_LOGGER.debug("Trying to download from %s", url)
# Path targets delegate to download_with_resume so a partial
# download persists (and resumes) across esphome runs.
if path_target is not None:
try:
download_with_resume(
url,
path_target,
attempts=_MIRROR_ATTEMPTS,
timeout=timeout,
# Pre-body failures (connect/HTTP errors) fall to the
# next mirror immediately; only mid-stream drops
# retry-with-resume on the same URL.
retry_connect_errors=False,
progress=progress,
)
return url
except (requests.RequestException, OSError, EsphomeError) as e:
# Everything download_with_resume classifies as a download
# failure; programming errors propagate.
_LOGGER.debug("Failed to download %s: %s", url, str(e))
failures.append((url, e))
continue
# File-like targets download here; mid-stream failures retry the
# same mirror with resume (see download_with_resume) instead of
# starting over. There is no checksum to verify a resumed file
# against, so a stitch is only trusted when the server proves
# consistency: the If-Range validator guarantees 206 only for
# unchanged content, and the expected total length (when the first
# response carried one) guards against short or shifted bodies.
# Without a validator the retry restarts from zero.
offset = 0
expected_total = 0
validator = None
for attempt in range(_MIRROR_ATTEMPTS):
try:
resp, offset = _open_ranged(url, offset, timeout, validator)
except (requests.RequestException, OSError) as e:
# Connect/HTTP error, no bytes flowed — next mirror. Wrap
# when earlier attempts were already spent on this mirror.
_LOGGER.debug("Failed to download %s: %s", url, str(e))
failures.append(
(url, _spent_attempts_error(e, attempt + 1) if attempt else e)
)
break
try:
# A None response means HTTP 416: the file already holds
# every byte the server has (a drop after the last byte);
# only the length check below remains.
if resp is not None:
with resp:
if offset == 0:
validator = _response_validator(resp)
expected_total = _content_length(resp)
_stream_response_to_file(resp, f, offset, progress=progress)
if expected_total and f.tell() != expected_total:
raise EsphomeError(
f"size mismatch: expected {expected_total}, got {f.tell()}"
)
if not expected_total:
# Same trust decision as download_with_resume's
# unverifiable promotion; surface it at the same level.
_LOGGER.debug(
"Downloaded %s without any way to verify completeness",
url,
)
_LOGGER.debug("Downloaded successfully from: %s", url)
# Reset file pointer and return
f.seek(0)
return url
except (requests.RequestException, OSError, EsphomeError) as e:
# Mid-stream drop: keep the received bytes and retry this
# mirror from the current position — but only when the
# server gave a validator to resume against safely AND a
# total length to prove the stitched file complete (the
# length check above is the only verification here).
_LOGGER.debug("Failed to download %s: %s", url, str(e))
if validator and expected_total:
offset = f.tell()
else:
_LOGGER.debug(
"Restarting %s from zero: cannot prove a "
"resumed file complete (validator=%s, total=%s)",
url,
validator is not None,
expected_total,
)
offset = 0
if attempt == _MIRROR_ATTEMPTS - 1:
failures.append((url, _spent_attempts_error(e, _MIRROR_ATTEMPTS)))
# Delegate to download_with_resume so a partial download persists
# (and resumes) across esphome runs.
try:
download_with_resume(
url,
path_target,
attempts=_MIRROR_ATTEMPTS,
timeout=timeout,
# Pre-body failures (connect/HTTP errors) fall to the
# next mirror immediately; only mid-stream drops
# retry-with-resume on the same URL.
retry_connect_errors=False,
progress=progress,
)
return url
except (requests.RequestException, OSError, EsphomeError) as e:
# Everything download_with_resume classifies as a download
# failure; programming errors propagate.
_LOGGER.debug("Failed to download %s: %s", url, str(e))
failures.append((url, e))
return None
def download_and_extract(
mirrors: list[str],
substitutions: dict[str, str],
archive_path: PathType,
extract_dir: PathType,
timeout: int = 30,
progress_header: str | None = None,
progress: Callable[[int], None] | None = None,
) -> str:
"""Download an archive from ``mirrors`` to ``archive_path``, extract it
into ``extract_dir``, and delete the archive.
The archive should live next to its destination (not in a temp dir) so
an interrupted download's ``.part`` file resumes on the next run. The
archive is deleted whether extraction succeeds or fails: a
complete-but-corrupt file (e.g. torn by an unclean shutdown) must not
poison the next run, and without a checksum only a failed extraction
can expose it.
Returns the source URL the download came from.
"""
archive_path = Path(archive_path)
url = download_from_mirrors(
mirrors, substitutions, archive_path, timeout=timeout, progress=progress
)
try:
archive_extract_all(archive_path, extract_dir, progress_header=progress_header)
finally:
# Best-effort: an AV handle on the just-written archive (Windows)
# must not replace the real extraction error or fail a successful
# extraction. A surviving archive is harmless; download_with_resume
# re-verifies or re-downloads it next run.
try:
archive_path.unlink(missing_ok=True)
except OSError as err:
_LOGGER.debug("Could not remove archive %s: %s", archive_path, err)
return url
def download_from_mirrors(
mirrors: list[str],
substitutions: dict[str, str],
target: io.RawIOBase | IO[bytes] | PathType,
target: PathType,
timeout: int = 30,
progress: Callable[[int], None] | None = None,
) -> str:
@@ -1246,7 +1198,7 @@ def download_from_mirrors(
Args:
mirrors: list of mirror URLs
substitutions: Dictionary of substitutions to apply to URLs
target: Target file path or file-like object
target: Target file path
timeout: Download timeout in seconds
progress: Passed through to the download (see ``download_with_resume``);
replaces the built-in per-file bar
@@ -1258,9 +1210,8 @@ def download_from_mirrors(
``substitutions`` are skipped, so callers can offer templates that only
apply to some downloads.
A path target downloads through ``download_with_resume``, so an
interrupted download resumes on the next esphome run; a file-like target
only resumes mid-stream drops within this call.
The target downloads through ``download_with_resume``, so an
interrupted download resumes on the next esphome run.
When every mirror fails and at least one failure is transient (dropped
connection, timeout, HTTP 429/5xx), the whole list is retried with a
@@ -1274,21 +1225,11 @@ def download_from_mirrors(
"""
from esphome.core import EsphomeError
ensure_happy_eyeballs()
if not isinstance(target, (str, os.PathLike)):
raise TypeError(f"target must be a str or Path: {type(target)}")
path_target = Path(target)
# 1. Classify the target: filesystem path or open file object
path_target: Path | None = None
f: IO[bytes] | None = None
if isinstance(target, (str, os.PathLike)):
path_target = Path(target)
elif isinstance(target, (io.RawIOBase, io.IOBase)):
f = target
else:
raise TypeError(
f"target must be str, Path, or file-like object: {type(target)}"
)
# 2. Resolve the mirror templates (invariant across retry sweeps)
# 1. Resolve the mirror templates (invariant across retry sweeps)
urls: list[str] = []
skipped: list[tuple[str, str]] = []
for mirror in mirrors:
@@ -1307,7 +1248,7 @@ def download_from_mirrors(
_LOGGER.warning("Skipping malformed mirror URL template %s: %r", mirror, e)
skipped.append((mirror, f"skipped ({e!r})"))
# 3. Sweep the mirror list, retrying transient failures with backoff:
# 2. Sweep the mirror list, retrying transient failures with backoff:
# a single pass keeps mirror failover fast, re-sweeping keeps one
# network blip from failing the build when only one mirror applies.
failures: list[tuple[str, Exception]] = []
@@ -1315,7 +1256,7 @@ def download_from_mirrors(
sweep_failures: list[tuple[str, Exception]] = []
if (
url := _try_mirrors_once(
urls, path_target, f, timeout, sweep_failures, progress
urls, path_target, timeout, sweep_failures, progress
)
) is not None:
return url
@@ -1342,14 +1283,11 @@ def download_from_mirrors(
# steady during the backoff instead of rewinding to zero
done = 0
if progress is not None:
if f is not None:
done = f.tell()
else:
part = _part_path(path_target)
done = part.stat().st_size if part.is_file() else 0
part = _part_path(path_target)
done = part.stat().st_size if part.is_file() else 0
_cancellable_sleep(delay, progress, done)
# 4. Report every attempted URL if all mirrors failed. failures spans
# 3. Report every attempted URL if all mirrors failed. failures spans
# all sweeps (deduplicated by URL and reason), so neither an early
# mirror's failure nor an earlier sweep's failure mode is hidden.
if failures:
esphome/happy_eyeballs.py
+13 -5
View File
@@ -12,6 +12,7 @@ from __future__ import annotations
import logging
import socket
import threading
from typing import TYPE_CHECKING, Any
if TYPE_CHECKING:
@@ -27,20 +28,27 @@ HAPPY_EYEBALLS_DELAY = 0.25
_THREAD_WAIT_BUFFER = 5.0
# Serialises the check-then-patch so concurrent first calls (download worker
# threads fanning out) build the replacement exactly once.
_PATCH_LOCK = threading.Lock()
def ensure_happy_eyeballs() -> None:
"""Make urllib3 (and therefore requests) connect with Happy Eyeballs.
Idempotent; call before performing requests-based downloads.
Idempotent and thread-safe; call before performing requests-based
downloads.
"""
stock: Callable[..., socket.socket] | None = None
try:
import urllib3.util.connection
stock = urllib3.util.connection.create_connection
if getattr(stock, "_esphome_patched", False):
return
with _PATCH_LOCK:
stock = urllib3.util.connection.create_connection
if getattr(stock, "_esphome_patched", False):
return
urllib3.util.connection.create_connection = _make_create_connection()
urllib3.util.connection.create_connection = _make_create_connection()
except (ImportError, AttributeError) as err: # urllib3 internals moved
# WARNING: degraded mode brings back the stalls this module prevents.
_LOGGER.warning(
esphome/net_retry.py
+38 -1
View File
@@ -1,4 +1,4 @@
"""Retry policy for HTTP downloads.
"""Retry policy and raw HTTP entry point for downloads.
Kept import-light on purpose: this module is imported at config time, so it
must not pull in requests (a heavy import, ~85ms) at module scope.
@@ -9,6 +9,12 @@ from __future__ import annotations
from collections.abc import Callable
import logging
import time
from typing import TYPE_CHECKING, Literal
from esphome.happy_eyeballs import ensure_happy_eyeballs
if TYPE_CHECKING:
import requests
_LOGGER = logging.getLogger(__name__)
@@ -112,3 +118,34 @@ def fetch_with_retry[T](url: str, fetch: Callable[[], T], what: str = "Download"
)
time.sleep(delay)
return fetch()
def http_request(
method: Literal["GET", "HEAD"],
url: str,
*,
timeout: float | tuple[float, float],
stream: bool = False,
headers: dict[str, str] | None = None,
allow_redirects: bool = True,
) -> requests.Response:
"""Perform one HTTP request with the Happy Eyeballs patch in place.
Every ESPHome file download funnels through here so the urllib3 patch
and the lazy requests import live in exactly one place. Status handling,
retries and streaming stay with the caller. The web server OTA and log
clients bypass this on purpose: they iterate already-resolved device
addresses themselves, so the patch buys them nothing.
"""
import requests
ensure_happy_eyeballs()
# Dispatched through requests.get/head/... (not requests.request) so
# tests patching those entry points keep working.
return getattr(requests, method.lower())(
url,
timeout=timeout,
stream=stream,
headers=headers or {},
allow_redirects=allow_redirects,
)
esphome/platformio/library.py
+15 -14
View File
@@ -24,7 +24,6 @@ import logging
import os
from pathlib import Path, PurePosixPath
import re
import tempfile
from typing import Any
from urllib.parse import urlsplit, urlunsplit
from urllib.request import url2pathname
@@ -32,8 +31,7 @@ from urllib.request import url2pathname
from esphome import git
from esphome.core import CORE, EsphomeError, Library
from esphome.framework_helpers import (
archive_extract_all,
download_from_mirrors,
download_and_extract,
failure_reason,
rmdir,
run_batch_downloads,
@@ -152,18 +150,21 @@ class URLSource(Source):
if not extracted_marker.is_file() or force:
rmdir(path, msg=f"Clean up library directory {path}")
# Download in temporary file
with tempfile.NamedTemporaryFile() as tmp:
if progress is None:
# A batch caller draws one combined bar and logs the list
_LOGGER.info("Downloading %s ...", self.url)
_LOGGER.debug("Location: %s", path)
if progress is None:
# A batch caller draws one combined bar and logs the list
_LOGGER.info("Downloading %s ...", self.url)
_LOGGER.debug("Location: %s", path)
download_from_mirrors([self.url], {}, tmp.file, progress=progress)
_LOGGER.debug("Extracting archive to %s ...", path)
archive_extract_all(tmp.file, path)
extracted_marker.touch()
# The sibling archive path lets an interrupted download's .part
# file survive and resume on the next esphome run.
download_and_extract(
[self.url],
{},
path.with_name(f"{path.name}.archive"),
path,
progress=progress,
)
extracted_marker.touch()
return path
def __str__(self):
esphome/platformio/registry.py
+22 -8
View File
@@ -5,7 +5,6 @@ from __future__ import annotations
from collections.abc import Callable, Collection
from functools import cache, partial
import io
import json
import logging
import os
@@ -21,6 +20,7 @@ from esphome.framework_helpers import (
rmdir,
run_batch_downloads,
)
from esphome.net_retry import fetch_with_retry, http_request
_LOGGER = logging.getLogger(__name__)
@@ -57,15 +57,29 @@ def get_systype() -> str:
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.
Cached per process so the prefetch and the install resolve each package
once (failures are not cached; the install retries them).
The metadata fetch goes through ``http_request``/``fetch_with_retry``
(the consolidated HTTP path) so it shares the Happy Eyeballs patch and
transient-retry policy of every other small fetch. Cached per process
so the prefetch and the install resolve each package once (failures
are not cached; the install retries them).
"""
buf = io.BytesIO()
download_from_mirrors([_REGISTRY_URL], {"package": package}, buf)
url = _REGISTRY_URL.format(package=package)
def _fetch() -> str:
resp = http_request("GET", url, timeout=30)
resp.raise_for_status()
return resp.text
import requests
try:
data = json.loads(buf.getvalue())
body = fetch_with_retry(url, _fetch, what="Registry lookup")
except requests.exceptions.RequestException as err:
raise EsphomeError(
f"Could not fetch registry metadata for {package}: {err}"
) from err
try:
data = json.loads(body)
except ValueError as err:
raise EsphomeError(
f"The package registry returned invalid JSON for {package}: {err}"
tests/component_tests/emontx/test_sensor_defaults.py
+109
View File
@@ -6,9 +6,28 @@ from esphome.components import sensor
from esphome.components.emontx.sensor import CONFIG_SCHEMA, apply_tag_defaults
from esphome.const import (
CONF_ACCURACY_DECIMALS,
CONF_DEVICE_CLASS,
CONF_STATE_CLASS,
CONF_UNIT_OF_MEASUREMENT,
DEVICE_CLASS_APPARENT_POWER,
DEVICE_CLASS_CURRENT,
DEVICE_CLASS_ENERGY,
DEVICE_CLASS_FREQUENCY,
DEVICE_CLASS_POWER,
DEVICE_CLASS_POWER_FACTOR,
DEVICE_CLASS_TEMPERATURE,
DEVICE_CLASS_VOLTAGE,
STATE_CLASS_MEASUREMENT,
STATE_CLASS_TOTAL_INCREASING,
UNIT_AMPERE,
UNIT_CELSIUS,
UNIT_EMPTY,
UNIT_HERTZ,
UNIT_PULSES,
UNIT_VOLT,
UNIT_VOLT_AMPS,
UNIT_WATT,
UNIT_WATT_HOURS,
)
@@ -61,9 +80,25 @@ def _make_config(tag: str) -> dict:
("PULSE1", STATE_CLASS_TOTAL_INCREASING, 0),
("PULSE12", STATE_CLASS_TOTAL_INCREASING, 0),
("PF1", STATE_CLASS_MEASUREMENT, 2),
("AP1", STATE_CLASS_MEASUREMENT, 2),
("AP12", STATE_CLASS_MEASUREMENT, 2),
# Frequency: reported as a single, un-numbered tag
("F", STATE_CLASS_MEASUREMENT, 2),
# Unknown / free-form tags fall back to generic defaults
("CUSTOM1", STATE_CLASS_MEASUREMENT, 0),
("X", STATE_CLASS_MEASUREMENT, 0),
# "F1" is not the exact "F" tag, so it falls back to generic defaults
("F1", STATE_CLASS_MEASUREMENT, 0),
# "PULSE" (no index) is how some real emonTx firmware reports a
# single pulse counter, so it still resolves to the PULSE defaults
("PULSE", STATE_CLASS_TOTAL_INCREASING, 0),
# Real firmware sends this lowercase; tag_upper's case-folding must
# still match it against the PULSE pattern
("pulse", STATE_CLASS_TOTAL_INCREASING, 0),
# PF/AP require a numeric index; the bare prefix alone (no index)
# falls back to generic defaults
("PF", STATE_CLASS_MEASUREMENT, 0),
("AP", STATE_CLASS_MEASUREMENT, 0),
],
)
def test_apply_tag_defaults(tag, expected_state_class, expected_decimals):
@@ -76,6 +111,80 @@ def test_apply_tag_defaults(tag, expected_state_class, expected_decimals):
assert result[CONF_ACCURACY_DECIMALS] == expected_decimals
@pytest.mark.parametrize(
("tag", "expected_unit", "expected_device_class"),
[
# Known numeric-index prefixes
("E1", UNIT_WATT_HOURS, DEVICE_CLASS_ENERGY),
("E12", UNIT_WATT_HOURS, DEVICE_CLASS_ENERGY),
("P1", UNIT_WATT, DEVICE_CLASS_POWER),
("V1", UNIT_VOLT, DEVICE_CLASS_VOLTAGE),
("I1", UNIT_AMPERE, DEVICE_CLASS_CURRENT),
("T1", UNIT_CELSIUS, DEVICE_CLASS_TEMPERATURE),
# Known patterns
("PULSE1", UNIT_PULSES, DEVICE_CLASS_ENERGY),
("PULSE12", UNIT_PULSES, DEVICE_CLASS_ENERGY),
# Bare "PULSE" (no index), as reported by some real emonTx firmware
("PULSE", UNIT_PULSES, DEVICE_CLASS_ENERGY),
# Real firmware sends this lowercase; tag_upper's case-folding must
# still match it against the PULSE pattern
("pulse", UNIT_PULSES, DEVICE_CLASS_ENERGY),
("PF1", UNIT_EMPTY, DEVICE_CLASS_POWER_FACTOR),
("AP1", UNIT_VOLT_AMPS, DEVICE_CLASS_APPARENT_POWER),
("AP12", UNIT_VOLT_AMPS, DEVICE_CLASS_APPARENT_POWER),
# Frequency: reported as a single, un-numbered tag
("F", UNIT_HERTZ, DEVICE_CLASS_FREQUENCY),
],
)
def test_apply_tag_defaults_unit_and_device_class(
tag, expected_unit, expected_device_class
):
"""apply_tag_defaults must inject the correct, validated unit_of_measurement
and device_class for each tag type when no user overrides are present."""
config = _make_config(tag)
result = apply_tag_defaults(config)
assert result[CONF_UNIT_OF_MEASUREMENT] == sensor.validate_unit_of_measurement(
expected_unit
)
assert result[CONF_DEVICE_CLASS] == sensor.validate_device_class(
expected_device_class
)
@pytest.mark.parametrize(
"tag",
[
"CUSTOM1",
"X",
# Non-numeric suffixes must not collide with a PATTERN_CONFIGS prefix
# (e.g. "APPLE" starting with "AP", "PFX" starting with "PF").
"APPLE",
"PFX",
"PULSE_A",
# "F1" is not the exact "F" tag
"F1",
# Bare "PF"/"AP" (no numeric index) don't match; unlike "PULSE",
# real firmware never reports these without an index
"PF",
"AP",
],
)
def test_apply_tag_defaults_unknown_tag_has_no_unit_or_device_class(tag):
"""Unknown / free-form tags only get generic state_class and
accuracy_decimals defaults; unit_of_measurement and device_class are left
for the user to set explicitly."""
config = _make_config(tag)
result = apply_tag_defaults(config)
assert CONF_UNIT_OF_MEASUREMENT not in result
assert CONF_DEVICE_CLASS not in result
assert result[CONF_STATE_CLASS] == sensor.validate_state_class(
STATE_CLASS_MEASUREMENT
)
assert result[CONF_ACCURACY_DECIMALS] == 0
@pytest.mark.parametrize(
("tag", "user_state_class", "user_decimals"),
[
tests/components/emontx/validate.esp32-idf.yaml
+15
View File
@@ -57,6 +57,21 @@ sensor:
name: Power Factor 1
emontx_id: test_emontx
# Apparent power sensor (AP pattern): expects state_class=measurement,
# unit=VA, device_class=apparent_power, accuracy_decimals=2
- platform: emontx
tag_name: AP1
name: Apparent Power 1
emontx_id: test_emontx
# Frequency sensor (F, matched exactly, not as a prefix): expects
# state_class=measurement, unit=Hz, device_class=frequency,
# accuracy_decimals=2
- platform: emontx
tag_name: F
name: Frequency
emontx_id: test_emontx
# Unknown tag: no prefix match, falls back to state_class=measurement,
# accuracy_decimals=0
- platform: emontx
tests/unit_tests/build_gen/test_espidf.py
+85 -6
View File
@@ -3,6 +3,7 @@
from __future__ import annotations
import json
import logging
from pathlib import Path
from unittest.mock import patch
@@ -35,22 +36,25 @@ def _reset_core(tmp_path: Path) -> None:
}
def _write_project_description(tmp_path: Path, components: dict[str, str]) -> None:
def _write_project_description(
tmp_path: Path, components: dict[str, str], idf_path: str = "/idf"
) -> None:
"""Stub a project_description.json with the given component_name -> dir map."""
build_dir = tmp_path / "build"
build_dir.mkdir(exist_ok=True)
(build_dir / "project_description.json").write_text(
json.dumps(
{
"idf_path": idf_path,
"build_component_info": {
name: {"dir": dir_} for name, dir_ in components.items()
}
},
}
)
)
def _render(minimal: bool = False) -> str:
def _render(minimal: bool = False, builtin_components: list[str] | None = None) -> str:
"""Render the top-level CMakeLists with the standard variant/name patches."""
with (
patch("esphome.build_gen.espidf.get_esp32_variant", return_value="ESP32"),
@@ -58,7 +62,9 @@ def _render(minimal: bool = False) -> str:
):
from esphome.build_gen.espidf import get_project_cmakelists
return get_project_cmakelists(minimal=minimal)
return get_project_cmakelists(
minimal=minimal, builtin_components=builtin_components
)
def test_get_available_components_returns_none_without_build_path() -> None:
@@ -77,8 +83,11 @@ def test_get_available_components_returns_none_without_project_description(
assert get_available_components() is None
def test_get_available_components_filters_src_managed_and_pio(tmp_path: Path) -> None:
"""Built-ins are returned; src/, managed_components/, pio_components/ skipped."""
def test_get_available_components_keeps_only_idf_tree_components(
tmp_path: Path,
) -> None:
"""Only components under idf_path/components are built-ins: src, managed,
converted PIO libs and Arduino component_stubs are all left out."""
_write_project_description(
tmp_path,
{
@@ -86,6 +95,7 @@ def test_get_available_components_filters_src_managed_and_pio(tmp_path: Path) ->
"esp_lcd": "/idf/components/esp_lcd",
"espressif__arduino-esp32": f"{tmp_path}/managed_components/arduino",
"JPEGDEC": f"{tmp_path}/pio_components/arduino/abc/bitbank2/JPEGDEC",
"cbor": f"{tmp_path}/component_stubs/cbor",
"freertos": "/idf/components/freertos",
},
)
@@ -94,6 +104,75 @@ def test_get_available_components_filters_src_managed_and_pio(tmp_path: Path) ->
assert sorted(get_available_components()) == ["esp_lcd", "freertos"]
def test_codegen_and_configure_writes_render_the_same_cmakelists(
tmp_path: Path,
) -> None:
"""write_project() at codegen time (no list) and the configure-time write
(discovered list) must agree, or ninja re-runs cmake on every build."""
_write_project_description(
tmp_path,
{
"lwip": "/idf/components/lwip",
"cbor": f"{tmp_path}/component_stubs/cbor",
},
)
from esphome.build_gen.espidf import get_available_components
assert _render() == _render(builtin_components=get_available_components())
assert "ESPHOME_PROJECT_BUILTIN_COMPONENTS cbor" not in _render()
def test_get_available_components_warns_when_nothing_is_under_idf_path(
tmp_path: Path, caplog: pytest.LogCaptureFixture
) -> None:
_write_project_description(tmp_path, {"cbor": f"{tmp_path}/component_stubs/cbor"})
from esphome.build_gen.espidf import (
get_available_components,
has_discovered_components,
)
assert get_available_components() == []
assert "No ESP-IDF components found under" in caplog.text
# An empty discovery must not count as configured, or it would be latched in.
assert not has_discovered_components()
def test_get_available_components_ignores_corrupt_or_unexpected_file(
tmp_path: Path, caplog: pytest.LogCaptureFixture
) -> None:
build_dir = tmp_path / "build"
build_dir.mkdir()
from esphome.build_gen.espidf import (
get_available_components,
has_discovered_components,
)
(build_dir / "project_description.json").write_text("{not json")
assert get_available_components() is None
assert not has_discovered_components()
(build_dir / "project_description.json").write_text('{"build_component_info": {}}')
with caplog.at_level(logging.DEBUG, logger="esphome.build_gen.espidf"):
assert get_available_components() is None
assert "Could not read" in caplog.text
def test_has_discovered_components_after_configure(tmp_path: Path) -> None:
_write_project_description(tmp_path, {"lwip": "/idf/components/lwip"})
from esphome.build_gen.espidf import has_discovered_components
assert has_discovered_components()
def test_get_project_cmakelists_uses_supplied_builtin_components() -> None:
"""A cached list replaces project_description.json and is still filtered
by EXCLUDE_COMPONENTS."""
with patch.dict(CORE.cmake_args, {"EXCLUDE_COMPONENTS": "fatfs;unity"}):
content = _render(builtin_components=["lwip", "fatfs", "esp_timer"])
assert "ESPHOME_PROJECT_BUILTIN_COMPONENTS esp_timer APPEND" in content
assert "ESPHOME_PROJECT_BUILTIN_COMPONENTS lwip APPEND" in content
assert "ESPHOME_PROJECT_BUILTIN_COMPONENTS fatfs APPEND" not in content
def test_get_project_cmakelists_minimal_omits_builtin_components_property(
tmp_path: Path,
) -> None:
tests/unit_tests/test_dashboard_import.py
+55
View File
@@ -10,8 +10,10 @@ during the adoption flow and depend on the output's ``esphome.name``
from __future__ import annotations
from pathlib import Path
from unittest.mock import MagicMock, patch
import pytest
import requests as req
import yaml as pyyaml
from esphome.components.dashboard_import import import_config
@@ -201,3 +203,56 @@ def test_import_refuses_to_overwrite_existing_yaml(tmp_path: Path) -> None:
)
# Original content survives unchanged.
assert yaml_path.read_text() == "# user's hand-edited config\n"
def _full_config_kwargs(yaml_path: Path) -> dict:
return {
"path": str(yaml_path),
"name": "kitchen",
"friendly_name": None,
"project_name": "acme.kitchen-light",
"import_url": "github://acme/firmware/kitchen.yaml@main?full_config",
}
def test_full_config_import_fetches_and_writes_contents(tmp_path: Path) -> None:
yaml_path = tmp_path / "kitchen.yaml"
resp = MagicMock(text="esphome:\n name: orig\n")
with patch(
"esphome.components.dashboard_import.http_request", return_value=resp
) as mock_req:
import_config(**_full_config_kwargs(yaml_path))
assert yaml_path.read_text() == "esphome:\n name: orig\n"
assert mock_req.call_args[0][0] == "GET"
def test_full_config_import_retries_transient_errors(tmp_path: Path) -> None:
"""The fetch goes through the shared retry policy: a transient network
error is retried instead of failing the adoption immediately."""
yaml_path = tmp_path / "kitchen.yaml"
resp = MagicMock(text="esphome:\n name: orig\n")
with (
patch(
"esphome.components.dashboard_import.http_request",
side_effect=[req.ConnectionError("reset"), resp],
),
patch("esphome.net_retry.time.sleep") as mock_sleep,
):
import_config(**_full_config_kwargs(yaml_path))
assert yaml_path.exists()
mock_sleep.assert_called_once_with(2)
def test_full_config_import_wraps_permanent_errors_in_value_error(
tmp_path: Path,
) -> None:
"""device-builder depends on the ValueError contract for fetch failures."""
resp = MagicMock()
resp.raise_for_status.side_effect = req.HTTPError(
"404", response=MagicMock(status_code=404)
)
with (
patch("esphome.components.dashboard_import.http_request", return_value=resp),
pytest.raises(ValueError, match="Error while fetching"),
):
import_config(**_full_config_kwargs(tmp_path / "kitchen.yaml"))
tests/unit_tests/test_espidf_framework.py
+10 -9
View File
@@ -371,10 +371,9 @@ def _fake_download_from_mirrors(
) -> str:
"""Stand-in for download_from_mirrors that creates path targets, since
the framework code opens the downloaded tarball afterwards."""
if isinstance(target, (str, os.PathLike)):
path = Path(target)
path.parent.mkdir(parents=True, exist_ok=True)
path.touch()
path = Path(target)
path.parent.mkdir(parents=True, exist_ok=True)
path.touch()
return "https://example.com/idf.tar.xz"
@@ -384,13 +383,15 @@ def espidf_mocks(setup_core: Path):
# archive_extract_all is mocked, so pre-create the framework dir that the
# extracted-marker touch writes into.
_get_framework_path(_IDF_VERSION).mkdir(parents=True, exist_ok=True)
# One mock covers the tarball (via framework_helpers.download_and_extract)
# and the constraints file (espidf-bound download_from_mirrors), so call
# counts and ordering assertions span the two.
download = MagicMock(side_effect=_fake_download_from_mirrors)
with (
patch("esphome.espidf.framework.rmdir") as rmdir_mock,
patch(
"esphome.espidf.framework.download_from_mirrors",
side_effect=_fake_download_from_mirrors,
) as download,
patch("esphome.espidf.framework.archive_extract_all") as extract,
patch("esphome.framework_helpers.download_from_mirrors", download),
patch("esphome.espidf.framework.download_from_mirrors", download),
patch("esphome.framework_helpers.archive_extract_all") as extract,
patch("esphome.espidf.framework.create_venv") as venv,
patch("esphome.espidf.framework.run_command_ok", return_value=True) as run_ok,
patch(
tests/unit_tests/test_espidf_toolchain.py
+241 -1
View File
@@ -2,6 +2,8 @@
# pylint: disable=protected-access
from collections.abc import Iterator
from contextlib import contextmanager
import json
import os
from pathlib import Path
@@ -309,6 +311,11 @@ def test_run_compile_restamps_cmakecache_after_discovery(setup_core: Path) -> No
with (
patch.object(toolchain, "need_reconfigure", return_value=True),
patch.object(toolchain, "load_cached_builtin_components", return_value=None),
patch.object(toolchain, "save_cached_builtin_components"),
patch(
"esphome.build_gen.espidf.get_available_components", return_value=["lwip"]
),
patch("esphome.build_gen.espidf.write_project"),
patch.object(toolchain, "run_reconfigure", return_value=0),
patch.object(toolchain, "run_idf_py", return_value=0),
@@ -329,6 +336,11 @@ def test_run_compile_discovery_without_cmakecache(setup_core: Path) -> None:
with (
patch.object(toolchain, "need_reconfigure", return_value=True),
patch.object(toolchain, "load_cached_builtin_components", return_value=None),
patch.object(toolchain, "save_cached_builtin_components"),
patch(
"esphome.build_gen.espidf.get_available_components", return_value=["lwip"]
),
patch("esphome.build_gen.espidf.write_project"),
patch.object(toolchain, "run_reconfigure", return_value=0),
patch.object(toolchain, "run_idf_py", return_value=0),
@@ -354,7 +366,7 @@ def test_run_compile_reconfigures_after_full_write_outside_testing_mode(
calls: list[tuple] = []
reconfigures = 0
def record_write(minimal: bool = False) -> None:
def record_write(minimal: bool = False, builtin_components=None) -> None:
calls.append(("write_project", minimal))
def record_reconfigure() -> int:
@@ -365,6 +377,11 @@ def test_run_compile_reconfigures_after_full_write_outside_testing_mode(
with (
patch.object(toolchain, "need_reconfigure", return_value=True),
patch.object(toolchain, "load_cached_builtin_components", return_value=None),
patch.object(toolchain, "save_cached_builtin_components"),
patch(
"esphome.build_gen.espidf.get_available_components", return_value=["lwip"]
),
patch("esphome.build_gen.espidf.write_project", side_effect=record_write),
patch.object(toolchain, "run_reconfigure", side_effect=record_reconfigure),
patch.object(toolchain, "run_idf_py", return_value=0) as mock_build,
@@ -383,6 +400,229 @@ def test_run_compile_reconfigures_after_full_write_outside_testing_mode(
assert cmakecache.stat().st_mtime == old
def _record_compile_calls(
cached: list[str] | None,
saved: list[str] | None = None,
reconfigure_rcs: tuple[int, ...] = (),
cache_file: Path | None = None,
) -> tuple[int, list[tuple]]:
"""Run run_compile with a stubbed cache and return (rc, call log).
``reconfigure_rcs`` overrides the exit codes of the first reconfigures;
later ones succeed.
"""
calls: list[tuple] = []
rcs = iter(reconfigure_rcs)
def record_reconfigure() -> int:
calls.append(("run_reconfigure",))
return next(rcs, 0)
def record_write(minimal: bool = False, builtin_components=None) -> None:
calls.append(("write_project", minimal, builtin_components))
def record_save(components: list[str]) -> None:
calls.append(("save", components))
with (
patch.object(toolchain, "need_reconfigure", return_value=True),
patch.object(toolchain, "load_cached_builtin_components", return_value=cached),
patch.object(
toolchain, "save_cached_builtin_components", side_effect=record_save
),
patch("esphome.build_gen.espidf.get_available_components", return_value=saved),
patch("esphome.build_gen.espidf.write_project", side_effect=record_write),
patch.object(toolchain, "run_reconfigure", side_effect=record_reconfigure),
patch.object(
toolchain, "_builtin_component_cache_path", return_value=cache_file
),
patch.object(
toolchain,
"run_idf_py",
side_effect=lambda *a, **kw: calls.append(("build",)) or 0,
),
patch.object(toolchain, "print_summary"),
):
rc = toolchain.run_compile({CONF_ESPHOME: {}}, verbose=False)
return rc, calls
def test_run_compile_poisoned_cache_is_dropped_and_rediscovered(
setup_core: Path, tmp_path: Path
) -> None:
"""A cached list that fails the configure is deleted and discovery runs
once more instead of every later build failing the same way."""
_setup_build(setup_core)
cache_file = tmp_path / "esp32-abc.json"
cache_file.write_text("[]")
rc, calls = _record_compile_calls(
["stale"], saved=["lwip"], reconfigure_rcs=(1,), cache_file=cache_file
)
assert rc == 0
assert not cache_file.exists()
assert calls == [
("write_project", False, ["stale"]),
("run_reconfigure",),
("write_project", True, None),
("run_reconfigure",),
("write_project", False, ["lwip"]),
("run_reconfigure",),
("save", ["lwip"]),
("build",),
]
def test_run_compile_cache_miss_discovers_and_saves(setup_core: Path) -> None:
"""Without a cached list the discovery configure runs, the discovered list
feeds the full write and is cached only after that configure succeeds."""
_setup_build(setup_core)
rc, calls = _record_compile_calls(None, saved=["lwip"])
assert rc == 0
assert calls == [
("write_project", True, None),
("run_reconfigure",),
("write_project", False, ["lwip"]),
("run_reconfigure",),
("save", ["lwip"]),
("build",),
]
def test_run_compile_discovery_failure_stops_before_full_write(
setup_core: Path,
) -> None:
"""A failed discovery configure returns its exit code and never writes
the full CMakeLists, a cache entry or a build."""
_setup_build(setup_core)
rc, calls = _record_compile_calls(None, reconfigure_rcs=(2,))
assert rc == 2
assert calls == [("write_project", True, None), ("run_reconfigure",)]
@pytest.mark.parametrize("discovered", [None, []], ids=["no_manifest", "empty"])
def test_run_compile_fails_when_discovery_finds_nothing(
setup_core: Path,
caplog: pytest.LogCaptureFixture,
discovered: list[str] | None,
) -> None:
_setup_build(setup_core)
rc, calls = _record_compile_calls(None, saved=discovered)
assert rc == 1
assert calls == [("write_project", True, None), ("run_reconfigure",)]
assert "found no built-in ESP-IDF components" in caplog.text
def test_run_compile_does_not_cache_a_list_that_failed_to_configure(
setup_core: Path,
) -> None:
_setup_build(setup_core)
rc, calls = _record_compile_calls(None, saved=["lwip"], reconfigure_rcs=(0, 3))
assert rc == 3
assert ("save", ["lwip"]) not in calls
assert ("build",) not in calls
def test_run_compile_cache_hit_skips_discovery(setup_core: Path) -> None:
"""A cached list goes straight to the full write; the explicit reconfigure
after it (#18730) still runs."""
_setup_build(setup_core)
rc, calls = _record_compile_calls(["esp_timer", "lwip"])
assert rc == 0
assert calls == [
("write_project", False, ["esp_timer", "lwip"]),
("run_reconfigure",),
("build",),
]
@contextmanager
def _cache_env(tmp_path: Path, excluded: str) -> Iterator[Path]:
"""Patch everything the cache key derives from onto a temp IDF tree and
yield that tree's path."""
idf_path = tmp_path / "idf"
(idf_path / "components").mkdir(parents=True, exist_ok=True)
with (
patch.object(toolchain, "_get_idf_path", return_value=idf_path),
patch.dict(CORE.data, {KEY_ESP32: {KEY_VARIANT: "ESP32"}}),
patch.dict(CORE.cmake_args, {"EXCLUDE_COMPONENTS": excluded}),
):
yield idf_path
def test_component_cache_round_trip(setup_core: Path, tmp_path: Path) -> None:
"""A saved list is read back until it is dropped."""
_setup_build(setup_core)
with _cache_env(tmp_path, "fatfs") as idf_path:
for name in ("lwip", "esp_timer"):
(idf_path / "components" / name).mkdir()
assert toolchain.load_cached_builtin_components() is None
toolchain.save_cached_builtin_components(["esp_timer", "lwip"])
assert toolchain.load_cached_builtin_components() == ["esp_timer", "lwip"]
toolchain._builtin_component_cache_path().unlink()
assert toolchain.load_cached_builtin_components() is None
def test_component_cache_misses_on_key_change_or_missing_component(
setup_core: Path, tmp_path: Path
) -> None:
"""A different exclusion set uses another entry, an entry naming a
component that no longer exists is ignored, and a custom IDF_PATH is
never cached."""
_setup_build(setup_core)
with _cache_env(tmp_path, "fatfs") as idf_path:
(idf_path / "components" / "lwip").mkdir()
toolchain.save_cached_builtin_components(["lwip"])
path = toolchain._builtin_component_cache_path()
assert path.parent == idf_path / ".esphome_component_lists"
assert path.name.startswith("esp32-")
assert toolchain.load_cached_builtin_components() == ["lwip"]
with patch.dict(os.environ, {"IDF_PATH": str(idf_path)}):
assert toolchain.load_cached_builtin_components() is None
with _cache_env(tmp_path, "fatfs;unity"):
assert toolchain.load_cached_builtin_components() is None
with _cache_env(tmp_path, "fatfs") as idf_path:
path.write_text(json.dumps(["lwip", "gone"]))
assert toolchain.load_cached_builtin_components() is None
# A plain file with the right name is not a component directory.
(idf_path / "components" / "gone").write_text("not a directory")
assert toolchain.load_cached_builtin_components() is None
def test_component_cache_save_skips_empty_list_or_custom_idf_path(
setup_core: Path, tmp_path: Path
) -> None:
_setup_build(setup_core)
with _cache_env(tmp_path, "") as idf_path:
toolchain.save_cached_builtin_components([])
with patch.dict(os.environ, {"IDF_PATH": str(idf_path)}):
toolchain.save_cached_builtin_components(["lwip"])
assert not (idf_path / ".esphome_component_lists").exists()
def test_component_cache_write_failure_is_logged(
setup_core: Path, tmp_path: Path, caplog: pytest.LogCaptureFixture
) -> None:
_setup_build(setup_core)
with (
_cache_env(tmp_path, ""),
patch.object(toolchain, "write_file", side_effect=EsphomeError("disk full")),
):
toolchain.save_cached_builtin_components(["lwip"])
assert toolchain.load_cached_builtin_components() is None
assert "Could not write component list cache" in caplog.text
def test_component_cache_ignores_corrupt_file(setup_core: Path, tmp_path: Path) -> None:
_setup_build(setup_core)
with _cache_env(tmp_path, ""):
path = toolchain._builtin_component_cache_path()
path.parent.mkdir(parents=True)
path.write_text("{not json")
assert toolchain.load_cached_builtin_components() is None
path.write_text(json.dumps({"components": ["lwip"]}))
assert toolchain.load_cached_builtin_components() is None
def test_run_compile_passes_compile_process_limit(setup_core: Path) -> None:
"""compile_process_limit is forwarded to run_idf_py as the job limit."""
_setup_build(setup_core)
tests/unit_tests/test_framework_helpers.py
+98 -148
View File
@@ -2,6 +2,7 @@
# pylint: disable=protected-access
import gzip
import hashlib
import importlib.util
import io
@@ -31,6 +32,7 @@ from esphome.framework_helpers import (
_zip_extract_all,
archive_extract_all,
create_venv,
download_and_extract,
download_from_mirrors,
download_with_resume,
get_project_compile_flags,
@@ -1339,20 +1341,20 @@ class TestDownloadFromMirrors:
assert url == "https://example.com/f"
assert target.read_bytes() == b"filedata"
def test_file_object_target_reports_progress(self) -> None:
"""The library prefetch's production path: a file-object target
streams through the mirror fallback and ticks the tracker."""
buf = io.BytesIO()
def test_progress_callback_reports_bytes(self, tmp_path: Path) -> None:
"""The library prefetch's production path: the mirror download ticks
the caller's tracker instead of drawing its own bar."""
target = tmp_path / "f.bin"
ticks: list[int] = []
with patch(
"requests.get",
return_value=_mock_response(b"filedata"),
):
url = download_from_mirrors(
["https://example.com/f"], {}, buf, progress=ticks.append
["https://example.com/f"], {}, target, progress=ticks.append
)
assert url == "https://example.com/f"
assert buf.getvalue() == b"filedata"
assert target.read_bytes() == b"filedata"
assert ticks and ticks[-1] == len(b"filedata")
def test_substitutions_applied_to_url(self, tmp_path: Path) -> None:
@@ -1460,8 +1462,8 @@ class TestDownloadFromMirrors:
ei.value
)
def test_falls_back_to_second_mirror(self) -> None:
buf = io.BytesIO()
def test_falls_back_to_second_mirror(self, tmp_path: Path) -> None:
target = tmp_path / "f.bin"
with patch(
"requests.get",
side_effect=[_mock_response(b"", ok=False), _mock_response(b"second")],
@@ -1469,18 +1471,18 @@ class TestDownloadFromMirrors:
url = download_from_mirrors(
["https://mirror1.com/f", "https://mirror2.com/f"],
{},
buf,
target,
)
assert url == "https://mirror2.com/f"
assert buf.getvalue() == b"second"
assert target.read_bytes() == b"second"
def test_mid_stream_drop_resumes_same_mirror(self) -> None:
def test_mid_stream_drop_resumes_same_mirror(self, tmp_path: Path) -> None:
"""A mid-stream failure retries the same mirror with Range and
If-Range headers, keeping the bytes already received, before falling
to the next."""
first = _interrupted_response(b"1234", etag='"v1"')
first.headers = {**first.headers, "content-length": "8"}
buf = io.BytesIO()
target = tmp_path / "f.bin"
with patch(
"requests.get",
side_effect=[first, _resumed_response(b"5678")],
@@ -1488,10 +1490,10 @@ class TestDownloadFromMirrors:
url = download_from_mirrors(
["https://mirror1.com/f", "https://mirror2.com/f"],
{},
buf,
target,
)
assert url == "https://mirror1.com/f"
assert buf.getvalue() == b"12345678"
assert target.read_bytes() == b"12345678"
assert mock_get.call_count == 2
assert mock_get.call_args_list[1][0][0] == "https://mirror1.com/f"
# the resume is conditional on the content being unchanged
@@ -1500,48 +1502,6 @@ class TestDownloadFromMirrors:
"If-Range": '"v1"',
}
def test_mid_stream_drop_without_validator_restarts(self) -> None:
"""A server offering no ETag/Last-Modified cannot be resumed safely;
the retry restarts from zero instead of stitching unverified bytes."""
buf = io.BytesIO()
with patch(
"requests.get",
side_effect=[_interrupted_response(b"1234"), _mock_response(b"full")],
) as mock_get:
download_from_mirrors(["https://mirror1.com/f"], {}, buf)
assert buf.getvalue() == b"full"
assert "Range" not in mock_get.call_args_list[1][1]["headers"]
def test_drop_after_last_byte_recovers_via_416(self) -> None:
"""A connection drop after the final body byte leaves a complete file;
the retry's 416 answer plus the length check turn it into success
instead of a wasted refetch."""
first = _interrupted_response(b"1234", etag='"v1"')
first.headers = {**first.headers, "content-length": "4"}
r416 = _mock_response(b"", ok=False)
r416.status_code = 416
buf = io.BytesIO()
with patch("requests.get", side_effect=[first, r416]) as mock_get:
url = download_from_mirrors(["https://mirror1.com/f"], {}, buf)
assert url == "https://mirror1.com/f"
assert buf.getvalue() == b"1234"
assert mock_get.call_count == 2
def test_mirror_drop_without_length_restarts(self) -> None:
"""With no content-length there is no way to prove a stitched file
complete, so the retry restarts even though a validator exists."""
buf = io.BytesIO()
with patch(
"requests.get",
side_effect=[
_interrupted_response(b"1234", etag='"v1"'),
_mock_response(b"full"),
],
) as mock_get:
download_from_mirrors(["https://mirror1.com/f"], {}, buf)
assert buf.getvalue() == b"full"
assert "Range" not in mock_get.call_args_list[1][1]["headers"]
def test_path_target_resumes_across_runs(self, tmp_path: Path) -> None:
"""A path target routes through download_with_resume: a part file and
metadata from a previous run resume instead of restarting."""
@@ -1573,32 +1533,14 @@ class TestDownloadFromMirrors:
assert url == "https://mirror2.com/f"
assert dest.read_bytes() == b"data"
def test_resumed_short_body_fails_length_check(self) -> None:
"""A stitched file whose final length disagrees with the advertised
total is rejected instead of reported as success."""
first = _interrupted_response(b"1234", etag='"v1"')
first.headers = {**first.headers, "content-length": "8"}
# the resume ends early (5 of 8 bytes); the poisoned part is then
# discarded and the fresh retry also delivers a short body
short_resume = _resumed_response(b"5")
short_fresh = _mock_response(b"56")
short_fresh.headers = {**short_fresh.headers, "content-length": "8"}
buf = io.BytesIO()
with (
patch("requests.get", side_effect=[first, short_resume, short_fresh]),
pytest.raises(EsphomeError, match="all mirrors"),
):
download_from_mirrors(["https://mirror1.com/f"], {}, buf)
def test_failed_mirror_leftovers_not_kept_for_next_mirror(self) -> None:
"""Bytes from a mirror that failed all attempts must not leak into the
next mirror's download (no bogus Range request, fresh content)."""
exhausted = [_interrupted_response(b"AAAA", etag='"a1"')]
for _ in range(2):
r = _interrupted_response(b"BB")
r.status_code = 206
exhausted.append(r)
buf = io.BytesIO()
def test_failed_mirror_leftovers_not_resumed_on_next_mirror(
self, tmp_path: Path
) -> None:
"""A part file left by a mirror that failed all attempts must not be
stitched onto the next mirror's download (its meta names the other
URL, so the retry restarts from zero without a Range request)."""
exhausted = [_interrupted_response(b"AAAA") for _ in range(3)]
target = tmp_path / "f.bin"
with patch(
"requests.get",
side_effect=exhausted + [_mock_response(b"clean")],
@@ -1606,15 +1548,17 @@ class TestDownloadFromMirrors:
url = download_from_mirrors(
["https://mirror1.com/f", "https://mirror2.com/f"],
{},
buf,
target,
)
assert url == "https://mirror2.com/f"
assert buf.getvalue() == b"clean"
assert target.read_bytes() == b"clean"
# the second mirror starts fresh, without a Range header
assert mock_get.call_args_list[3][0][0] == "https://mirror2.com/f"
assert "Range" not in mock_get.call_args_list[3][1]["headers"]
def test_all_mirrors_fail_raises_error_listing_every_attempt(self) -> None:
def test_all_mirrors_fail_raises_error_listing_every_attempt(
self, tmp_path: Path
) -> None:
with (
patch(
"requests.get",
@@ -1625,7 +1569,7 @@ class TestDownloadFromMirrors:
download_from_mirrors(
["https://mirror1.com/f", "https://mirror2.com/f"],
{},
io.BytesIO(),
tmp_path / "out.bin",
)
# Every attempted URL appears in the message, and the first mirror's
# exception (the primary URL, usually the one that matters) is chained.
@@ -1641,16 +1585,6 @@ class TestDownloadFromMirrors:
with pytest.raises(TypeError, match="target must be"):
download_from_mirrors(["https://example.com/f"], {}, 42) # type: ignore[arg-type]
def test_file_like_target_written(self) -> None:
buf = io.BytesIO()
with patch(
"requests.get",
return_value=_mock_response(b"bytes"),
):
download_from_mirrors(["https://example.com/f"], {}, buf)
buf.seek(0)
assert buf.read() == b"bytes"
def test_progress_bar_shown_when_content_length_known(self, tmp_path: Path) -> None:
r = _mock_response(b"1234567890")
r.headers = {"content-length": "10"}
@@ -1676,13 +1610,10 @@ class TestDownloadFromMirrors:
assert target.exists()
assert target.read_bytes() == b""
@pytest.mark.parametrize("target_kind", ["path", "file-like"])
def test_transient_failure_retries_mirror_sweep(
self, tmp_path: Path, target_kind: str
) -> None:
def test_transient_failure_retries_mirror_sweep(self, tmp_path: Path) -> None:
"""A transient connect error on the only applicable mirror retries the
whole mirror list with backoff instead of failing the build."""
target = tmp_path / "idf.tar.xz" if target_kind == "path" else io.BytesIO()
target = tmp_path / "idf.tar.xz"
with (
patch(
"requests.get",
@@ -1695,33 +1626,10 @@ class TestDownloadFromMirrors:
):
url = download_from_mirrors(["https://mirror1.com/f"], {}, target)
assert url == "https://mirror1.com/f"
data = target.read_bytes() if target_kind == "path" else target.getvalue()
assert data == b"data"
assert target.read_bytes() == b"data"
assert mock_get.call_count == 2
mock_sleep.assert_called_once_with(2)
def test_backoff_tick_reports_filelike_bytes(self) -> None:
"""For a file-like target the backoff tick carries f.tell(), so the
combined bar holds steady through the sweep retry."""
target = io.BytesIO()
ticks: list[int] = []
with (
patch(
"requests.get",
side_effect=[
req.ConnectionError("down"),
_mock_response(b"data"),
],
),
patch("esphome.framework_helpers._cancellable_sleep") as mock_sleep,
):
download_from_mirrors(
["https://mirror1.com/f"], {}, target, progress=ticks.append
)
# No bytes had streamed at backoff time, so the tick carries 0
assert mock_sleep.call_args == call(2, ticks.append, 0)
assert target.getvalue() == b"data"
def test_backoff_tick_reports_partial_bytes(self, tmp_path: Path) -> None:
"""The backoff tick carries the bytes already in the part file, so a
combined bar holds steady instead of rewinding to zero."""
@@ -1831,41 +1739,83 @@ class TestDownloadFromMirrors:
assert isinstance(ei.value.__cause__, req.ConnectionError)
mock_sleep.assert_called_once_with(2)
def test_exhausted_mid_stream_attempts_not_swept(self) -> None:
"""A file-like mirror that spent all its mid-stream attempts is not
retried again at the sweep level (unlike a path target, it has no
part file to resume from on a later sweep)."""
buf = io.BytesIO()
def test_exhausted_mid_stream_attempts_not_swept(self, tmp_path: Path) -> None:
"""A mirror that spent all its mid-stream attempts fails permanently
instead of re-arming the sweep, and its part file survives so the
next esphome run resumes it."""
with (
patch(
"requests.get",
side_effect=[_interrupted_response(b"1234") for _ in range(3)],
) as mock_get,
patch("esphome.framework_helpers.time.sleep") as mock_sleep,
pytest.raises(EsphomeError, match="failed after 3 attempts"),
pytest.raises(EsphomeError, match="after 3 attempts"),
):
download_from_mirrors(["https://mirror1.com/f"], {}, buf)
download_from_mirrors(["https://mirror1.com/f"], {}, tmp_path / "out.bin")
assert mock_get.call_count == 3
mock_sleep.assert_not_called()
assert (tmp_path / "out.bin.part").exists()
class TestDownloadAndExtract:
def test_downloads_extracts_and_deletes_archive(self, tmp_path: Path) -> None:
content = gzip.compress(
_make_tar([_reg("file.txt")], {"file.txt": b"data"}).getvalue()
)
dest = tmp_path / "out"
with patch("requests.get", return_value=_mock_response(content)):
url = download_and_extract(
["https://example.com/lib.tar.gz"],
{},
tmp_path / "lib.archive",
dest,
)
assert url == "https://example.com/lib.tar.gz"
assert (dest / "file.txt").read_bytes() == b"data"
# the archive is consumed; only the extraction remains
assert not (tmp_path / "lib.archive").exists()
def test_locked_archive_does_not_mask_result(self, tmp_path: Path) -> None:
"""A cleanup unlink blocked by e.g. an AV handle (Windows) must not
replace the extraction result; the archive simply survives."""
content = gzip.compress(
_make_tar([_reg("file.txt")], {"file.txt": b"data"}).getvalue()
)
real_unlink = Path.unlink
def locked_unlink(self: Path, missing_ok: bool = False) -> None:
if self.name.endswith(".archive"):
raise PermissionError("held by antivirus")
real_unlink(self, missing_ok=missing_ok)
def test_mid_stream_drop_then_connect_error_not_swept(self) -> None:
"""A connect error on a later attempt (after a mid-stream drop spent
one) also counts as spent budget and does not re-arm the sweep."""
buf = io.BytesIO()
with (
patch(
"requests.get",
side_effect=[
_interrupted_response(b"1234"),
req.ConnectionError("down"),
],
) as mock_get,
patch("esphome.framework_helpers.time.sleep") as mock_sleep,
pytest.raises(EsphomeError, match="failed after 2 attempts"),
patch("requests.get", return_value=_mock_response(content)),
patch("pathlib.Path.unlink", locked_unlink),
):
download_from_mirrors(["https://mirror1.com/f"], {}, buf)
assert mock_get.call_count == 2
mock_sleep.assert_not_called()
url = download_and_extract(
["https://example.com/lib.tar.gz"],
{},
tmp_path / "lib.archive",
tmp_path / "out",
)
assert url == "https://example.com/lib.tar.gz"
assert (tmp_path / "out" / "file.txt").read_bytes() == b"data"
assert (tmp_path / "lib.archive").exists() # left behind, harmless
def test_corrupt_archive_deleted_on_extract_failure(self, tmp_path: Path) -> None:
"""A complete-but-corrupt archive must not survive to poison the next
run; without a checksum only a failed extraction can expose it."""
with (
patch("requests.get", return_value=_mock_response(b"not an archive")),
pytest.raises(ValueError, match="Unsupported archive format"),
):
download_and_extract(
["https://example.com/lib.tar.gz"],
{},
tmp_path / "lib.archive",
tmp_path / "out",
)
assert not (tmp_path / "lib.archive").exists()
def test_importing_framework_helpers_does_not_import_requests() -> None:
tests/unit_tests/test_happy_eyeballs.py
+37
View File
@@ -4,7 +4,9 @@ from __future__ import annotations
import asyncio
from collections.abc import Generator
from concurrent.futures import ThreadPoolExecutor
import socket
import threading
from typing import Any
from unittest.mock import Mock, patch
@@ -61,6 +63,41 @@ def test_ensure_happy_eyeballs_patches_and_is_idempotent(
assert urllib3.util.connection.create_connection is patched
def test_ensure_happy_eyeballs_concurrent_first_calls_patch_once(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""Worker threads fanning out (download_content_many, run_batch_downloads)
may race the first call; the replacement is built exactly once."""
import urllib3.util.connection
from esphome import happy_eyeballs
def stock(*args: Any, **kwargs: Any) -> None:
pass
monkeypatch.setattr(urllib3.util.connection, "create_connection", stock)
barrier = threading.Barrier(8)
builds: list[int] = []
real_make = happy_eyeballs._make_create_connection
def counting_make() -> Any:
builds.append(1)
return real_make()
monkeypatch.setattr(happy_eyeballs, "_make_create_connection", counting_make)
def racer() -> None:
barrier.wait(timeout=10)
ensure_happy_eyeballs()
with ThreadPoolExecutor(max_workers=8) as ex:
list(ex.map(lambda _: racer(), range(8)))
assert builds == [1]
assert urllib3.util.connection.create_connection._esphome_patched
def test_connects_and_restores_socket_state(
create_connection: Any, listener: tuple[str, int], mock_gai: Any
) -> None:
tests/unit_tests/test_net_retry.py
+42 -1
View File
@@ -7,7 +7,11 @@ import pytest
import requests as req
from esphome.core import EsphomeError
from esphome.net_retry import fetch_with_retry, is_transient_download_error
from esphome.net_retry import (
fetch_with_retry,
http_request,
is_transient_download_error,
)
def _http_error(status: int) -> req.HTTPError:
@@ -141,3 +145,40 @@ class TestFetchWithRetry:
assert mock_sleep.call_args_list == [call(2), call(4)]
assert "(attempt 2/3)" in caplog.text
assert "(attempt 3/3)" in caplog.text
class TestHttpRequest:
def test_applies_happy_eyeballs_and_forwards_arguments(self) -> None:
with (
patch("esphome.net_retry.ensure_happy_eyeballs") as mock_he,
patch("requests.get", return_value=MagicMock()) as mock_get,
):
resp = http_request(
"GET",
"https://example.com/f",
timeout=30,
stream=True,
headers={"Range": "bytes=4-"},
)
mock_he.assert_called_once_with()
assert resp is mock_get.return_value
assert mock_get.call_args == call(
"https://example.com/f",
timeout=30,
stream=True,
headers={"Range": "bytes=4-"},
allow_redirects=True,
)
def test_dispatches_head_through_requests_head(self) -> None:
"""Dispatch goes through requests.get/head so tests patching those
entry points keep working."""
with patch("requests.head", return_value=MagicMock()) as mock_head:
http_request("HEAD", "https://example.com/f", timeout=(5, 30))
assert mock_head.call_args[1]["timeout"] == (5, 30)
def test_no_status_handling(self) -> None:
"""Error statuses are the caller's problem; nothing raises here."""
resp = MagicMock(status_code=404)
with patch("requests.get", return_value=resp):
assert http_request("GET", "https://example.com/f", timeout=1) is resp
tests/unit_tests/test_nrf52_framework.py
+4 -2
View File
@@ -107,11 +107,13 @@ def mock_nrf52_ops():
patch(
"esphome.components.nrf52.framework.run_command_ok", return_value=True
) as mock_run_cmd,
# download_and_extract resolves its internals in framework_helpers,
# so the download/extract seams are patched there.
patch(
"esphome.components.nrf52.framework.download_from_mirrors",
"esphome.framework_helpers.download_from_mirrors",
return_value="https://example.com/tc.tar.xz",
) as mock_download,
patch("esphome.components.nrf52.framework.archive_extract_all") as mock_extract,
patch("esphome.framework_helpers.archive_extract_all") as mock_extract,
):
yield SimpleNamespace(
rmdir=mock_rmdir,
tests/unit_tests/test_platformio_library.py
+23 -8
View File
@@ -162,16 +162,12 @@ def test_urlsource_download_extracts_then_reuses_marker(
):
monkeypatch.setattr(lib, "rmdir", lambda path, msg="": None)
dl_calls: list[list[str]] = []
monkeypatch.setattr(
lib,
"download_from_mirrors",
lambda urls, headers, f, progress=None: dl_calls.append(urls),
)
def fake_extract(fileobj, path):
Path(path).mkdir(parents=True, exist_ok=True)
def fake_download_and_extract(urls, subs, archive_path, extract_dir, **kwargs):
dl_calls.append(urls)
Path(extract_dir).mkdir(parents=True, exist_ok=True)
monkeypatch.setattr(lib, "archive_extract_all", fake_extract)
monkeypatch.setattr(lib, "download_and_extract", fake_download_and_extract)
src = URLSource("http://example.test/lib.tar.gz")
out = src.download("mylib")
@@ -191,6 +187,25 @@ def test_urlsource_download_extracts_then_reuses_marker(
assert "Downloading" not in caplog.text
def test_urlsource_downloads_to_sibling_archive_path(setup_core, monkeypatch):
"""The archive downloads to a deterministic path next to the cache dir
(not a random temp file), so an interrupted download's .part file
resumes on the next run."""
monkeypatch.setattr(lib, "rmdir", lambda path, msg="": None)
targets: list[Path] = []
def fake_download_and_extract(urls, subs, archive_path, extract_dir, **kwargs):
targets.append(Path(archive_path))
Path(extract_dir).mkdir(parents=True, exist_ok=True)
monkeypatch.setattr(lib, "download_and_extract", fake_download_and_extract)
src = URLSource("http://example.test/lib.tar.gz")
out = src.download("mylib")
assert targets == [out.with_name(f"{out.name}.archive")]
def test_resolve_registry_version_raises_without_pkg_file(monkeypatch):
registry = lib._make_registry_client()
monkeypatch.setattr(
tests/unit_tests/test_platformio_registry.py
+62 -59
View File
@@ -6,7 +6,7 @@ from contextlib import contextmanager
import json
import os
from pathlib import Path
from unittest.mock import patch
from unittest.mock import MagicMock, patch
import pytest
@@ -32,12 +32,11 @@ def test_registry_download_resolves_once_per_process() -> None:
]
}
def fake_download(mirrors, substitutions, target):
calls.append(substitutions)
target.write(json.dumps(payload).encode())
return mirrors[0]
def fake_request(method, url, **kwargs):
calls.append(url)
return _http_response(json.dumps(payload))
with patch.object(registry, "download_from_mirrors", side_effect=fake_download):
with patch.object(registry, "http_request", side_effect=fake_request):
first = registry.registry_download("o/pkg", "1.0.0")
second = registry.registry_download("o/pkg", "1.0.0")
assert first == second
@@ -105,41 +104,47 @@ def test_get_systype_windows_empty_machine() -> None:
assert registry.get_systype() == "windows_amd64"
def _http_response(text: str) -> MagicMock:
resp = MagicMock()
resp.text = text
resp.raise_for_status.return_value = None
return resp
def _registry_response(files: list[dict]):
"""Patch the shared downloader to serve a canned registry response."""
"""Patch the consolidated HTTP path 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)
return patch.object(
registry, "http_request", return_value=_http_response(json.dumps(payload))
)
def test_registry_download_uses_shared_downloader() -> None:
"""The metadata fetch delegates its retries and error reporting to
download_from_mirrors; failures surface unchanged."""
def test_registry_download_uses_shared_http_path() -> None:
"""The metadata fetch delegates to the consolidated http_request path;
request failures surface as a named EsphomeError."""
import requests as req
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"),
"http_request",
side_effect=req.exceptions.ConnectionError("registry down"),
) as mock_request,
pytest.raises(EsphomeError, match="Could not fetch registry metadata"),
):
registry.registry_download("pkg", "1.0.0")
(mirrors, substitutions, _), _ = mock_download.call_args
assert mirrors == [registry._REGISTRY_URL]
assert substitutions == {"package": "pkg"}
(method, url), _ = mock_request.call_args
assert method == "GET"
assert url == registry._REGISTRY_URL.format(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),
patch.object(
registry,
"http_request",
return_value=_http_response("<html>not json</html>"),
),
pytest.raises(EsphomeError, match="invalid JSON"),
):
registry.registry_download("pkg", "1.0.0")
@@ -224,14 +229,14 @@ def test_registry_download_no_system_match() -> None:
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),
patch.object(
registry,
"http_request",
return_value=_http_response(
json.dumps({"versions": [{"name": "2.0.0", "files": []}]})
),
),
pytest.raises(EsphomeError, match="not found"),
):
registry.registry_download("pkg", "1.0.0")
@@ -391,12 +396,12 @@ def test_registry_download_empty_system_list_does_not_match() -> None:
def test_registry_download_unexpected_payload_is_named() -> None:
"""An error envelope without a versions list is not 'version not found'."""
def fake_download(mirrors: list[str], substitutions: dict, target) -> str:
target.write(json.dumps({"message": "rate limited"}).encode())
return "http://x"
with (
patch.object(registry, "download_from_mirrors", side_effect=fake_download),
patch.object(
registry,
"http_request",
return_value=_http_response(json.dumps({"message": "rate limited"})),
),
pytest.raises(EsphomeError, match="Unexpected package registry response"),
):
registry.registry_download("pkg", "1.0.0")
@@ -441,28 +446,26 @@ def test_registry_download_non_dict_version_entry_is_named() -> None:
"""A versions list of bare strings is an unexpected payload, not an
AttributeError traceback."""
def fake_download(mirrors: list[str], substitutions: dict, target) -> str:
target.write(json.dumps({"versions": ["1.0.0", "2.0.0"]}).encode())
return "http://x"
with (
patch.object(registry, "download_from_mirrors", side_effect=fake_download),
patch.object(
registry,
"http_request",
return_value=_http_response(json.dumps({"versions": ["1.0.0", "2.0.0"]})),
),
pytest.raises(EsphomeError, match="Unexpected package registry response"),
):
registry.registry_download("pkg", "1.0.0")
def test_registry_download_non_dict_file_entry_is_named() -> None:
def fake_download(mirrors: list[str], substitutions: dict, target) -> str:
target.write(
json.dumps(
{"versions": [{"name": "1.0.0", "files": ["a.tar.gz"]}]}
).encode()
)
return "http://x"
with (
patch.object(registry, "download_from_mirrors", side_effect=fake_download),
patch.object(
registry,
"http_request",
return_value=_http_response(
json.dumps({"versions": [{"name": "1.0.0", "files": ["a.tar.gz"]}]})
),
),
pytest.raises(EsphomeError, match="Unexpected package registry response"),
):
registry.registry_download("pkg", "1.0.0")
@@ -471,12 +474,12 @@ def test_registry_download_non_dict_file_entry_is_named() -> None:
def test_registry_download_non_dict_payload_is_named() -> None:
"""A JSON array answer is an unexpected payload at the outermost level."""
def fake_download(mirrors: list[str], substitutions: dict, target) -> str:
target.write(json.dumps(["1.0.0"]).encode())
return "http://x"
with (
patch.object(registry, "download_from_mirrors", side_effect=fake_download),
patch.object(
registry,
"http_request",
return_value=_http_response(json.dumps(["1.0.0"])),
),
pytest.raises(EsphomeError, match="Unexpected package registry response"),
):
registry.registry_download("pkg", "1.0.0")