Compare commits

..
Author SHA1 Message Date
J. Nick Koston e0385fae4a Merge remote-tracking branch 'origin/dev' into sscanf_stub
# Conflicts:
#	esphome/core/config.py
2026-10-21 18:49:23 -10:00
J. Nick Koston 966193cb09 [core] Emit generated lambda code after the user includes (#21307) 2026-10-21 18:28:04 -10:00
J. Nick Koston 163a0d388b Merge remote-tracking branch 'origin/dev' into sscanf_stub
# Conflicts:
#	tests/unit_tests/test_config_helpers.py
2026-10-21 16:39:59 -10:00
J. Nick Koston aa8e032081 Merge remote-tracking branch 'origin/dev' into sscanf_stub 2026-10-14 10:39:22 -10:00
J. Nick Koston efbee480dc Merge remote-tracking branch 'origin/dev' into sscanf_stub
# Conflicts:
#	esphome/components/esp32/__init__.py
#	esphome/core/defines.h
2026-10-09 16:08:43 -10:00
J. Nick Koston 1e72cdb8a0 Merge remote-tracking branch 'origin/dev' into sscanf_stub
# Conflicts:
#	esphome/writer.py
2026-10-09 13:09:48 -10:00
J. Nick Koston f117264eda Merge remote-tracking branch 'origin/dev' into sscanf_stub 2026-10-07 10:29:23 -10:00
J. Nick Koston 6ea7651a5c Merge remote-tracking branch 'origin/dev' into sscanf_stub 2026-10-06 21:09:14 -10:00
J. Nick Koston d253f6af4e Merge remote-tracking branch 'origin/dev' into sscanf_stub
# Conflicts:
#	esphome/core/config.py
2026-10-02 09:20:38 -05:00
J. Nick Koston bbfdf02903 Merge remote-tracking branch 'origin/dev' into sscanf_stub
# Conflicts:
#	esphome/core/config.py
2026-10-02 08:15:46 -05:00
J. Nick Koston 20ff795229 Merge remote-tracking branch 'origin/dev' into sscanf_stub 2026-10-01 15:23:04 -05:00
J. Nick Koston c9fbc2f046 Merge remote-tracking branch 'origin/dev' into sscanf_stub
# Conflicts:
#	esphome/components/esp8266/__init__.py
#	esphome/writer.py
2026-09-28 19:52:32 +02:00
J. Nick Koston 7c420ec6ae [core] Use a component key in the keep_float_scanf test 2026-09-26 01:14:48 +01:00
J. Nick Koston feb139fd20 [core] Pass the config through to the external scan and scan once 2026-09-26 01:14:11 +01:00
J. Nick Koston 5af8b8a972 Merge remote-tracking branch 'origin/dev' into sscanf_stub
# Conflicts:
#	esphome/writer.py
#	tests/component_tests/esp32/test_esp32.py
2026-09-26 00:12:06 +01:00
J. Nick Koston 39fd5b6a87 Merge remote-tracking branch 'upstream/dev' into HEAD 2026-09-17 08:48:11 -05:00
J. Nick Koston 1167d95312 Merge remote-tracking branch 'origin/dev' into sscanf_stub 2026-09-16 09:22:35 -05:00
J. Nick Koston 9828037ff1 [core] Let the scanf buffer argument contain a call with commas 2026-09-11 14:18:53 -05:00
J. Nick Koston c3546f8320 [core] Find the scanf format literal with one regex instead of a call walker 2026-09-11 14:03:01 -05:00
J. Nick Koston 9d400ab787 Revert "[core] Treat a non-literal scanf format as a float scan"
This reverts commit 4f0fec3e50.
2026-09-11 14:00:05 -05:00
J. Nick Koston 4f0fec3e50 [core] Treat a non-literal scanf format as a float scan 2026-09-11 13:57:20 -05:00
J. Nick Koston f73dc0e7ce [core] Move the scanf scan into code_scan.py on the loader's component list and trim the tests 2026-09-11 13:40:28 -05:00
J. Nick Koston cd214cc98b [core] Cover an external component without a float scanf 2026-09-11 13:28:37 -05:00
J. Nick Koston 3d71e9050f [core] Cover the esphome core domain in the external scan test 2026-09-11 13:22:23 -05:00
J. Nick Koston 55bf8c4e0c [core] Treat everything in the esphome package as in-tree for the float scanf scan 2026-09-11 13:21:54 -05:00
J. Nick Koston 7e089621d1 [core] Scan external component sources for float scanf 2026-09-11 12:59:13 -05:00
J. Nick Koston 1d1758c508 [core] Share the include extension set and cover the scanf call walker 2026-09-11 12:19:49 -05:00
J. Nick Koston 2606c88f3c [core] Match newlib for %%, support %p, and find float scans past a semicolon 2026-09-11 11:27:59 -05:00
J. Nick Koston 5c0607fc6e [esp32] Keep the sscanf wrap independent of enable_full_printf and say the override aborts 2026-09-11 10:33:23 -05:00
J. Nick Koston 2b3921b565 [core] Test include sorting and add_includes 2026-09-11 10:13:12 -05:00
J. Nick Koston 6020b76587 [core] Compare include basenames as posix in the test 2026-09-11 10:12:14 -05:00
J. Nick Koston 6f1d7d78f6 [core] Reuse SOURCE_FILE_EXTENSIONS for the include scan 2026-09-11 10:07:23 -05:00
J. Nick Koston aef04162d9 [esp32] Share include iteration and the float scanf resolver, register the sscanf job with the other wraps 2026-09-11 10:06:23 -05:00
J. Nick Koston e92530a0d5 [esp32] Scan includes for float scanf and name the format in the abort 2026-09-11 09:56:41 -05:00
J. Nick Koston 3c8c511e12 [esp32] Trim the sscanf wrap comments and docstrings 2026-09-11 09:53:38 -05:00
J. Nick Koston c6062a73c3 [esp32] Name the newlib wrap predicate for what it decides and test width with hh 2026-09-11 09:21:09 -05:00
J. Nick Koston bd193c0aff [core] Use fixed-width types in the sscanf scanner tests for clang-tidy 2026-09-11 08:45:48 -05:00
J. Nick Koston 79754065cd [esp32] Guard the sscanf wrap explicitly on Arduino and warn on a forced float lambda 2026-09-11 08:24:32 -05:00
J. Nick Koston 0e800b2bd8 [core] Mark the va_list parameter of the sscanf scanner for clang-tidy 2026-09-11 08:12:06 -05:00
J. Nick Koston 2c7b2e6cb1 [core] Use fixed-width types in the sscanf scanner for clang-tidy 2026-09-11 08:05:59 -05:00
J. Nick Koston ac4abd7f1f [core] Pass the sscanf destination pointer instead of a va_list reference 2026-09-11 08:01:06 -05:00
J. Nick Koston da70e10832 [core] Move the sscanf scanner into a header and add unit tests 2026-09-11 07:50:20 -05:00
J. Nick Koston 8aa1424ae4 [esp32] Share the wrap stub helpers and trim the sscanf scanner 2026-09-11 07:44:14 -05:00
J. Nick Koston 7d8f4b60e1 [esp32] Keep the libc sscanf when a lambda scans a float 2026-09-11 07:33:23 -05:00
J. Nick Koston d1943fe835 [esp32] Wrap sscanf to drop newlib's scanf engine from Bluetooth builds 2026-09-11 07:28:40 -05:00
35 changed files with 1474 additions and 1068 deletions
+2
View File
@@ -44,6 +44,7 @@ from esphome.platformio.library import (
normalize_dependencies,
parse_library_json,
parse_library_properties,
warn_properties_depends,
)
_LOGGER = logging.getLogger(__name__)
@@ -282,6 +283,7 @@ def _bundled_library(framework_path: Path, name: str) -> ArduinoLibrary:
"resolved automatically; add them with add_library() if needed",
name,
)
warn_properties_depends(name, data)
build = data.get("build")
if isinstance(build, dict) and build.get("extraScript"):
# Scripts only run on the converted path; building without
+126
View File
@@ -0,0 +1,126 @@
"""Build-time scans of the user's C++ sources: lambdas, ``includes:`` files and
external components.
The only scan today finds scanf calls with float conversions, which decides
whether float scanf support (or, on ESP-IDF Bluetooth builds, the libc sscanf)
must stay linked. Other source heuristics belong here too.
"""
import logging
from pathlib import Path
import re
from esphome.config_helpers import iter_include_files
from esphome.const import (
CONF_ESPHOME,
CONF_INCLUDES,
CONF_INCLUDES_C,
SOURCE_FILE_EXTENSIONS,
)
from esphome.core import Lambda
from esphome.types import ConfigType
_LOGGER = logging.getLogger(__name__)
# The string literal handed to a scanf family call as its format: an optional
# first argument (sscanf/fscanf; a literal or a call with its own commas is
# fine) then the literal.
_SCANF_FORMAT_RE = re.compile(
r'scanf\s*\((?:(?:"(?:[^"\\]|\\.)*"|\([^()]*\)|[^,;"()])*,)?\s*"((?:[^"\\]|\\.)*)"'
)
# Standard scanf float conversions %f %F %e %E %g %G %a %A with optional
# suppression, width and length; also the invalid %.2f users write by analogy
# with printf.
_SCANF_FLOAT_SPEC_RE = re.compile(r"%[*\d.]*[hlL]*[feEgGaAF]")
def source_uses_scanf_float(src: str) -> bool:
"""Heuristic: does C++ source call a scanf family function with a float conversion?"""
if "scanf" not in src:
return False
src = Lambda.comment_remover(src)
return any(
_SCANF_FLOAT_SPEC_RE.search(match.group(1))
for match in _SCANF_FORMAT_RE.finditer(src)
)
def lambdas_use_scanf_float(config: ConfigType) -> bool:
"""Check if any lambda in the config uses scanf with a float format specifier."""
stack: list = [config]
while stack:
obj = stack.pop()
if isinstance(obj, Lambda):
if source_uses_scanf_float(obj.value):
return True
elif isinstance(obj, dict):
stack.extend(obj.values())
elif isinstance(obj, list):
stack.extend(obj)
return False
def includes_use_scanf_float(config: ConfigType) -> bool:
"""Check if an ``includes:`` or ``includes_c:`` file uses scanf with a float format specifier."""
esphome_config = config.get(CONF_ESPHOME, {})
includes = esphome_config.get(CONF_INCLUDES, []) + esphome_config.get(
CONF_INCLUDES_C, []
)
for path, _ in iter_include_files(includes):
if path.suffix not in SOURCE_FILE_EXTENSIONS:
continue
if source_uses_scanf_float(path.read_text(encoding="utf-8", errors="replace")):
return True
return False
def external_components_use_scanf_float(config: ConfigType) -> bool:
"""Whether a component from outside the esphome package scans a float."""
from esphome.config import iter_components
from esphome.loader import CORE_COMPONENTS_PATH
package_root = CORE_COMPONENTS_PATH.parent
for _, component in iter_components(config):
package_dir = Path(component.module.__file__).resolve().parent
if package_dir.is_relative_to(package_root):
continue
for resource in component.resources:
src = (package_dir / resource.resource).read_text(
encoding="utf-8", errors="replace"
)
if source_uses_scanf_float(src):
return True
return False
def user_code_uses_scanf_float(config: ConfigType) -> bool:
"""Whether a lambda, an ``includes:`` file or an external component scans a float."""
return (
lambdas_use_scanf_float(config)
or includes_use_scanf_float(config)
or external_components_use_scanf_float(config)
)
def keep_float_scanf(
option: bool | None, config: ConfigType, flash_note: str, override_note: str
) -> bool:
"""Resolve a tri-state float-scanf option: unset means "only if user code scans a float"."""
if option:
return True
found = user_code_uses_scanf_float(config)
if option is None:
if found:
_LOGGER.warning(
"Lambda, include or external component uses scanf with a float format "
"specifier; keeping float scanf support (%s)",
flash_note,
)
return found
if found:
_LOGGER.warning(
"Float scanf support is disabled but a lambda, include or external "
"component uses scanf with a float format specifier; %s",
override_note,
)
return False
+53 -29
View File
@@ -10,11 +10,13 @@ import subprocess
from typing import Any
from esphome import yaml_util
from esphome.code_scan import keep_float_scanf
import esphome.codegen as cg
from esphome.components.const import (
CONF_ENABLE_OTA_DOWNGRADE_PROTECTION,
CONF_IGNORE_NOT_FOUND,
)
from esphome.config_helpers import filter_source_files_from_defines
import esphome.config_validation as cv
from esphome.const import (
CONF_ADVANCED,
@@ -204,6 +206,10 @@ PSRAM_XIP_VARIANTS = {
# neither, so the engine is already in the image and the wrap saves nothing.
ROM_VSNPRINTF_WITHOUT_VASPRINTF_VARIANTS = {VARIANT_ESP32C6}
# Variants whose ROM exports sscanf with the newlib format in use (C6 normal
# format, C2 nano format by IDF default), so the sscanf wrap saves nothing.
ROM_SSCANF_VARIANTS = {VARIANT_ESP32C2, VARIANT_ESP32C6}
# NVS encryption (HMAC peripheral scheme) is only available on variants that
# expose the HMAC peripheral (SOC_HMAC_SUPPORTED in soc_caps.h). The original
# ESP32 and ESP32-C2 do not have it. New variants with an HMAC peripheral
@@ -2052,6 +2058,7 @@ CONF_RINGBUF_IN_IRAM = "ringbuf_in_iram"
CONF_HEAP_IN_IRAM = "heap_in_iram"
CONF_LOOP_TASK_STACK_SIZE = "loop_task_stack_size"
CONF_USE_FULL_CERTIFICATE_BUNDLE = "use_full_certificate_bundle"
CONF_ENABLE_FULL_SCANF = "enable_full_scanf"
CONF_DISABLE_DEBUG_STUBS = "disable_debug_stubs"
CONF_DISABLE_OCD_AWARE = "disable_ocd_aware"
CONF_DISABLE_USB_SERIAL_JTAG_SECONDARY = "disable_usb_serial_jtag_secondary"
@@ -2347,6 +2354,7 @@ FRAMEWORK_SCHEMA = cv.Schema(
CONF_INCLUDE_BUILTIN_IDF_COMPONENTS, default=[]
): cv.ensure_list(cv.string_strict),
cv.Optional(CONF_ENABLE_FULL_PRINTF, default=False): cv.boolean,
cv.Optional(CONF_ENABLE_FULL_SCANF): cv.boolean,
cv.Optional(CONF_DISABLE_DEBUG_STUBS, default=True): cv.boolean,
cv.Optional(CONF_DISABLE_OCD_AWARE, default=True): cv.boolean,
cv.Optional(
@@ -2649,6 +2657,32 @@ async def _set_libc_picolibc_newlib_compat() -> None:
)
def _add_wrap_stub(symbol: str, define: str) -> None:
"""Emit a linker wrap for ``symbol`` served by a stub compiled under ``define``.
--undefined pulls the stub from libsrc.a, scanned before the IDF library that
references the symbol.
"""
cg.add_define(define)
cg.add_build_flag(f"-Wl,--wrap={symbol}")
cg.add_build_flag(f"-Wl,--undefined=__wrap_{symbol}")
@coroutine_with_priority(CoroPriority.FINAL)
async def _add_sscanf_stub(enable_full_scanf: bool | None) -> None:
"""Wrap sscanf when bluedroid is the only reason newlib's scanf engine links.
FINAL priority so every request_bluetooth() call has happened.
"""
if get_esp32_variant() in ROM_SSCANF_VARIANTS or not _network_sdkconfig().bluetooth:
return
if keep_float_scanf(
enable_full_scanf, CORE.config, "~13KB flash", "that call will abort the device"
):
return
_add_wrap_stub("sscanf", "USE_ESP32_SSCANF_STUB")
@coroutine_with_priority(CoroPriority.FINAL)
async def _apply_nvs_cache_in_psram(explicit: bool) -> None:
"""Keep the NVS cache in PSRAM unless NVS encryption is on, however it was enabled."""
@@ -3090,9 +3124,8 @@ async def to_code(config):
# saves nothing while costing ~170 B of shim. IDF 5.x defaults to
# newlib on every variant; IDF 6.0+ switches to picolibc on every
# variant.
if conf[CONF_ADVANCED][CONF_ENABLE_FULL_PRINTF] or idf_version() >= cv.Version(
6, 0, 0
):
uses_newlib = idf_version() < cv.Version(6, 0, 0)
if conf[CONF_ADVANCED][CONF_ENABLE_FULL_PRINTF] or not uses_newlib:
cg.add_define("USE_FULL_PRINTF")
else:
for symbol in ("vprintf", "printf", "fprintf", "vfprintf"):
@@ -3100,14 +3133,14 @@ async def to_code(config):
# esp_http_client calls vasprintf, which on the ESP32-C6 is the only
# reference to newlib's full printf engine (~20 KB: _svfprintf_r,
# _dtoa_r and their helpers); every other caller resolves to the
# ROM. See vasprintf_stubs.cpp. The --undefined flag is needed
# because libsrc.a is scanned before the IDF libraries that
# reference the symbol, so the stub would otherwise never be pulled
# from the archive.
# ROM. See vasprintf_stubs.cpp.
if variant in ROM_VSNPRINTF_WITHOUT_VASPRINTF_VARIANTS:
cg.add_define("USE_ESP32_VASPRINTF_STUB")
cg.add_build_flag("-Wl,--wrap=vasprintf")
cg.add_build_flag("-Wl,--undefined=__wrap_vasprintf")
_add_wrap_stub("vasprintf", "USE_ESP32_VASPRINTF_STUB")
if uses_newlib:
# bluedroid's sscanf calls; see sscanf_stubs.cpp
CORE.add_job(
_add_sscanf_stub, conf[CONF_ADVANCED].get(CONF_ENABLE_FULL_SCANF)
)
else:
cg.add_build_flag("-DUSE_ARDUINO")
cg.add_build_flag("-DUSE_ESP32_FRAMEWORK_ARDUINO")
@@ -3813,20 +3846,6 @@ def _write_sdkconfig():
clean_build(clear_pio_cache=False)
def _arduino_provides(name: str) -> bool:
"""Whether arduino-esp32 bundles the library a manifest depends on, so it is
never fetched by name; it still has to be enabled to build."""
if name not in ARDUINO_DISABLED_LIBRARIES:
return False
if name not in CORE.data[KEY_ESP32][KEY_ARDUINO_LIBRARIES]:
_LOGGER.warning(
"A library depends on the Arduino library %s, which is not enabled; "
"add it under esphome: libraries:",
name,
)
return True
def _write_idf_component_yml():
yml_path = CORE.relative_build_path("src/idf_component.yml")
dependencies: dict[str, dict] = {}
@@ -3848,11 +3867,7 @@ def _write_idf_component_yml():
# IDF sees the same requirement twice; converted components reach it through
# ${ESPHOME_PROJECT_MANAGED_COMPONENTS}.
managed = set(CORE.data[KEY_ESP32].get(KEY_COMPONENTS, {}))
converted = generate_idf_components(
libraries,
managed=managed,
provides=_arduino_provides if CORE.using_arduino else None,
)
converted = generate_idf_components(libraries, managed=managed)
# IDF names a component after its directory and a later registration of the
# same name replaces the earlier one, so a stub beside a converted library of
# the same name (espressif/libsodium vs esphome/libsodium) would win or lose
@@ -4115,3 +4130,12 @@ def process_stacktrace(config, line, backtrace_state):
_decode_pc(config, addr.group())
return backtrace_state
# Each stub only compiles under its define, so keep the sources out of other builds
FILTER_SOURCE_FILES = filter_source_files_from_defines(
{
"sscanf_stubs.cpp": "USE_ESP32_SSCANF_STUB",
"vasprintf_stubs.cpp": "USE_ESP32_VASPRINTF_STUB",
}
)
+43
View File
@@ -0,0 +1,43 @@
/*
* Linker wrap for sscanf on ESP-IDF with newlib.
*
* bluedroid's two "%02x" parses are the only sscanf callers in an ESPHome
* image and link newlib's whole scanf engine (~13 KB with _strtod_l). Routing
* them through core/sscanf_no_float.h drops it; floats would pull _strtod_l
* back, so those abort naming enable_full_scanf. See _add_sscanf_stub in
* esp32/__init__.py for when this is emitted.
*/
#include "esphome/core/defines.h"
#if defined(USE_ESP_IDF) && defined(USE_ESP32_SSCANF_STUB)
#include <cstdarg>
#include <cstdio>
#include "esp_system.h"
#include "esphome/core/sscanf_no_float.h"
namespace esphome::esp32 {}
// NOLINTBEGIN(bugprone-reserved-identifier,cert-dcl37-c,cert-dcl51-cpp,readability-identifier-naming)
extern "C" {
int __wrap_sscanf(const char *str, const char *fmt, ...) {
va_list ap;
va_start(ap, fmt);
int result = esphome::vsscanf_no_float(str, fmt, ap);
va_end(ap);
if (result == esphome::SSCANF_UNSUPPORTED) {
char msg[160];
snprintf(msg, sizeof(msg),
"sscanf: unsupported conversion in \"%s\"; set enable_full_scanf: true in esp32 framework", fmt);
esp_system_abort(msg);
}
return result;
}
} // extern "C"
// NOLINTEND(bugprone-reserved-identifier,cert-dcl37-c,cert-dcl51-cpp,readability-identifier-naming)
#endif // USE_ESP_IDF && USE_ESP32_SSCANF_STUB
+8 -43
View File
@@ -6,6 +6,7 @@ import subprocess
from typing import Any
from esphome.build_helpers.native import native_backend
from esphome.code_scan import keep_float_scanf
import esphome.codegen as cg
from esphome.components import network
import esphome.config_validation as cv
@@ -26,13 +27,7 @@ from esphome.const import (
ThreadModel,
Toolchain,
)
from esphome.core import (
CORE,
CoroPriority,
EsphomeError,
Lambda,
coroutine_with_priority,
)
from esphome.core import CORE, CoroPriority, EsphomeError, coroutine_with_priority
from esphome.core.config import BOARD_MAX_LENGTH
from esphome.helpers import IS_MACOS, copy_file_if_changed
from esphome.platformio.toolchain import copy_ccache_script
@@ -67,15 +62,6 @@ from .const import (
from .gpio import PinInitialState, add_pin_initial_states_array
CONF_ENABLE_SCANF_FLOAT = "enable_scanf_float"
# Heuristically matches scanf/sscanf calls with float format specifiers.
# Standard scanf float conversions: %f %F %e %E %g %G %a %A
# With optional modifiers: %*f (suppression), %8f (width), %lf %Lf (length)
# Also matches non-standard patterns like %.2f as a heuristic — these are
# invalid in scanf but users may write them by analogy with printf.
# Uses [^;]*? to stay within a single statement, preventing false positives
# from e.g. sscanf(buf, "%d", &x); printf("%f", val);
_SCANF_FLOAT_RE = re.compile(r"scanf\s*\([^;]*?%[*\d.]*[hlL]*[feEgGaAF]")
CODEOWNERS = ["@esphome/core"]
DOMAIN = "esp8266"
_LOGGER = logging.getLogger(__name__)
@@ -83,26 +69,6 @@ AUTO_LOAD = ["preferences"]
IS_TARGET_PLATFORM = True
def lambdas_use_scanf_float(config: ConfigType) -> bool:
"""Check if any lambda in the config uses scanf with a float format specifier.
Comments are stripped before matching to avoid false positives from
commented-out code. The cost of a false positive is only ~8KB flash.
"""
stack: list = [config]
while stack:
obj = stack.pop()
if isinstance(obj, Lambda):
src = obj.comment_remover(obj.value)
if _SCANF_FLOAT_RE.search(src):
return True
elif isinstance(obj, dict):
stack.extend(obj.values())
elif isinstance(obj, list):
stack.extend(obj)
return False
def set_core_data(config: ConfigType) -> ConfigType:
CORE.data[KEY_ESP8266] = {}
CORE.data[KEY_CORE][KEY_TARGET_PLATFORM] = PLATFORM_ESP8266
@@ -425,13 +391,12 @@ async def to_code(config: ConfigType) -> None:
cg.add_define("USE_ESP8266_CRASH_HANDLER")
cg.add_define("USE_CRASH_HANDLER")
enable_scanf_float = config.get(CONF_ENABLE_SCANF_FLOAT)
if enable_scanf_float is None and lambdas_use_scanf_float(CORE.config):
enable_scanf_float = True
_LOGGER.warning(
"Lambda uses scanf with a float format specifier; "
"enabling scanf float support (~8KB flash)"
)
enable_scanf_float = keep_float_scanf(
config.get(CONF_ENABLE_SCANF_FLOAT),
CORE.config,
"~8KB flash",
"that call will not parse",
)
# The native generator reads the same decision (KEY_SCANF_FLOAT)
CORE.data[KEY_ESP8266][KEY_SCANF_FLOAT] = bool(enable_scanf_float)
+19 -1
View File
@@ -1,4 +1,6 @@
from collections.abc import Callable, Collection
from collections.abc import Callable, Collection, Iterator
import logging
from pathlib import Path
from esphome.const import (
CONF_LEVEL,
@@ -9,8 +11,11 @@ from esphome.const import (
PlatformFramework,
)
from esphome.core import CORE
from esphome.helpers import walk_files
from esphome.util import OrderedDict
_LOGGER = logging.getLogger(__name__)
# Pre-build lookup map from (platform, framework) tuples to PlatformFramework enum
_PLATFORM_FRAMEWORK_LOOKUP = {
(pf.value[0].value, pf.value[1].value): pf for pf in PlatformFramework
@@ -191,3 +196,16 @@ def get_logger_level() -> str:
logger_config = CORE.config[CONF_LOGGER]
return logger_config.get(CONF_LEVEL, "DEBUG")
def iter_include_files(includes: list[str]) -> Iterator[tuple[Path, Path]]:
"""Yield ``(path, basename)`` per local ``includes:`` file; directory members keep the directory prefix."""
for include in includes:
if include.startswith("<") and include.endswith(">"):
continue
path = CORE.relative_config_path(include)
if path.is_dir():
for file in walk_files(path):
yield file, file.relative_to(path.parent)
else:
yield path, Path(path.name)
+18 -11
View File
@@ -658,6 +658,8 @@ class EsphomeCore:
self.main_statements: list[Statement] = []
# A list of statements to insert in the global block (includes and global variables)
self.global_statements: list[Statement] = []
# Generated code emitted after the user includes, so it can use what they declare
self.global_statements_after_includes: list[Statement] = []
# A map of platformio libraries to add to the project (shortname: (name, version, repository))
self.platformio_libraries: dict[str, Library] = {}
# A set of build flags to set in the platformio project
@@ -728,6 +730,7 @@ class EsphomeCore:
self.variables = {}
self.main_statements = []
self.global_statements = []
self.global_statements_after_includes = []
self.platformio_libraries = {}
self.build_flags = set()
self.cmake_args = {}
@@ -1045,8 +1048,8 @@ class EsphomeCore:
except RuntimeError as e:
raise EsphomeError(str(e)) from e
def add(self, expression, prepend=False) -> "Statement":
"""Add an expression or statement to the main setup() block."""
@staticmethod
def _as_statement(expression) -> "Statement":
from esphome.cpp_generator import Expression, Statement, statement
if isinstance(expression, Expression):
@@ -1055,7 +1058,11 @@ class EsphomeCore:
raise ValueError(
f"Add '{expression}' must be expression or statement, not {type(expression)}"
)
return expression
def add(self, expression, prepend=False) -> "Statement":
"""Add an expression or statement to the main setup() block."""
expression = self._as_statement(expression)
if prepend:
self.main_statements.insert(0, expression)
else:
@@ -1064,14 +1071,7 @@ class EsphomeCore:
return expression
def add_global(self, expression, prepend=False) -> "Statement":
from esphome.cpp_generator import Expression, Statement, statement
if isinstance(expression, Expression):
expression = statement(expression)
if not isinstance(expression, Statement):
raise ValueError(
f"Add '{expression}' must be expression or statement, not {type(expression)}"
)
expression = self._as_statement(expression)
if prepend:
self.global_statements.insert(0, expression)
else:
@@ -1079,6 +1079,13 @@ class EsphomeCore:
_LOGGER.debug("Adding global: %s", expression)
return expression
def add_global_after_includes(self, expression) -> "Statement":
"""Add a global after the user includes, for generated code that may use them."""
expression = self._as_statement(expression)
self.global_statements_after_includes.append(expression)
_LOGGER.debug("Adding global after includes: %s", expression)
return expression
def add_library(self, library: Library):
if not isinstance(library, Library):
raise TypeError(
@@ -1258,7 +1265,7 @@ class EsphomeCore:
from esphome.cpp_generator import statement
global_code = []
for exp in self.global_statements:
for exp in (*self.global_statements, *self.global_statements_after_includes):
text = str(statement(exp))
text = text.rstrip()
global_code.append(text)
+7 -14
View File
@@ -10,6 +10,7 @@ import esphome.codegen as cg
from esphome.config_helpers import (
filter_source_files_from_defines,
filter_source_files_from_platform,
iter_include_files,
)
import esphome.config_validation as cv
from esphome.const import (
@@ -48,6 +49,7 @@ from esphome.const import (
PLATFORM_ESP32,
PLATFORM_HOST,
PLATFORM_RP2,
SOURCE_FILE_EXTENSIONS,
PlatformFramework,
__version__ as ESPHOME_VERSION,
)
@@ -64,7 +66,6 @@ from esphome.helpers import (
fnv1a_32bit_hash,
get_str_env,
get_usable_cpu_count,
walk_files,
)
from esphome.types import ConfigType
@@ -132,7 +133,7 @@ SHARED_DEVICE_RELATION_DEFINES: dict[str, str | None] = {
}
Area = cg.esphome_ns.class_("Area")
VALID_INCLUDE_EXTS = {".h", ".hpp", ".tcc", ".ino", ".cpp", ".c"}
VALID_INCLUDE_EXTS = SOURCE_FILE_EXTENSIONS
def validate_hostname(config):
@@ -590,18 +591,10 @@ async def add_arduino_global_workaround():
@coroutine_with_priority(CoroPriority.FINAL)
async def add_includes(includes: list[str], is_c_header: bool = False) -> None:
# Add includes at the very end, so that the included files can access global variables
for include in includes:
path = CORE.relative_config_path(include)
if path.is_dir():
# Directory, copy tree
for p in walk_files(path):
basename = p.relative_to(path.parent)
include_file(p, basename, is_c_header)
else:
# Copy file
basename = Path(path.name)
include_file(path, basename, is_c_header)
# After the id declarations, so the headers can use id(x); the generated function bodies
# and flash tables come later still, so their lambdas can use the headers.
for path, basename in iter_include_files(includes):
include_file(path, basename, is_c_header)
def _add_library_str(lib: str) -> None:
+1
View File
@@ -434,6 +434,7 @@
#ifdef USE_ESP32
#define USE_ESP32_CRASH_HANDLER
#define USE_ESP32_VASPRINTF_STUB
#define USE_ESP32_SSCANF_STUB
#define USE_MQTT_IDF_ENQUEUE
#define USE_MQTT_TLS
#define USE_MQTT_CERT_BUNDLE
+277
View File
@@ -0,0 +1,277 @@
#pragma once
// sscanf without floating-point conversions, backing the linker wrap in
// components/esp32/sscanf_stubs.cpp. Integer, character, string and scanset
// conversions behave like libc; anything else returns SSCANF_UNSUPPORTED.
// Header only so the host unit tests can cover it.
#include <cstdarg>
#include <cstddef>
#include <cstdint>
#include <cstdio>
#include "esphome/core/helpers.h"
namespace esphome {
/// Returned by vsscanf_no_float() for a conversion it does not implement.
static constexpr int SSCANF_UNSUPPORTED = -2;
namespace sscanf_no_float_internal {
enum class SscanfLength : uint8_t {
SSCANF_LENGTH_NONE,
SSCANF_LENGTH_HH,
SSCANF_LENGTH_H,
SSCANF_LENGTH_L,
SSCANF_LENGTH_LL,
SSCANF_LENGTH_Z,
SSCANF_LENGTH_T,
SSCANF_LENGTH_PTR, // %p
};
inline bool is_space(char c) { return c == ' ' || (c >= '\t' && c <= '\r'); }
// The unsigned bit pattern is what the signed conversions want as well
inline void store_int(void *dest, SscanfLength length, uint64_t value) {
switch (length) {
case SscanfLength::SSCANF_LENGTH_HH:
*static_cast<uint8_t *>(dest) = static_cast<uint8_t>(value);
break;
case SscanfLength::SSCANF_LENGTH_H:
*static_cast<uint16_t *>(dest) = static_cast<uint16_t>(value);
break;
case SscanfLength::SSCANF_LENGTH_L:
// %l is defined in terms of long, whatever its width on the target
*static_cast<unsigned long *>(dest) = static_cast<unsigned long>(value); // NOLINT(google-runtime-int)
break;
case SscanfLength::SSCANF_LENGTH_LL:
*static_cast<uint64_t *>(dest) = value;
break;
case SscanfLength::SSCANF_LENGTH_Z:
*static_cast<size_t *>(dest) = static_cast<size_t>(value);
break;
case SscanfLength::SSCANF_LENGTH_T:
*static_cast<ptrdiff_t *>(dest) = static_cast<ptrdiff_t>(value);
break;
case SscanfLength::SSCANF_LENGTH_PTR:
*static_cast<uintptr_t *>(dest) = static_cast<uintptr_t>(value);
break;
default:
*static_cast<unsigned *>(dest) = static_cast<unsigned>(value);
break;
}
}
struct Scanset {
uint8_t bits[32]{};
bool negate{false};
bool contains(char c) const {
auto u = static_cast<uint8_t>(c);
return ((this->bits[u >> 3] >> (u & 7)) & 1u) != this->negate;
}
};
// Parses "[set]" from just after the '[' and returns the position after ']'
inline const char *parse_scanset(const char *f, Scanset &set) {
if (*f == '^') {
set.negate = true;
f++;
}
// A ']' right after '[' or '[^' is a member, not the terminator.
bool first = true;
for (; *f != '\0' && (first || *f != ']'); f++) {
first = false;
auto lo = static_cast<uint8_t>(*f);
auto hi = lo;
if (f[1] == '-' && f[2] != ']' && f[2] != '\0') {
hi = static_cast<uint8_t>(f[2]);
f += 2;
}
for (unsigned c = lo; c <= hi; c++)
set.bits[c >> 3] |= static_cast<uint8_t>(1u << (c & 7));
}
return *f == ']' ? f + 1 : f;
}
// Radix of an integer conversion: 0 detects from the prefix (%i), 0xFF is unsupported
inline unsigned int_base(char conv) {
switch (conv) {
case 'd':
case 'u':
return 10;
case 'i':
return 0;
case 'o':
return 8;
case 'x':
case 'X':
case 'p':
return 16;
default:
return 0xFF;
}
}
} // namespace sscanf_no_float_internal
/// sscanf semantics for the integer, character, string and scanset
/// conversions; returns SSCANF_UNSUPPORTED on any other conversion.
inline int vsscanf_no_float(const char *str, const char *fmt, va_list ap) { // NOLINT(readability-non-const-parameter)
using namespace sscanf_no_float_internal;
const char *in = str;
int assigned = 0;
bool input_failure = false;
for (const char *f = fmt; *f != '\0'; f++) {
if (is_space(*f)) {
while (is_space(*in))
in++;
continue;
}
bool literal = *f != '%';
if (!literal && f[1] == '%') {
// "%%" matches one '%' with no whitespace skip, like newlib
f++;
literal = true;
}
if (literal) {
if (*in != *f) {
input_failure = *in == '\0';
break;
}
in++;
continue;
}
f++;
bool suppress = false;
if (*f == '*') {
suppress = true;
f++;
}
size_t width = 0;
while (*f >= '0' && *f <= '9') {
width = width * 10 + static_cast<size_t>(*f - '0');
f++;
}
if (width == 0)
width = SIZE_MAX;
auto length = SscanfLength::SSCANF_LENGTH_NONE;
switch (*f) {
case 'h':
f++;
if (*f == 'h') {
length = SscanfLength::SSCANF_LENGTH_HH;
f++;
} else {
length = SscanfLength::SSCANF_LENGTH_H;
}
break;
case 'l':
f++;
if (*f == 'l') {
length = SscanfLength::SSCANF_LENGTH_LL;
f++;
} else {
length = SscanfLength::SSCANF_LENGTH_L;
}
break;
case 'j':
length = SscanfLength::SSCANF_LENGTH_LL;
f++;
break;
case 'z':
length = SscanfLength::SSCANF_LENGTH_Z;
f++;
break;
case 't':
length = SscanfLength::SSCANF_LENGTH_T;
f++;
break;
default:
break;
}
const char conv = *f;
if (conv == '\0')
break;
if (conv == 'n') {
if (!suppress)
store_int(va_arg(ap, void *), length, static_cast<uint64_t>(in - str));
continue;
}
const bool string_conv = conv == 'c' || conv == 's' || conv == '[';
if (string_conv && length != SscanfLength::SSCANF_LENGTH_NONE)
return SSCANF_UNSUPPORTED; // wide characters
if (conv == 'p')
length = SscanfLength::SSCANF_LENGTH_PTR;
unsigned base = string_conv ? 0 : int_base(conv);
if (base == 0xFF)
return SSCANF_UNSUPPORTED;
if (conv != 'c' && conv != '[') {
while (is_space(*in))
in++;
}
if (*in == '\0') {
input_failure = true;
break;
}
if (!string_conv) {
const char *p = in;
size_t n = 0;
bool negative = false;
if (*p == '+' || *p == '-') {
negative = *p == '-';
p++;
n++;
}
if ((base == 0 || base == 16) && p[0] == '0' && (p[1] == 'x' || p[1] == 'X') && n + 2 < width &&
parse_hex_char(p[2]) != INVALID_HEX_CHAR) {
p += 2;
n += 2;
base = 16;
}
if (base == 0)
base = *p == '0' ? 8 : 10;
uint64_t value = 0;
bool any_digit = false;
for (uint8_t digit; n < width && (digit = parse_hex_char(*p)) < base; p++, n++) {
value = value * base + digit;
any_digit = true;
}
if (!any_digit)
break;
in = p;
if (!suppress) {
store_int(va_arg(ap, void *), length, negative ? 0 - value : value);
assigned++;
}
continue;
}
char *dest = suppress ? nullptr : va_arg(ap, char *);
Scanset set;
if (conv == '[')
f = parse_scanset(f + 1, set) - 1; // the loop's f++ steps past the ']'
// %c reads width characters (default 1); a short read still assigns, like newlib
const size_t limit = conv == 'c' && width == SIZE_MAX ? 1 : width;
size_t n = 0;
while (n < limit && in[n] != '\0' && (conv == 'c' || (conv == '[' ? set.contains(in[n]) : !is_space(in[n])))) {
if (dest != nullptr)
dest[n] = in[n];
n++;
}
if (conv != 'c') {
if (n == 0)
break;
if (dest != nullptr)
dest[n] = '\0';
}
in += n;
if (!suppress)
assigned++;
}
if (input_failure && assigned == 0)
return EOF;
return assigned;
}
} // namespace esphome
+20 -10
View File
@@ -536,8 +536,10 @@ def shared_progmem_array(
used |= {str(i) for i in CORE.variables}
used |= set(RESERVED_IDS) | CORE.loaded_integrations
id_ = ID(ensure_unique_string(name, used), is_declaration=True, type=type_)
# Global, so any scope can use it; anything a lambda references is already declared.
CORE.add_global(ProgmemAssignmentExpression(type_, id_, rhs, constexpr))
# After the user includes, so a lambda in the table can use what their headers declare.
CORE.add_global_after_includes(
ProgmemAssignmentExpression(type_, id_, rhs, constexpr)
)
array = MockObj(id_, ".")
CORE.register_variable(id_, array)
if share:
@@ -789,15 +791,16 @@ def static_function(
"""Emit ``static <return_type> <name>(parameters) { body }`` at global scope and return an
expression naming it, for use as a template argument or a function pointer.
Every id the body names must already be declared, which holds when the statements were
rendered through ``get_variable`` or ``process_lambda``.
The declaration goes with the other globals, so a template argument can name the function
at once; the body goes after the user includes, where every id and everything those
headers declare is in scope.
"""
params = ParameterListExpression(*parameters)
add_global(
RawStatement(
f"static {safe_exp(return_type)} {name}({params}) {{\n"
f"{indent(chr(10).join(body))}\n}}"
)
signature = (
f"static {safe_exp(return_type)} {name}({ParameterListExpression(*parameters)})"
)
add_global(RawStatement(f"{signature};"))
add_global_after_includes(
RawStatement(f"{signature} {{\n{indent(chr(10).join(body))}\n}}")
)
return RawExpression(name)
@@ -816,6 +819,13 @@ def add_global(expression: SafeExpType | Statement, prepend: bool = False):
CORE.add_global(expression, prepend)
def add_global_after_includes(expression: SafeExpType | Statement):
"""Add a global after the user's ``includes:`` headers, for generated code whose
lambdas may use what those headers declare.
"""
CORE.add_global_after_includes(expression)
def add_library(name: str, version: str | None, repository: str | None = None):
"""Add a library to the codegen library storage.
+2 -12
View File
@@ -266,24 +266,14 @@ def _convert_pio_libs(
in the real esp32 build; converting them too would give IDF the same requirement
twice. Arduino below IDF 6.0 has them rule-gated off, so there they convert.
"""
from esphome.components.esp32 import (
ARDUINO_DISABLED_LIBRARIES,
arduino_bundles_libsodium,
)
from esphome.components.esp32 import arduino_bundles_libsodium
from esphome.espidf.component import generate_idf_components
libraries = _parse_lib_deps(platformio_ini, framework)
# Same rule as noise._use_managed_components and the manifest
managed = set() if arduino_bundles_libsodium() else _esphome_manifest_deps()
deps: dict[str, dict[str, str]] = {}
# As in the real esp32 build, a dependency on an Arduino-bundled library is
# left to arduino-esp32 instead of being fetched from the registry
provides = (
ARDUINO_DISABLED_LIBRARIES.__contains__ if framework == "arduino" else None
)
for component in generate_idf_components(
libraries, managed=managed, provides=provides
):
for component in generate_idf_components(libraries, managed=managed):
deps[component.get_sanitized_name()] = {"override_path": str(component.path)}
return deps
+1 -7
View File
@@ -7,7 +7,6 @@ for each resolved library, running any PlatformIO ``extraScript``, and the
ESP-IDF platform/framework compatibility defaults.
"""
from collections.abc import Callable
import logging
import os
from pathlib import Path
@@ -289,23 +288,18 @@ def _emit_idf_component(component: IDFComponent) -> None:
def generate_idf_components(
libraries: list[Library],
managed: set[str] | None = None,
provides: Callable[[str], bool] | None = None,
libraries: list[Library], managed: set[str] | None = None
) -> list[IDFComponent]:
"""Resolve and convert a batch of PlatformIO libraries to IDF components.
``managed`` names registry components already in the project manifest; they
are not converted, since IDF rejects the same requirement from two components.
Converted components reach them through ``${ESPHOME_PROJECT_MANAGED_COMPONENTS}``.
``provides`` names the libraries the framework bundles, so a manifest dependency on
one is never fetched from the registry.
"""
backend = LibraryBackend(
platform=ESP32_PLATFORM,
framework=_idf_framework(),
emit=_emit_idf_component,
cache_key="idf",
provides=provides,
)
return convert_libraries(libraries, backend, provided=managed)
+101 -236
View File
@@ -609,34 +609,9 @@ def parse_library_properties(library_properties_path: PathType):
if not value.strip():
continue
data[key.strip()] = value.strip()
if depends := data.pop("depends", None):
data["dependencies"] = parse_properties_depends(depends)
return data
def parse_properties_depends(depends: str) -> list[dict[str, str]]:
"""The Arduino ``depends=Name, Other (>=1.0)`` line as name-only dependency entries.
An Arduino constraint is not PlatformIO syntax, so it is dropped; the name is
satisfied by a listed library or the registry's current version, as the walk
does for every version-less dependency.
"""
entries: list[dict[str, str]] = []
for item in depends.split(","):
name, _, constraint = item.strip().partition("(")
if not (name := name.strip()):
continue
if constraint:
# A plain lower bound is met by the registry's current version; an
# exact or capped one may not be, so say so
log = (
_LOGGER.debug if constraint.lstrip().startswith(">=") else _LOGGER.info
)
log("Ignoring version constraint on depends=%s", item.strip())
entries.append({"name": name})
return entries
def _make_registry_client() -> Any:
"""Create a minimal PlatformIO registry client with no system filtering.
@@ -755,6 +730,20 @@ def join_flag_args(tokens: Iterable[str], owner: str) -> list[str]:
return out
def warn_properties_depends(name: str, data: object) -> None:
"""Warn for ``depends=``-only manifests; the walk reads only the JSON
``dependencies`` key, so they would otherwise drop silently."""
if isinstance(data, dict) and not data.get("dependencies") and data.get("depends"):
# INFO: common and unactionable for transitive libraries; a WARNING
# on every build would train users to ignore the stream
_LOGGER.info(
"Library %s declares dependencies via library.properties "
"depends=, which are not resolved automatically; add them with "
"add_library() if needed",
name,
)
def dependency_is_usable(
dep: dict, platform: str | None, framework: str, requester: str
) -> bool:
@@ -773,20 +762,6 @@ def dependency_is_usable(
return True
def _split_owner(entry: dict) -> dict:
"""``owner/Name`` written as one name becomes owner and name, as in the dict form,
so the name matches the package the registry resolves it to."""
name, owner = entry.get("name"), entry.get("owner")
if isinstance(name, str) and owner is None and "://" not in name:
head, sep, tail = name.partition("/")
# A registry owner is a plain identifier; anything else is left to the
# backend's own checks
owner_ok = head.replace("-", "").replace("_", "").isalnum()
if sep and owner_ok and tail and "/" not in tail:
return {**entry, "owner": head, "name": tail}
return entry
def _valid_dependency_entry(entry: dict, manifest_name: str) -> bool:
"""Whether a normalized entry carries a usable name (non-empty string),
version (string, if present), and owner (string, if present); invalid
@@ -815,12 +790,16 @@ def normalize_dependencies(
so callers see a uniform list. ``manifest_name`` names the manifest in the
warning for entries that cannot be normalized.
Bare-name spellings carry no version; the walk settles them once the graph is known.
Bare-name spellings carry no version; the dependency walk later drops
version-less entries at DEBUG, since they are usually names of bundled
framework libraries (Wire, SPI) that need no registry install.
"""
if dependencies is None:
return []
if isinstance(dependencies, str):
return [_split_owner(entry) for entry in parse_properties_depends(dependencies)]
# A plain string is one or more comma-separated names; iterating it
# as a list would shred it into one-character "libraries"
return [{"name": n.strip()} for n in dependencies.split(",") if n.strip()]
if isinstance(dependencies, dict):
normalized = []
for raw_name, spec in dependencies.items():
@@ -847,10 +826,10 @@ def normalize_dependencies(
for entry in dependencies:
if isinstance(entry, dict):
if _valid_dependency_entry(entry, manifest_name):
normalized.append(_split_owner(entry))
normalized.append(entry)
elif isinstance(entry, str) and entry:
# PIO also accepts a bare list of names ("dependencies": ["Wire"])
normalized.append(_split_owner({"name": entry}))
normalized.append({"name": entry})
else:
_LOGGER.warning(
"Ignoring unrecognized dependency entry %r of %s",
@@ -992,55 +971,54 @@ def is_lib_ignored(name: str | None, lib_ignore: set[str]) -> bool:
)
def _satisfying_key(
name: str,
owner: str | None,
requester: str,
nodes: dict[str, _LibNode],
def _reconcile_versionless_skips(
skipped_versionless: list[tuple[Any, Any, str]],
components: dict[str, ConvertedLibrary],
) -> tuple[str, bool] | None:
"""The key of a resolved library a version-less dependency names, and whether
only its manifest name matched: the registry package name (with the owner,
when the dependency gives one) is checked first; never the requester itself."""
wanted = name.lower()
def owner_of(key: str, component: ConvertedLibrary) -> str:
# The owner as requested, else the one the registry (or a git path) gave
head, sep, _ = component.name.partition("/")
return nodes[key].owner or (head if sep else "")
candidates = [
(key, component)
for key, component in components.items()
if key != requester
and (not owner or owner_of(key, component).lower() == owner.lower())
]
for key, _ in candidates:
if (nodes[key].pkgname or key).lower() == wanted:
return key, False
if owner:
return None
for key, component in candidates:
manifest_name = component.data.get("name")
if isinstance(manifest_name, str) and manifest_name.lower() == wanted:
return key, True
return None
def _own_names(
key: str, nodes: dict[str, _LibNode], components: dict[str, ConvertedLibrary]
) -> set[str]:
"""The lower-case key, requested and resolved package names and the manifest
name a resolved library goes by."""
names = {
key.lower(),
(nodes[key].pkgname or key).lower(),
components[key].name.lower(),
}
manifest_name = components[key].data.get("name")
if isinstance(manifest_name, str):
names.add(manifest_name.lower())
return names
backend: LibraryBackend,
) -> None:
"""Warn for version-less deps nothing satisfied, and record the
backend-provided ones in ``backend.provided_requests`` for its
post-emit reconciliation; a silent drop surfaces as link errors far
from the cause."""
resolved_manifest_names = {c.data.get("name") for c in components.values()}
# A treeless backend can never supply a bundled name; noise for it
log = _LOGGER.warning if backend.provides is not None else _LOGGER.debug
warned: set[str] = set()
for dep_name, dep_owner, requester in skipped_versionless:
if not isinstance(dep_name, str) or not dep_name or dep_name in warned:
continue
if dep_name in components:
# A version-less dep's request key is the name itself
continue
if (
not dep_owner
and backend.provides is not None
and backend.provides(dep_name)
):
# provides() only satisfies owner-less names (same guard as
# the walk's skip); record for the post-emit reconciliation.
# Checked before the manifest-name evidence so the overlap
# case warns once, in the backend's own suppression loop
backend.provided_requests.add(dep_name)
continue
if dep_name in resolved_manifest_names:
# Name-only evidence: a coincidental collision must stay
# visible where the user could pin it
warned.add(dep_name)
log(
"Version-less dependency %s of %s assumed satisfied by a "
"resolved library's manifest name only",
dep_name,
requester,
)
continue
warned.add(dep_name)
log(
"Dependency %s of %s has no version to resolve and nothing "
"provides it; skipping",
dep_name,
requester,
)
def _fetch_source(
@@ -1169,12 +1147,7 @@ def convert_libraries(
managed components from ``add_idf_component``); unlike ``backend.provides`` they
carry an owner. They are excluded like ``lib_ignore`` so nothing is both
converted and managed, which ESP-IDF refuses to build.
A dependency named without a version is settled once the graph is known: wired
to a library already in it, left to ``backend.provides``, or fetched by name.
"""
from platformio.package.exception import UnknownPackageError
nodes: dict[str, _LibNode] = {}
# Folded into one set so every is_lib_ignored() call site honors both.
@@ -1192,16 +1165,10 @@ def convert_libraries(
else ""
)
resolved_requirements: dict[str, frozenset[str]] = {}
def add_spec(name: str | None, version: str | None, repository: str | None) -> str:
key, kind, locator = _node_key(name, version, repository)
node = nodes.get(key) or _LibNode(key=key, is_git=kind == "git")
nodes[key] = node
if kind != "registry" and not (node.is_git or node.is_local):
# A registry node (perhaps one the registry did not know) becomes a
# git or local source: resolve it again from that source
resolved_requirements.pop(key, None)
# The same key requested from two different kinds of source (or two
# different local paths) is a config mistake: one silently wins. Warn so
# it isn't a surprise. (git-vs-registry is reported separately below.)
@@ -1253,69 +1220,10 @@ def convert_libraries(
# requirement set has grown since the last time, so every requirement in the
# graph is accounted for before conversion.
components: dict[str, ConvertedLibrary] = {}
resolved_requirements: dict[str, frozenset[str]] = {}
top_level_keys = set(top_level)
def fetched_by_name(key: str) -> bool:
"""A registry node only version-less manifest dependencies asked for: a
miss skips it, and an explicit library of the same name replaces it."""
node = nodes[key]
return (
key not in top_level_keys
and not (node.is_git or node.is_local)
and not node.requirements
)
def backend_provides(name: str, owner: str | None) -> bool:
"""Whether the framework bundles an owner-less name; asked once per name,
since the backend may warn, and recorded in ``provided_requests``."""
if owner or backend.provides is None:
return False
if name in backend.provided_requests or backend.provides(name):
backend.provided_requests.add(name)
return True
return False
# Version-less dependencies per requester key, kept for the whole walk and
# wired once the graph is final; a re-walked requester replaces its entries
pending_versionless: dict[str, list[dict]] = {}
# Names fetched by name so far, whatever their case: each is asked for once
fetched: dict[tuple[str, str], str] = {}
def settle_versionless() -> list[str]:
"""Fetch by name the deferred dependencies nothing in the graph or the
framework satisfies yet; return the new keys to walk. Only a dependency
on the requester itself or with a bad name needs no wiring later."""
new_keys: list[str] = []
for requester, dependencies in pending_versionless.items():
kept: list[dict] = []
for dependency in dependencies:
name, owner = dependency["name"], dependency.get("owner")
if _owner_pkgname_to_name(owner, name).lower() in _own_names(
requester, nodes, components
):
_LOGGER.debug("Dependency %s of %s names itself", name, requester)
continue
# A library in the graph beats a bundled copy, as on the Arduino backend
if _satisfying_key(
name, owner, requester, nodes, components
) is None and not backend_provides(name, owner):
spelling = (name.lower(), (owner or "").lower())
if spelling not in fetched:
try:
dep_key = add_spec(
_owner_pkgname_to_name(owner, name), None, None
)
except RuntimeError as e:
_LOGGER.warning(
"Skipping dependency %s of %s: %s", name, requester, e
)
continue
fetched[spelling] = dep_key
new_keys.append(dep_key)
kept.append(dependency)
dependencies[:] = kept
return new_keys
# (name, owner, requester) reconciled against the final resolution set
skipped_versionless: list[tuple[Any, Any, str]] = []
worklist = deque(dict.fromkeys(top_level))
while worklist:
# Drain the frontier sequentially (spec resolution mutates shared
@@ -1337,17 +1245,9 @@ def convert_libraries(
elif node.is_local:
component = ConvertedLibrary(key, "*", LocalSource(node.local_path))
else:
try:
owner, name, version, url, size = _resolve_registry_version(
node.owner, node.pkgname, node.requirements
)
except UnknownPackageError as e:
# A requirement conflict is a RuntimeError and always raises
if not fetched_by_name(key):
raise
# Reported per requester once the graph is final
_LOGGER.debug("Library %s is not in the registry: %s", key, e)
continue
owner, name, version, url, size = _resolve_registry_version(
node.owner, node.pkgname, node.requirements
)
component = ConvertedLibrary(
_owner_pkgname_to_name(owner, name), version, URLSource(url, size)
)
@@ -1355,9 +1255,9 @@ def convert_libraries(
_prefetch_wave(wave, salt, backend.cache_key)
for key, component in wave:
node = nodes[key]
if frozenset(node.requirements) != resolved_requirements.get(key):
# Requirements grew (or the source changed) mid-wave: skip parsing
# a manifest the next wave will re-resolve and replace
if frozenset(node.requirements) != resolved_requirements[key]:
# Requirements grew mid-wave: skip parsing a manifest the
# next wave will re-resolve and replace
worklist.append(key)
continue
component.download(salt=salt, namespace=backend.cache_key)
@@ -1413,6 +1313,8 @@ def convert_libraries(
raise EsphomeError(f"Library {key} has a malformed manifest")
_LOGGER.warning("Skipping dependency %s: malformed manifest", key)
continue
warn_properties_depends(component.name, component.data)
try:
check_library_data(component.data, backend.platform, backend.framework)
except InvalidLibrary as e:
@@ -1435,19 +1337,27 @@ def convert_libraries(
# Requirements changed (we got past the short-circuit above), so
# (re)walk this component's dependencies.
node.edges = set()
pending = pending_versionless[key] = []
for dependency in normalize_dependencies(
component.data.get("dependencies"), component.name
):
if not dependency.get("version"):
# An empty version is as good as none; it settles the same way
if "version" not in dependency:
# Cannot resolve from the registry; the post-emit
# reconciliation owns the drop warning
dep_name = dependency.get("name")
_LOGGER.debug(
"Skip version-less dependency %r of %s",
dep_name,
component.name,
)
if not is_lib_ignored(
dependency.get("name"), lib_ignore
dep_name, lib_ignore
) and dependency_is_usable(
dependency, backend.platform, backend.framework, component.name
):
# Filtered or ignored deps are deliberately absent
pending.append(dependency)
skipped_versionless.append(
(dep_name, dependency.get("owner"), component.name)
)
continue
if not dependency_is_usable(
dependency, backend.platform, backend.framework, component.name
@@ -1465,7 +1375,11 @@ def convert_libraries(
dep_url = _url_or_none(dep_version)
if dep_url is not None:
dep_version = None
elif backend_provides(dep_name, dependency.get("owner")):
elif (
backend.provides is not None
and not dependency.get("owner")
and backend.provides(dep_name)
):
# The backend adds it from its own tree; resolving here
# would fetch a same-named registry package
if dep_version and dep_version != "*":
@@ -1478,12 +1392,11 @@ def convert_libraries(
)
else:
_LOGGER.debug("Skip backend-provided dependency %s", dep_name)
backend.provided_requests.add(dep_name)
continue
dep_key = add_spec(dep_name, dep_version, dep_url)
node.edges.add(dep_key)
worklist.append(dep_key)
if not worklist:
worklist.extend(settle_versionless())
# A git or local source wins over the same component requested from the
# registry. That's intentional, but warn so the dropped registry spec isn't
@@ -1507,56 +1420,6 @@ def convert_libraries(
source,
)
# Every version-less request resolves to the explicit library that goes by
# its name if the final graph has one, else to the node fetched by name for
# that package, else to the framework's bundled copy, else to nothing
explicit = {k: c for k, c in components.items() if not fetched_by_name(k)}
by_name = {k: c for k, c in components.items() if fetched_by_name(k)}
unresolved: dict[tuple[str, str], None] = {}
for requester, dependencies in pending_versionless.items():
for dependency in dependencies:
name, owner = dependency["name"], dependency.get("owner")
found = _satisfying_key(
name, owner, requester, nodes, explicit
) or _satisfying_key(name, owner, requester, nodes, by_name)
if found is None:
if not backend_provides(name, owner):
unresolved[(requester, name)] = None
continue
key, by_manifest_name = found
if by_manifest_name:
# Visible, so a chance name collision can be pinned to the right package
_LOGGER.info(
"Dependency %s of %s satisfied by %s through its manifest name",
name,
requester,
key,
)
else:
_LOGGER.debug(
"Dependency %s of %s satisfied by %s", name, requester, key
)
nodes[requester].edges.add(key)
# What no listed library needs any more is not built: a node fetched by name
# that an explicit library replaced, and whatever only it pulled in
reachable: set[str] = set()
stack = [key for key in top_level if key in components]
while stack:
if (key := stack.pop()) in reachable:
continue
reachable.add(key)
stack.extend(edge for edge in nodes[key].edges if edge in components)
for key in [key for key in components if key not in reachable]:
_LOGGER.debug("Library %s is not needed; dropping it", key)
del components[key]
for requester, name in unresolved:
if requester in components:
_LOGGER.warning(
"Dependency %s of %s was not resolved; skipping", name, requester
)
# Two graph nodes that resolve to the same component name (e.g. a package
# referenced both bare and as ``owner/name``) are not deduplicated and can
# produce conflicting component definitions. Warn so it's not silent.
@@ -1584,4 +1447,6 @@ def convert_libraries(
for component in components.values():
backend.emit(component)
_reconcile_versionless_skips(skipped_versionless, components, backend)
return [components[key] for key in top_level if key in components]
+2
View File
@@ -219,6 +219,8 @@ ESPHOME_H_EXCLUDE = {
Path(ENTITY_TYPES_H_TARGET),
# main.cpp includes it after defining esphome_controllers()
Path("esphome/core/controller_dispatch.h"),
# only the esp32 sscanf wrap includes it
Path("esphome/core/sscanf_no_float.h"),
# build machinery, not user API
Path("esphome/core/pch_prefix.h"),
}
@@ -0,0 +1,9 @@
esphome:
name: test
esp32:
variant: esp32c6
framework:
type: esp-idf
esp32_ble_tracker:
@@ -0,0 +1,11 @@
esphome:
name: test
esp32:
board: esp32dev
framework:
type: esp-idf
advanced:
enable_full_printf: true
esp32_ble_tracker:
@@ -0,0 +1,11 @@
esphome:
name: test
esp32:
board: esp32dev
framework:
type: esp-idf
advanced:
enable_full_scanf: true
esp32_ble_tracker:
@@ -0,0 +1,17 @@
esphome:
name: test
esp32:
board: esp32dev
framework:
type: esp-idf
esp32_ble_tracker:
sensor:
- platform: template
name: Parsed
lambda: |-
float value = 0;
sscanf("1.5", "%f", &value);
return value;
+34 -11
View File
@@ -2260,28 +2260,51 @@ def test_mbedtls_tls_zigbee_requires_extras(
assert mbedtls.tls_extras_required == _CCM_EXTRAS
_VASPRINTF_STUB_FLAGS = {"-Wl,--wrap=vasprintf", "-Wl,--undefined=__wrap_vasprintf"}
@pytest.mark.parametrize(
("config_file", "expected"),
("config_file", "symbol", "expected"),
[
pytest.param("vasprintf_stub_c6.yaml", True, id="c6"),
pytest.param("vasprintf_stub_c6_full_printf.yaml", False, id="c6_full_printf"),
pytest.param("exclusion_reincludes.yaml", False, id="esp32"),
# vasprintf: only where the ROM lacks vasprintf but has vsnprintf
pytest.param("vasprintf_stub_c6.yaml", "vasprintf", True, id="vasprintf_c6"),
pytest.param(
"vasprintf_stub_c6_full_printf.yaml",
"vasprintf",
False,
id="vasprintf_c6_full_printf",
),
pytest.param(
"exclusion_reincludes.yaml", "vasprintf", False, id="vasprintf_esp32"
),
# sscanf: only for Bluetooth builds on variants whose ROM lacks sscanf,
# and not when a lambda scans a float
pytest.param(
"exclusion_reincludes_wifi_ble.yaml", "sscanf", True, id="sscanf_esp32_ble"
),
pytest.param("sscanf_stub_ble_c6.yaml", "sscanf", False, id="sscanf_c6_rom"),
pytest.param(
"sscanf_stub_ble_full_scanf.yaml", "sscanf", False, id="sscanf_full_scanf"
),
pytest.param(
"sscanf_stub_ble_lambda.yaml", "sscanf", False, id="sscanf_lambda_float"
),
pytest.param(
"sscanf_stub_ble_full_printf.yaml", "sscanf", True, id="sscanf_full_printf"
),
pytest.param("exclusion_reincludes.yaml", "sscanf", False, id="sscanf_no_ble"),
],
)
def test_vasprintf_stub_only_on_rom_vsnprintf_variants(
def test_wrap_stub_emission(
generate_main: Callable[[str | Path], str],
component_config_path: Callable[[str], Path],
config_file: str,
symbol: str,
expected: bool,
) -> None:
"""The vasprintf wrap is emitted only where the ROM lacks vasprintf but has vsnprintf."""
"""A linker wrap stub is emitted as a define plus --wrap/--undefined flag pair."""
generate_main(component_config_path(config_file))
assert (CORE.build_flags >= _VASPRINTF_STUB_FLAGS) is expected
flags = {f"-Wl,--wrap={symbol}", f"-Wl,--undefined=__wrap_{symbol}"}
assert (CORE.build_flags >= flags) is expected
defines = {define.name for define in CORE.defines}
assert ("USE_ESP32_VASPRINTF_STUB" in defines) is expected
assert (f"USE_ESP32_{symbol.upper()}_STUB" in defines) is expected
@pytest.mark.parametrize(
@@ -0,0 +1,250 @@
#include <gtest/gtest.h>
#include <cstdarg>
#include <cstdint>
#include <cstdio>
#include "esphome/core/sscanf_no_float.h"
namespace esphome::core::testing {
static int sscanf_no_float(const char *str, const char *fmt, ...) {
va_list ap;
va_start(ap, fmt);
int result = vsscanf_no_float(str, fmt, ap);
va_end(ap);
return result;
}
// --- The two bluedroid call sites the wrap exists for ---
TEST(SscanfNoFloat, BluedroidHexByte) {
unsigned int value = 0;
EXPECT_EQ(sscanf_no_float("ab", "%02x", &value), 1);
EXPECT_EQ(value, 0xabu);
// Width stops after two digits even when more follow
EXPECT_EQ(sscanf_no_float("ab12", "%02x", &value), 1);
EXPECT_EQ(value, 0xabu);
}
TEST(SscanfNoFloat, BluedroidAddress) {
uint32_t a[6] = {0};
EXPECT_EQ(sscanf_no_float("AA:bb:0C:dd:ee:FF", "%02x:%02x:%02x:%02x:%02x:%02x", a, a + 1, a + 2, a + 3, a + 4, a + 5),
6);
EXPECT_EQ(a[0], 0xaau);
EXPECT_EQ(a[1], 0xbbu);
EXPECT_EQ(a[2], 0x0cu);
EXPECT_EQ(a[3], 0xddu);
EXPECT_EQ(a[4], 0xeeu);
EXPECT_EQ(a[5], 0xffu);
// Short input assigns what matched, like libc
EXPECT_EQ(sscanf_no_float("AA:bb:0C", "%02x:%02x:%02x:%02x:%02x:%02x", a, a + 1, a + 2, a + 3, a + 4, a + 5), 3);
}
TEST(SscanfNoFloat, WidthWithByteLength) {
uint8_t mac[2] = {0};
EXPECT_EQ(sscanf_no_float("AA:bb", "%02hhx:%02hhx", mac, mac + 1), 2);
EXPECT_EQ(mac[0], 0xaa);
EXPECT_EQ(mac[1], 0xbb);
// The width still bounds the digits when a length modifier is present
EXPECT_EQ(sscanf_no_float("abcd", "%02hhx", mac), 1);
EXPECT_EQ(mac[0], 0xab);
}
// --- Formats bluedroid's hands-free client uses: %hu with %n ---
TEST(SscanfNoFloat, ShortWithPosition) {
uint16_t h = 0, m = 0, s = 0;
int n1 = 0, n2 = 0, n3 = 0;
EXPECT_EQ(sscanf_no_float("6:30:15", "%hu%n:%hu%n:%hu%n", &h, &n1, &m, &n2, &s, &n3), 3);
EXPECT_EQ(h, 6);
EXPECT_EQ(m, 30);
EXPECT_EQ(s, 15);
EXPECT_EQ(n1, 1);
EXPECT_EQ(n2, 4);
EXPECT_EQ(n3, 7);
EXPECT_EQ(sscanf_no_float("M3.2.0", "M%hu%n.%hu%n.%hu%n", &h, &n1, &m, &n2, &s, &n3), 3);
EXPECT_EQ(sscanf_no_float("X3.2.0", "M%hu", &h), 0);
}
// --- Integer conversions ---
TEST(SscanfNoFloat, SignedDecimalSkipsLeadingSpace) {
int value = 0;
EXPECT_EQ(sscanf_no_float(" -42 rest", "%d", &value), 1);
EXPECT_EQ(value, -42);
EXPECT_EQ(sscanf_no_float("+7", "%u", &value), 1);
EXPECT_EQ(value, 7);
}
TEST(SscanfNoFloat, AutoBase) {
int value = 0;
EXPECT_EQ(sscanf_no_float("0x1F", "%i", &value), 1);
EXPECT_EQ(value, 31);
EXPECT_EQ(sscanf_no_float("017", "%i", &value), 1);
EXPECT_EQ(value, 15);
EXPECT_EQ(sscanf_no_float("0", "%i", &value), 1);
EXPECT_EQ(value, 0);
EXPECT_EQ(sscanf_no_float("-0x10", "%i", &value), 1);
EXPECT_EQ(value, -16);
}
TEST(SscanfNoFloat, OctalAndHexPrefix) {
unsigned int value = 0;
EXPECT_EQ(sscanf_no_float("17", "%o", &value), 1);
EXPECT_EQ(value, 15u);
EXPECT_EQ(sscanf_no_float("+0x10", "%x", &value), 1);
EXPECT_EQ(value, 16u);
// "0x" with no digit after it: only the 0 is consumed
EXPECT_EQ(sscanf_no_float("0xg", "%x", &value), 1);
EXPECT_EQ(value, 0u);
EXPECT_EQ(sscanf_no_float("zz", "%x", &value), 0);
}
TEST(SscanfNoFloat, WidthSplitsDigits) {
int a = 0;
unsigned int b = 0;
EXPECT_EQ(sscanf_no_float("12345", "%3d%d", &a, &b), 2);
EXPECT_EQ(a, 123);
EXPECT_EQ(b, 45u);
EXPECT_EQ(sscanf_no_float("ff", "%1x", &b), 1);
EXPECT_EQ(b, 15u);
}
TEST(SscanfNoFloat, LengthModifiers) {
unsigned char hh = 0;
int16_t h = 0;
long l = 0; // NOLINT(google-runtime-int) %ld is defined in terms of long
uint64_t ll = 0;
size_t z = 0;
ptrdiff_t t = 0;
EXPECT_EQ(sscanf_no_float("255", "%hhu", &hh), 1);
EXPECT_EQ(hh, 255);
EXPECT_EQ(sscanf_no_float("-2", "%hd", &h), 1);
EXPECT_EQ(h, -2);
EXPECT_EQ(sscanf_no_float("-1", "%ld", &l), 1);
EXPECT_EQ(l, -1L);
EXPECT_EQ(sscanf_no_float("18446744073709551615", "%llu", &ll), 1);
EXPECT_EQ(ll, UINT64_MAX);
EXPECT_EQ(sscanf_no_float("123", "%ju", &ll), 1);
EXPECT_EQ(ll, 123u);
EXPECT_EQ(sscanf_no_float("9", "%zu", &z), 1);
EXPECT_EQ(z, 9u);
EXPECT_EQ(sscanf_no_float("-9", "%td", &t), 1);
EXPECT_EQ(t, -9);
}
TEST(SscanfNoFloat, Pointer) {
void *p = nullptr;
EXPECT_EQ(sscanf_no_float("0x1234", "%p", &p), 1);
EXPECT_EQ(reinterpret_cast<uintptr_t>(p), 0x1234u);
EXPECT_EQ(sscanf_no_float("abcd rest", "%p", &p), 1);
EXPECT_EQ(reinterpret_cast<uintptr_t>(p), 0xabcdu);
}
TEST(SscanfNoFloat, Suppression) {
int value = 0;
EXPECT_EQ(sscanf_no_float("1 2 3", "%*d %d %*d", &value), 1);
EXPECT_EQ(value, 2);
}
// --- Character, string and scanset conversions ---
TEST(SscanfNoFloat, CharDoesNotSkipSpace) {
char c[3] = {0};
EXPECT_EQ(sscanf_no_float("xyz", "%c%c", c, c + 1), 2);
EXPECT_EQ(c[0], 'x');
EXPECT_EQ(c[1], 'y');
EXPECT_EQ(sscanf_no_float(" x", "%c", c), 1);
EXPECT_EQ(c[0], ' ');
}
TEST(SscanfNoFloat, CharWidthShortRead) {
char c[4] = {0};
EXPECT_EQ(sscanf_no_float("abc", "%2c", c), 1);
EXPECT_EQ(c[0], 'a');
EXPECT_EQ(c[1], 'b');
EXPECT_EQ(c[2], '\0');
// Fewer characters than the width still count as one assignment, like newlib
EXPECT_EQ(sscanf_no_float("ab", "%3c", c), 1);
EXPECT_EQ(sscanf_no_float("", "%c", c), EOF);
}
TEST(SscanfNoFloat, StringStopsAtSpaceAndWidth) {
char buf[16] = {0};
EXPECT_EQ(sscanf_no_float("hello world", "%5s", buf), 1);
EXPECT_STREQ(buf, "hello");
EXPECT_EQ(sscanf_no_float(" x", "%1s", buf), 1);
EXPECT_STREQ(buf, "x");
EXPECT_EQ(sscanf_no_float(" ", "%s", buf), EOF);
}
TEST(SscanfNoFloat, Scansets) {
char buf[16] = {0};
char rest[16] = {0};
EXPECT_EQ(sscanf_no_float("key=val", "%[^=]=%s", buf, rest), 2);
EXPECT_STREQ(buf, "key");
EXPECT_STREQ(rest, "val");
// ']' first is a member; '-' last is literal
EXPECT_EQ(sscanf_no_float("]]ab", "%[]]", buf), 1);
EXPECT_STREQ(buf, "]]");
EXPECT_EQ(sscanf_no_float("a-z]x", "%[a-]", buf), 1);
EXPECT_STREQ(buf, "a-");
EXPECT_EQ(sscanf_no_float("abc", "%[^]]", buf), 1);
EXPECT_STREQ(buf, "abc");
// No member matched is a matching failure
EXPECT_EQ(sscanf_no_float("123", "%[a-z]", buf), 0);
}
// --- Literals, whitespace and return value ---
TEST(SscanfNoFloat, LiteralsAndPercent) {
int value = 0;
EXPECT_EQ(sscanf_no_float("50%", "%d%%", &value), 1);
EXPECT_EQ(value, 50);
// Like newlib, %% does not skip whitespace; an explicit space in the format does
EXPECT_EQ(sscanf_no_float("50 % 7", "%d%%%d", &value, &value), 1);
EXPECT_EQ(sscanf_no_float("50 % 7", "%d %%%d", &value, &value), 2);
EXPECT_EQ(value, 7);
EXPECT_EQ(sscanf_no_float("1,2", "%d,%d", &value, &value), 2);
EXPECT_EQ(sscanf_no_float("1 ,2", "%d,%d", &value, &value), 1);
EXPECT_EQ(sscanf_no_float("1, 2", "%d,%d", &value, &value), 2);
EXPECT_EQ(sscanf_no_float("12\n34", "%d%d", &value, &value), 2);
}
TEST(SscanfNoFloat, PositionOnly) {
int n = 0;
EXPECT_EQ(sscanf_no_float("ab", "ab%n", &n), 0);
EXPECT_EQ(n, 2);
EXPECT_EQ(sscanf_no_float("", "%n", &n), 0);
EXPECT_EQ(n, 0);
}
TEST(SscanfNoFloat, EofVersusMatchingFailure) {
int value = 0;
EXPECT_EQ(sscanf_no_float("", "%d", &value), EOF);
EXPECT_EQ(sscanf_no_float(" ", "%d", &value), EOF);
EXPECT_EQ(sscanf_no_float("a", "ab"), EOF);
EXPECT_EQ(sscanf_no_float("abc", "%d", &value), 0);
EXPECT_EQ(sscanf_no_float("ac", "ab"), 0);
EXPECT_EQ(sscanf_no_float("12abc", "%d%d", &value, &value), 1);
EXPECT_EQ(sscanf_no_float("12", "%d%d", &value, &value), 1);
}
// --- Conversions the scanner refuses rather than misparses ---
TEST(SscanfNoFloat, UnsupportedConversions) {
float f = 0;
double d = 0;
wchar_t w[4] = {0};
EXPECT_EQ(sscanf_no_float("1.5", "%f", &f), SSCANF_UNSUPPORTED);
EXPECT_EQ(sscanf_no_float("1.5", "%lf", &d), SSCANF_UNSUPPORTED);
EXPECT_EQ(sscanf_no_float("1e3", "%g", &d), SSCANF_UNSUPPORTED);
EXPECT_EQ(sscanf_no_float("0x1p3", "%a", &d), SSCANF_UNSUPPORTED);
EXPECT_EQ(sscanf_no_float("abc", "%ls", w), SSCANF_UNSUPPORTED);
EXPECT_EQ(sscanf_no_float("abc", "%lc", w), SSCANF_UNSUPPORTED);
// Earlier conversions do not hide a later unsupported one
int value = 0;
EXPECT_EQ(sscanf_no_float("1 2.5", "%d %f", &value, &f), SSCANF_UNSUPPORTED);
}
} // namespace esphome::core::testing
+4
View File
@@ -236,6 +236,10 @@ async def yaml_config(request: pytest.FixtureRequest, unused_tcp_port: int) -> s
external_components_path = str(FIXTURES_DIR / "external_components")
content = content.replace("EXTERNAL_COMPONENT_PATH", external_components_path)
# Replace the user includes path placeholder if present
if "INCLUDES_PATH" in content:
content = content.replace("INCLUDES_PATH", str(FIXTURES_DIR / "includes"))
if shared_name is not None:
# _compile verifies the marked test compiles this content unmodified
request.node._shared_yaml_content = content
@@ -0,0 +1,6 @@
#pragma once
static constexpr int HEADER_CONSTANT = 42;
// Names the global by its id, so the header must come after the id declarations.
inline int header_reads_global() { return header_value->value() * 2; }
@@ -0,0 +1,27 @@
esphome:
name: includes-lambda-order
includes:
- INCLUDES_PATH/lambda_order.h
on_boot:
# An apply action value is rendered at file scope.
- globals.set:
id: header_value
value: !lambda return HEADER_CONSTANT;
host:
api:
logger:
globals:
- id: header_value
type: int
initial_value: "0"
sensor:
- platform: template
name: Header Value
update_interval: 100ms
lambda: return header_reads_global();
filters:
# A lambda in a flash table.
- filter_out: !lambda return HEADER_CONSTANT - 1;
@@ -0,0 +1,31 @@
"""Lambdas rendered at file scope can use the user's includes, which can name ids."""
from __future__ import annotations
from aioesphomeapi import SensorInfo, SensorState
import pytest
from .state_utils import require_entity, wait_for_state
from .types import APIClientConnectedFactory, RunCompiledFunction
@pytest.mark.asyncio
async def test_includes_lambda_order(
yaml_config: str,
run_compiled: RunCompiledFunction,
api_client_connected: APIClientConnectedFactory,
) -> None:
"""The apply action and the sensor filter use the header, the header reads the global."""
async with run_compiled(yaml_config), api_client_connected() as client:
entities, _ = await client.list_entities_services()
sensor = require_entity(entities, "header_value", SensorInfo)
state = await wait_for_state(
client,
lambda s: (
isinstance(s, SensorState)
and s.key == sensor.key
and not s.missing_state
),
)
# globals.set stored HEADER_CONSTANT and the header read it back doubled.
assert state.state == 84.0
@@ -62,9 +62,8 @@ def test_write_idf_component_yml_passes_managed_components(
converted.get_sanitized_name.return_value = "esphome/other-lib"
converted.path = tmp_path / "pio_components" / "other-lib"
def fake_generate_idf_components(libraries, managed=None, provides=None):
def fake_generate_idf_components(libraries, managed=None):
captured["managed"] = managed
captured["provides"] = provides
return [converted]
monkeypatch.setattr(esp32, "generate_idf_components", fake_generate_idf_components)
@@ -72,8 +71,6 @@ def test_write_idf_component_yml_passes_managed_components(
esp32._write_idf_component_yml()
assert captured["managed"] == {"esphome/noise-c"}
# ESP-IDF bundles no Arduino libraries, so every name may be fetched
assert captured["provides"] is None
# The managed component itself is still written into the manifest deps
# directly (from KEY_COMPONENTS), just not converted a second time.
yml_path = tmp_path / "src" / "idf_component.yml"
@@ -96,7 +93,7 @@ def test_write_idf_component_yml_empty_managed_when_no_components(
captured: dict[str, set[str] | None] = {}
def fake_generate_idf_components(libraries, managed=None, provides=None):
def fake_generate_idf_components(libraries, managed=None):
captured["managed"] = managed
return []
@@ -107,50 +104,6 @@ def test_write_idf_component_yml_empty_managed_when_no_components(
assert captured["managed"] == set()
def test_write_idf_component_yml_marks_arduino_bundled_libraries_provided(
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""Under Arduino the converter is told which names the core bundles, so a
manifest dependency on Wire or SPI is never fetched from the registry."""
_setup_core(tmp_path)
CORE.data[KEY_CORE][KEY_TARGET_FRAMEWORK] = str(Framework.ARDUINO)
CORE.data[KEY_CORE][KEY_FRAMEWORK_VERSION] = cv.Version(3, 3, 12)
CORE.data[esp32.KEY_ESP32] = {
esp32.KEY_COMPONENTS: {},
esp32.KEY_IDF_VERSION: cv.Version(5, 5, 4),
esp32.KEY_ARDUINO_LIBRARIES: {"Wire", "SPI"},
}
captured: dict[str, object] = {}
def fake_generate_idf_components(libraries, managed=None, provides=None):
captured["provides"] = provides
return []
monkeypatch.setattr(esp32, "generate_idf_components", fake_generate_idf_components)
esp32._write_idf_component_yml()
provides = captured["provides"]
assert provides("Wire") and provides("SPI")
assert not provides("Adafruit BusIO")
def test_arduino_provides_warns_for_a_bundled_library_not_enabled(
caplog: pytest.LogCaptureFixture,
) -> None:
"""A manifest dependency on a bundled Arduino library is never fetched, but
the library still has to be enabled, so an unenabled one is called out."""
CORE.reset()
CORE.data[esp32.KEY_ESP32] = {esp32.KEY_ARDUINO_LIBRARIES: {"Wire"}}
assert esp32._arduino_provides("Wire")
assert "Wire" not in caplog.text
assert esp32._arduino_provides("SPI")
assert "Arduino library SPI, which is not enabled" in caplog.text
assert not esp32._arduino_provides("Adafruit BusIO")
@pytest.mark.parametrize(
("version", "libsodium_stubbed"),
[
@@ -247,9 +200,7 @@ def test_write_idf_component_yml_stub_yields_to_converted_library_of_same_name(
converted.get_sanitized_name.return_value = "esphome/libsodium"
converted.path = tmp_path / "pio_components" / "esphome" / "libsodium"
monkeypatch.setattr(
esp32,
"generate_idf_components",
lambda libraries, managed=None, provides=None: [converted],
esp32, "generate_idf_components", lambda libraries, managed=None: [converted]
)
esp32._write_idf_component_yml()
+2 -58
View File
@@ -8,64 +8,8 @@ from unittest.mock import MagicMock, patch
import pytest
from esphome.components import esp8266
from esphome.components.esp8266 import check_rosetta, lambdas_use_scanf_float
from esphome.core import EsphomeError, Lambda
from esphome.types import ConfigType
@pytest.mark.parametrize(
("src", "expected"),
[
# Basic float formats
('sscanf(buf, "%f", &v)', True),
('sscanf(buf, "%F", &v)', True),
('sscanf(buf, "%e", &v)', True),
('sscanf(buf, "%E", &v)', True),
('sscanf(buf, "%g", &v)', True),
('sscanf(buf, "%G", &v)', True),
('sscanf(buf, "%a", &v)', True),
('sscanf(buf, "%A", &v)', True),
# With modifiers
('sscanf(buf, "%lf", &v)', True),
('sscanf(buf, "%Lf", &v)', True),
('sscanf(buf, "%8lf", &v)', True),
('sscanf(buf, "%*f")', True),
('sscanf(buf, "%.2f", &v)', True),
# Mixed formats
('sscanf(buf, "%d,%f", &a, &b)', True),
# fscanf and std::sscanf
('fscanf(fp, "%f", &v)', True),
('std::sscanf(buf, "%f", &v)', True),
# Multi-line
('sscanf(buf,\n"%f", &v)', True),
# No float format
('sscanf(buf, "%d", &v)', False),
('sscanf(buf, "%s", s)', False),
# printf not scanf
('printf("%f", val)', False),
# %f in a different statement after scanf
('sscanf(buf, "%d", &x); printf("%f", val);', False),
# scanf %f in comment only
('// sscanf(buf, "%f", &v)\nsscanf(buf, "%d", &x)', False),
('/* sscanf(buf, "%f") */\nsscanf(buf, "%d", &x)', False),
],
)
def test_lambdas_use_scanf_float(src: str, expected: bool) -> None:
"""Test scanf float detection in lambda source."""
config: ConfigType = {"test": [Lambda(src)]}
assert lambdas_use_scanf_float(config) is expected
def test_lambdas_use_scanf_float_no_lambdas() -> None:
"""Test with config containing no lambdas."""
config: ConfigType = {"key": "value", "list": [1, 2]}
assert lambdas_use_scanf_float(config) is False
def test_lambdas_use_scanf_float_nested() -> None:
"""Test detection in deeply nested config."""
config: ConfigType = {"a": {"b": {"c": [Lambda('sscanf(buf, "%f", &v)')]}}}
assert lambdas_use_scanf_float(config) is True
from esphome.components.esp8266 import check_rosetta
from esphome.core import EsphomeError
@pytest.fixture
+35 -1
View File
@@ -26,7 +26,7 @@ from esphome.const import (
KEY_TARGET_PLATFORM,
Toolchain,
)
from esphome.core import CORE, KEY_CONTROLLER_REGISTRY_CONTROLLERS, config
from esphome.core import CORE, ID, KEY_CONTROLLER_REGISTRY_CONTROLLERS, config
from esphome.core.config import (
CONF_RELATION,
CONF_SUSPEND_LOOP,
@@ -1114,6 +1114,40 @@ async def test_add_includes_with_single_file(
assert any('#include "my_header.h"' in inc for inc in includes_added)
@pytest.mark.asyncio
async def test_add_includes_sit_between_declarations_and_generated_code(
tmp_path: Path, mock_copy_file_if_changed: Mock
) -> None:
"""A header can name an id, and generated functions and tables can use the header."""
CORE.config_path = tmp_path / "config.yaml"
CORE.build_path = tmp_path / "build"
CORE.build_path.mkdir(parents=True, exist_ok=True)
header = tmp_path / "my_header.h"
header.write_text("static constexpr int MY_CONSTANT = 42;")
CORE.config = {}
value = cg.new_Pvariable(ID("my_value", is_declaration=True, type=cg.int32))
cg.static_function(
"esphome__my_action__fn", cg.void, [], [f"*{value} = MY_CONSTANT;"]
)
cg.shared_progmem_array("my_table", cg.int32, [cg.RawExpression("MY_CONSTANT")])
await config.add_includes([str(header)])
# A job that runs after the includes still lands before the generated code.
cg.add_global(cg.RawStatement("static constexpr int FINAL_JOB = 1;"))
section = CORE.cpp_global_section
expected_order = [
"*const my_value",
"static void esphome__my_action__fn();",
'#include "my_header.h"',
"static constexpr int FINAL_JOB = 1;",
"static void esphome__my_action__fn() {",
"my_table[] PROGMEM",
]
positions = [section.index(text) for text in expected_order]
assert positions == sorted(positions), section
@pytest.mark.asyncio
@pytest.mark.skipif(os.name == "nt", reason="Unix-specific test")
async def test_add_includes_with_directory_unix(
+4 -3
View File
@@ -846,14 +846,15 @@ def test_library_info_dot_a_linkage_malformed_raises(tmp_path: Path) -> None:
def test_bundled_library_properties_depends_warns(
tmp_path: Path, caplog: pytest.LogCaptureFixture
) -> None:
"""The library.properties depends= spelling is a dependency list too, so a
bundled manifest using it gets the same visibility warning."""
"""The library.properties depends= spelling reaches the visibility
warning too; the shared parser returns it raw."""
framework = _make_framework(tmp_path)
wire = framework / "libraries" / "Wire"
(wire / "library.properties").write_text("name=Wire\nversion=1.0\ndepends=SPI\n")
_add_library("Wire", None)
caplog.set_level("INFO")
_resolve(framework)
assert "Bundled library Wire declares dependencies" in caplog.text
assert "Library Wire declares dependencies via library.properties" in caplog.text
def test_bundled_library_extra_script_raises(tmp_path: Path) -> None:
+24 -5
View File
@@ -785,8 +785,9 @@ class MockCodegen(NamedTuple):
new_pvariable: MagicMock
register_parented: AsyncMock
add_global: MagicMock
add_global_after_includes: MagicMock
add: MagicMock
calls: MagicMock # new_pvariable and add_global attached, to check their order
calls: MagicMock # the global emitters and new_pvariable attached, to check their order
@pytest.fixture
@@ -848,6 +849,9 @@ def mock_cg() -> Generator[MockCodegen]:
"esphome.codegen.register_parented", new_callable=AsyncMock
) as register_parented,
patch("esphome.cpp_generator.add_global") as add_global,
patch(
"esphome.cpp_generator.add_global_after_includes"
) as add_global_after_includes,
patch("esphome.codegen.add") as add,
):
get_variable.return_value = PARENT_OBJ
@@ -855,8 +859,15 @@ def mock_cg() -> Generator[MockCodegen]:
calls = MagicMock()
calls.attach_mock(new_pvariable, "new_pvariable")
calls.attach_mock(add_global, "add_global")
calls.attach_mock(add_global_after_includes, "add_global_after_includes")
yield MockCodegen(
get_variable, new_pvariable, register_parented, add_global, add, calls
get_variable,
new_pvariable,
register_parented,
add_global,
add_global_after_includes,
add,
calls,
)
@@ -984,7 +995,7 @@ async def _run_apply_condition(
def _apply_definition(mock_cg: MockCodegen) -> str:
return str(mock_cg.add_global.call_args.args[0])
return str(mock_cg.add_global_after_includes.call_args.args[0])
@pytest.mark.asyncio
@@ -998,8 +1009,16 @@ async def test_register_apply_action_entry(
action_id, template_arg = mock_cg.new_pvariable.call_args.args
assert action_id == ID("obj_1")
assert str(template_arg) == "<esphome__obj_1__fn, int32_t>"
# The definition must precede the storage line that names the function.
assert [c[0] for c in mock_cg.calls.mock_calls] == ["add_global", "new_pvariable"]
# The declaration must precede the storage line that names the function; the body
# goes after the user includes.
assert [c[0] for c in mock_cg.calls.mock_calls] == [
"add_global",
"add_global_after_includes",
"new_pvariable",
]
assert str(mock_cg.add_global.call_args.args[0]) == (
"static void esphome__obj_1__fn(const std::remove_cvref_t<int32_t> & x);"
)
assert _apply_definition(mock_cg).startswith(
"static void esphome__obj_1__fn(const std::remove_cvref_t<int32_t> & x) {"
)
+162
View File
@@ -0,0 +1,162 @@
"""Unit tests for esphome.code_scan."""
from pathlib import Path
from unittest.mock import MagicMock, patch
import pytest
from esphome.code_scan import (
external_components_use_scanf_float,
includes_use_scanf_float,
keep_float_scanf,
lambdas_use_scanf_float,
user_code_uses_scanf_float,
)
from esphome.const import CONF_ESPHOME, CONF_INCLUDES, CONF_INCLUDES_C
from esphome.core import CORE, Lambda
from esphome.loader import CORE_COMPONENTS_PATH, FileResource
from esphome.types import ConfigType
@pytest.mark.parametrize(
("src", "expected"),
[
# Basic float formats
('sscanf(buf, "%f", &v)', True),
('sscanf(buf, "%F", &v)', True),
('sscanf(buf, "%e", &v)', True),
('sscanf(buf, "%E", &v)', True),
('sscanf(buf, "%g", &v)', True),
('sscanf(buf, "%G", &v)', True),
('sscanf(buf, "%a", &v)', True),
('sscanf(buf, "%A", &v)', True),
# With modifiers
('sscanf(buf, "%lf", &v)', True),
('sscanf(buf, "%Lf", &v)', True),
('sscanf(buf, "%8lf", &v)', True),
('sscanf(buf, "%*f")', True),
('sscanf(buf, "%.2f", &v)', True),
# Mixed formats, including a ';' inside the format string
('sscanf(buf, "%d,%f", &a, &b)', True),
('sscanf(buf, "%d;%f", &a, &b)', True),
('sscanf(buf, "a;b", &a); sscanf(buf, "%d", &b); printf("%f", v)', False),
# Escaped quotes, nested calls, a ')' inside the format, an unterminated call
('sscanf(buf, "\\"%f\\"", &v)', True),
('sscanf(f(x), "%f", &v)', True),
('sscanf(s.substr(0, 2).c_str(), "%f", &v)', True),
('sscanf(strtok(s, ","), "%f", &v)', True),
('sscanf(strtok(s, ","), "%d", &v)', False),
('sscanf(buf, "%d)", &a); g("%f")', False),
('sscanf(buf, "%d", &a; g("%f")', False),
('sscanf(buf, "%f"', True),
# fscanf and std::sscanf
('fscanf(fp, "%f", &v)', True),
('std::sscanf(buf, "%f", &v)', True),
# Multi-line
('sscanf(buf,\n"%f", &v)', True),
# No float format
('sscanf(buf, "%d", &v)', False),
('sscanf(buf, "%s", s)', False),
# printf not scanf
('printf("%f", val)', False),
# %f in a different statement after scanf
('sscanf(buf, "%d", &x); printf("%f", val);', False),
# scanf %f in comment only
('// sscanf(buf, "%f", &v)\nsscanf(buf, "%d", &x)', False),
('/* sscanf(buf, "%f") */\nsscanf(buf, "%d", &x)', False),
],
)
def test_lambdas_use_scanf_float(src: str, expected: bool) -> None:
"""Test scanf float detection in lambda source."""
config: ConfigType = {"test": [Lambda(src)]}
assert lambdas_use_scanf_float(config) is expected
def test_lambdas_use_scanf_float_no_lambdas() -> None:
"""Test with config containing no lambdas."""
config: ConfigType = {"key": "value", "list": [1, 2]}
assert lambdas_use_scanf_float(config) is False
def test_lambdas_use_scanf_float_nested() -> None:
"""Test detection in deeply nested config."""
config: ConfigType = {"a": {"b": {"c": [Lambda('sscanf(buf, "%f", &v)')]}}}
assert lambdas_use_scanf_float(config) is True
def test_includes_use_scanf_float(setup_core: Path) -> None:
"""A float scanf in an includes: or includes_c: file is found; comments and non-source files are ignored."""
(setup_core / "parse.h").write_text(
'static void f(const char *b) { float v; sscanf(b, "%f", &v); }'
)
(setup_core / "lib").mkdir()
(setup_core / "lib" / "int.h").write_text(
'static int g(const char *b) { int v; sscanf(b, "%d", &v); return v; }'
)
(setup_core / "lib" / "notes.txt").write_text('sscanf(b, "%f", &v)')
(setup_core / "commented.h").write_text(
'// sscanf(b, "%f", &v)\n/* sscanf(b, "%g", &v) */'
)
assert (
includes_use_scanf_float({CONF_ESPHOME: {CONF_INCLUDES: ["parse.h"]}}) is True
)
assert (
includes_use_scanf_float({CONF_ESPHOME: {CONF_INCLUDES_C: ["parse.h"]}}) is True
)
assert includes_use_scanf_float({CONF_ESPHOME: {CONF_INCLUDES: ["lib"]}}) is False
assert (
includes_use_scanf_float(
{CONF_ESPHOME: {CONF_INCLUDES: ["commented.h", "<cstdio>"]}}
)
is False
)
assert includes_use_scanf_float({}) is False
def test_keep_float_scanf() -> None:
"""Explicit values win; unset follows whether user code scans a float."""
float_config: ConfigType = {
CONF_ESPHOME: {"on_boot": [Lambda('sscanf(b, "%f", &v)')]}
}
int_config: ConfigType = {
CONF_ESPHOME: {"on_boot": [Lambda('sscanf(b, "%d", &v)')]}
}
assert keep_float_scanf(True, int_config, "x", "y") is True
assert keep_float_scanf(False, float_config, "x", "y") is False
assert keep_float_scanf(False, int_config, "x", "y") is False
assert keep_float_scanf(None, float_config, "x", "y") is True
assert keep_float_scanf(None, int_config, "x", "y") is False
def test_external_components_use_scanf_float(setup_core: Path) -> None:
"""Sources of components outside the esphome package are scanned; in-tree ones are skipped."""
def manifest(package_dir: Path, *sources: str) -> MagicMock:
m = MagicMock()
m.is_platform_component = False
m.module.__file__ = str(package_dir / "__init__.py")
m.resources = [FileResource("pkg", name) for name in sources]
return m
(setup_core / "ext_float").mkdir()
(setup_core / "ext_float" / "parse.cpp").write_text(
'void f(const char *b) { float v; sscanf(b, "%f", &v); }'
)
(setup_core / "ext_int").mkdir()
(setup_core / "ext_int" / "parse.cpp").write_text(
'int g(const char *b) { int v; sscanf(b, "%d", &v); return v; }'
)
lookup = {
"ext_float": manifest(setup_core / "ext_float", "parse.cpp"),
"ext_int": manifest(setup_core / "ext_int", "parse.cpp"),
"sensor": manifest(CORE_COMPONENTS_PATH / "sensor", "sensor.cpp"),
"esphome": manifest(CORE_COMPONENTS_PATH.parent / "core", "config.cpp"),
}
with patch("esphome.config.get_component", side_effect=lookup.get):
CORE.config = {"esphome": {}, "sensor": {}, "ext_int": {}}
assert external_components_use_scanf_float(CORE.config) is False
CORE.config = {"sensor": {}, "ext_float": {}}
assert external_components_use_scanf_float(CORE.config) is True
assert user_code_uses_scanf_float(CORE.config) is True
+14
View File
@@ -1,6 +1,7 @@
"""Unit tests for esphome.config_helpers module."""
from collections.abc import Callable
from pathlib import Path
from unittest.mock import patch
import pytest
@@ -12,6 +13,7 @@ from esphome.config_helpers import (
filter_source_files_from_platform,
frameworks_for_platforms,
get_logger_level,
iter_include_files,
)
from esphome.const import (
CONF_LEVEL,
@@ -209,3 +211,15 @@ def test_filter_source_files_from_platform_unknown_pair() -> None:
{"a.cpp": {PlatformFramework.ESP32_ARDUINO}}
)
assert func() == []
def test_iter_include_files(setup_core: Path) -> None:
"""Files yield their name, directory members keep the directory prefix, system headers are skipped."""
(setup_core / "one.h").write_text("")
(setup_core / "lib").mkdir()
(setup_core / "lib" / "two.h").write_text("")
files = {
basename.as_posix()
for _, basename in iter_include_files(["one.h", "lib", "<cstdio>"])
}
assert files == {"one.h", "lib/two.h"}
+25 -4
View File
@@ -866,7 +866,8 @@ class TestSharedProgmemArray:
c = cg.shared_progmem_array("table", ct.uint8, [4])
assert a is b
assert str(a) != str(c)
assert sum("PROGMEM" in str(st) for st in CORE.global_statements) == 2
late = CORE.global_statements_after_includes
assert sum("PROGMEM" in str(st) for st in late) == 2
assert not any("PROGMEM" in str(st) for st in CORE.main_statements)
def test_share_false_always_emits_a_new_array(self) -> None:
@@ -874,7 +875,8 @@ class TestSharedProgmemArray:
a = cg.shared_progmem_array("table", ct.uint8, [1], share=False)
b = cg.shared_progmem_array("table", ct.uint8, [1], share=False)
assert str(a) != str(b)
assert sum("PROGMEM" in str(st) for st in CORE.global_statements) == 2
late = CORE.global_statements_after_includes
assert sum("PROGMEM" in str(st) for st in late) == 2
def test_is_static_pointer_only_for_placement_new(self) -> None:
CORE.config = {}
@@ -910,7 +912,9 @@ class TestSharedProgmemArray:
CORE.config = {}
cg.shared_progmem_array("table", ct.uint8, [1], constexpr=False)
(statement,) = (
str(st) for st in CORE.global_statements if "PROGMEM" in str(st)
str(st)
for st in CORE.global_statements_after_includes
if "PROGMEM" in str(st)
)
assert statement.startswith(
"ESPHOME_FLASH_CONSTINIT static uint8_t const table[] PROGMEM = {1}"
@@ -948,7 +952,7 @@ class TestProgmemString:
assert a is b
assert str(a) == "progmem_str"
assert str(c) == "progmem_str_2"
assert [str(st) for st in CORE.global_statements] == [
assert [str(st) for st in CORE.global_statements_after_includes] == [
'static constexpr char progmem_str[] PROGMEM = "level";',
'static constexpr char progmem_str_2[] PROGMEM = "say \\042hi\\042";',
]
@@ -957,6 +961,7 @@ class TestProgmemString:
self._set_platform(PLATFORM_ESP32)
assert str(cg.shared_progmem_string('say "hi"')) == '"say \\042hi\\042"'
assert not CORE.global_statements
assert not CORE.global_statements_after_includes
def test_extern_progmem_array_is_an_externally_linked_global() -> None:
@@ -967,6 +972,22 @@ def test_extern_progmem_array_is_an_externally_linked_global() -> None:
]
def test_static_function_declares_early_and_defines_after_includes() -> None:
fn = cg.static_function("esphome__act__fn", ct.void, [(ct.int32, "x")], ["return;"])
assert str(fn) == "esphome__act__fn"
assert [str(st) for st in CORE.global_statements] == [
"static void esphome__act__fn(int32_t x);"
]
assert [str(st) for st in CORE.global_statements_after_includes] == [
"static void esphome__act__fn(int32_t x) {\n return;\n}"
]
def test_add_global_after_includes_rejects_non_statements() -> None:
with pytest.raises(ValueError, match="must be expression or statement"):
CORE.add_global_after_includes("not a statement")
class TestEnumOptions:
def test_class_enum_with_prefix(self) -> None:
mode = cg.MockObj("esphome::foo", "::").enum("Mode", is_class=True)
+3 -9
View File
@@ -215,9 +215,8 @@ def test_convert_pio_libs_arduino_framework_passes_empty_managed(
path=tmp_path / "other-lib",
)
def fake_generate_idf_components(libraries, managed=None, provides=None):
def fake_generate_idf_components(libraries, managed=None):
captured["managed"] = managed
captured["provides"] = provides
return [converted]
monkeypatch.setattr(
@@ -227,8 +226,6 @@ def test_convert_pio_libs_arduino_framework_passes_empty_managed(
result = _convert_pio_libs(tmp_path / "platformio.ini", "arduino")
assert captured["managed"] == set()
# Bundled Arduino names are left to arduino-esp32, as in the real build
assert captured["provides"]("Wire") and not captured["provides"]("Adafruit BusIO")
assert result == {
"esphome/other-lib": {"override_path": str(tmp_path / "other-lib")}
}
@@ -245,7 +242,7 @@ def test_convert_pio_libs_arduino_idf_6_passes_manifest_deps(
captured: dict[str, set[str] | None] = {}
def fake_generate_idf_components(libraries, managed=None, provides=None):
def fake_generate_idf_components(libraries, managed=None):
captured["managed"] = managed
return []
@@ -268,9 +265,8 @@ def test_convert_pio_libs_espidf_framework_passes_manifest_deps(
captured: dict[str, set[str] | None] = {}
def fake_generate_idf_components(libraries, managed=None, provides=None):
def fake_generate_idf_components(libraries, managed=None):
captured["managed"] = managed
captured["provides"] = provides
return []
monkeypatch.setattr(
@@ -280,8 +276,6 @@ def test_convert_pio_libs_espidf_framework_passes_manifest_deps(
result = _convert_pio_libs(tmp_path / "platformio.ini", "espidf")
assert captured["managed"] == _esphome_manifest_deps()
# ESP-IDF bundles no Arduino libraries
assert captured["provides"] is None
assert "esphome/noise-c" in captured["managed"]
assert result == {}
+122 -562
View File
@@ -4,13 +4,11 @@ Covers the shared download/parse/resolve/dependency-walk paths in
``esphome.platformio.library`` directly (the ESP-IDF and Zephyr backends are
exercised in their own test modules)."""
from collections.abc import Collection
import json
import logging
from pathlib import Path
from unittest.mock import Mock
from platformio.package.exception import UnknownPackageError
import pytest
from esphome.core import CORE, EsphomeError, Library
@@ -263,37 +261,23 @@ def test_make_registry_client_creates_http_cache_dir(
lib._make_registry_client()
def _patch_registry_resolve(
monkeypatch: pytest.MonkeyPatch,
*,
unknown: Collection[str] = (),
owner: str | None = None,
) -> list[tuple[str | None, str]]:
"""Stub the registry lookup so tests never touch the network; returns the
(owner, name) pairs looked up. ``unknown`` names miss; ``owner`` is what the
registry answers for an owner-less name."""
asked: list[tuple[str | None, str]] = []
def resolve(requested_owner, pkgname, requirements):
asked.append((requested_owner, pkgname))
if pkgname in unknown:
raise UnknownPackageError(pkgname)
return _resolution(requested_owner or owner, pkgname)
monkeypatch.setattr(lib, "_resolve_registry_version", resolve)
return asked
def _patch_registry_resolve(monkeypatch: pytest.MonkeyPatch) -> None:
"""Stub the registry lookup so tests never touch the network."""
monkeypatch.setattr(
lib,
"_resolve_registry_version",
lambda owner, pkgname, requirements: (
owner,
pkgname,
"1.0.0",
f"http://x/{pkgname}.tar.gz",
None,
),
)
def _resolution(owner, pkgname):
"""What the stubbed registry answers for a package."""
return owner, pkgname, "1.0.0", f"http://x/{pkgname}.tar.gz", None
def _patch_download_with_manifests(
monkeypatch, tmp_path, manifests, *, properties=(), unknown=(), owner=None
):
"""Fake ConvertedLibrary.download to materialize canned manifests on disk;
returns the registry lookups made (see _patch_registry_resolve)."""
def _patch_download_with_manifests(monkeypatch, tmp_path, manifests, *, properties=()):
"""Fake ConvertedLibrary.download to materialize canned manifests on disk."""
def fake_download(self, force=False, salt="", namespace=""):
self.path = tmp_path / self.get_require_name()
@@ -304,7 +288,7 @@ def _patch_download_with_manifests(
(self.path / "library.json").write_text(json.dumps(manifests[self.name]))
monkeypatch.setattr(ConvertedLibrary, "download", fake_download)
return _patch_registry_resolve(monkeypatch, unknown=unknown, owner=owner)
_patch_registry_resolve(monkeypatch)
def test_wave_requirement_growth_defers_the_superseded_download(tmp_path, monkeypatch):
@@ -359,22 +343,18 @@ def test_convert_libraries_parses_library_properties(tmp_path, monkeypatch):
assert emitted[0].data["version"] == "1.0"
def test_convert_libraries_fetches_dependency_without_version(tmp_path, monkeypatch):
# A dependency entry lacking a version resolves by name alone, as PlatformIO did.
asked = _patch_download_with_manifests(
def test_convert_libraries_skips_dependency_without_version(tmp_path, monkeypatch):
# A dependency entry lacking a version is malformed and silently skipped.
_patch_download_with_manifests(
monkeypatch,
tmp_path,
{
"esphome/A": {"name": "A", "dependencies": [{"name": "C"}]},
"C": {"name": "C"},
},
{"esphome/A": {"name": "A", "dependencies": [{"name": "C"}]}},
)
# No version on the top-level spec exercises the "no requirement" path too.
top = convert_libraries([Library("esphome/A", None, None)], _backend())
assert [d.name for d in top[0].dependencies] == ["C"]
assert asked == [("esphome", "A"), (None, "C")]
assert top[0].dependencies == []
def test_convert_libraries_handles_unparsable_dependency_version(tmp_path, monkeypatch):
@@ -1033,24 +1013,20 @@ def test_convert_libraries_malformed_transitive_dep_skips(
assert "Skipping dependency esphome/B: malformed manifest" in caplog.text
def test_walk_resolves_properties_only_depends(tmp_path, monkeypatch) -> None:
"""A library.properties depends= line is a dependency list: a bundled name
goes to the backend, the rest resolve by name."""
asked = _patch_download_with_manifests(
def test_walk_warns_for_properties_only_depends(
tmp_path, monkeypatch, caplog: pytest.LogCaptureFixture
) -> None:
"""A manifest declaring dependencies only as library.properties depends=
warns in the shared walk, so every backend reports the drop."""
_patch_download_with_manifests(
monkeypatch,
tmp_path,
{
"esphome/A": "name=A\nversion=1.0\ndepends=Wire, SPI\n",
"SPI": {"name": "SPI"},
},
{"esphome/A": "name=A\nversion=1.0\ndepends=Wire, SPI\n"},
properties=("esphome/A",),
)
backend = _backend(provides=lambda name: name == "Wire")
top = convert_libraries([Library("esphome/A", "1.0.0", None)], backend)
assert [d.name for d in top[0].dependencies] == ["SPI"]
assert "depends" not in top[0].data
assert backend.provided_requests == {"Wire"}
assert (None, "SPI") in asked
caplog.set_level("INFO")
convert_libraries([Library("esphome/A", "1.0.0", None)], _backend())
assert "declares dependencies via library.properties" in caplog.text
def test_walk_warns_for_nonplatform_invalid_library(
@@ -1117,12 +1093,12 @@ def test_source_kind_map_shape() -> None:
assert SOURCE_KIND_FOR_SUFFIX[".C++"] == "cxx"
def test_versionless_platform_filtered_dependency_is_not_fetched(
tmp_path, monkeypatch
def test_versionless_platform_filtered_dependency_stays_quiet(
tmp_path, monkeypatch, caplog: pytest.LogCaptureFixture
) -> None:
"""A version-less dependency the platform filter excludes is deliberately
absent, not fetched by name."""
asked = _patch_download_with_manifests(
"""A version-less dependency the platform filter excludes is
deliberately absent, not a drop to warn about."""
_patch_download_with_manifests(
monkeypatch,
tmp_path,
{
@@ -1132,52 +1108,59 @@ def test_versionless_platform_filtered_dependency_is_not_fetched(
}
},
)
top = convert_libraries([Library("esphome/A", None, None)], _backend())
assert top[0].dependencies == []
assert (None, "Hash") not in asked
convert_libraries([Library("esphome/A", None, None)], _backend())
assert "has no version to resolve" not in caplog.text
def test_versionless_ignored_dependency_is_not_fetched(tmp_path, monkeypatch) -> None:
def test_versionless_ignored_dependency_stays_quiet(
tmp_path, monkeypatch, caplog: pytest.LogCaptureFixture
) -> None:
"""A lib_ignore'd version-less dependency is deliberately excluded, not
fetched by name."""
asked = _patch_download_with_manifests(
a drop; no reconciliation warning."""
_patch_download_with_manifests(
monkeypatch,
tmp_path,
{"esphome/A": {"name": "A", "dependencies": [{"name": "Hash"}]}},
)
CORE.platformio_options = {"lib_ignore": ["Hash"]}
top = convert_libraries([Library("esphome/A", None, None)], _backend())
assert top[0].dependencies == []
assert (None, "Hash") not in asked
convert_libraries([Library("esphome/A", None, None)], _backend())
assert "has no version to resolve" not in caplog.text
def test_versionless_dependency_registry_miss_is_skipped(
def test_versionless_dependency_without_provider_warns(
tmp_path, monkeypatch, caplog: pytest.LogCaptureFixture
) -> None:
"""A name-only dependency the registry does not know is skipped with one
warning naming the requester, and the requester still converts; a listed
library the registry does not know still fails the build."""
"""A backend whose tree could supply the name warns on the drop; one
without provides() can never act on it, so it stays at debug."""
_patch_download_with_manifests(
monkeypatch,
tmp_path,
{
"esphome/A": {
"name": "A",
# The duplicate entry is one node, so it warns once
# The duplicate entry warns once (reconciliation dedup)
"dependencies": [{"name": "Hash"}, {"name": "Hash"}],
}
},
unknown={"Hash"},
)
top = convert_libraries([Library("esphome/A", None, None)], _backend())
assert top[0].dependencies == []
convert_libraries(
[Library("esphome/A", None, None)], _backend(provides=lambda name: False)
)
assert (
caplog.text.count("Dependency Hash of esphome/A was not resolved; skipping")
caplog.text.count(
"Hash of esphome/A has no version to resolve and nothing provides it"
)
== 1
)
with pytest.raises(UnknownPackageError):
convert_libraries([Library("Hash", None, None)], _backend())
caplog.clear()
with caplog.at_level(logging.DEBUG):
convert_libraries([Library("esphome/A", None, None)], _backend())
records = [
r
for r in caplog.records
if "has no version to resolve and nothing provides it" in r.message
]
assert records and all(r.levelno == logging.DEBUG for r in records)
def test_url_version_dependency_is_not_substituted_by_provides(
@@ -1208,525 +1191,102 @@ def test_url_version_dependency_is_not_substituted_by_provides(
assert any("o/hash" in n.lower() for n in emitted)
def test_versionless_owner_qualified_dependency_is_fetched_despite_provides(
tmp_path, monkeypatch
def test_versionless_owner_qualified_dependency_warns_despite_provides(
tmp_path, monkeypatch, caplog: pytest.LogCaptureFixture
) -> None:
"""An owner-qualified version-less dependency names a registry package,
so provides() does not stand in for it."""
asked = _patch_download_with_manifests(
"""An owner-qualified version-less dependency is not satisfied by
provides(); it must still warn."""
_patch_download_with_manifests(
monkeypatch,
tmp_path,
{
"esphome/A": {
"name": "A",
"dependencies": [{"name": "Wire", "owner": "Foo"}],
},
"Foo/Wire": {"name": "Wire"},
}
},
)
backend = _backend(provides=lambda name: name == "Wire")
top = convert_libraries([Library("esphome/A", None, None)], backend)
assert [d.name for d in top[0].dependencies] == ["Foo/Wire"]
assert ("Foo", "Wire") in asked
assert not backend.provided_requests
convert_libraries(
[Library("esphome/A", None, None)],
_backend(provides=lambda name: name == "Wire"),
)
assert "Wire of esphome/A has no version to resolve" in caplog.text
def test_versionless_provided_dependency_stays_quiet(
tmp_path, monkeypatch, caplog: pytest.LogCaptureFixture
) -> None:
"""An owner-less version-less dependency the backend provides is added
by the backend after emit; no reconciliation warning."""
_patch_download_with_manifests(
monkeypatch,
tmp_path,
{"esphome/A": {"name": "A", "dependencies": [{"name": "Wire"}]}},
)
convert_libraries(
[Library("esphome/A", None, None)],
_backend(provides=lambda name: name == "Wire"),
)
assert "has no version to resolve" not in caplog.text
def test_versionless_dependency_requested_top_level_stays_quiet(
tmp_path, monkeypatch, caplog: pytest.LogCaptureFixture
) -> None:
"""A version-less dependency the config also requests top-level is in
the build; no drop warning even without a provides backend."""
_patch_download_with_manifests(
monkeypatch,
tmp_path,
{
"esphome/A": {"name": "A", "dependencies": [{"name": "Hash"}]},
"Hash": {"name": "Hash"},
},
)
convert_libraries(
[Library("esphome/A", None, None), Library("Hash", None, None)],
_backend(),
)
assert "has no version to resolve" not in caplog.text
def test_versionless_url_ish_dependency_name_warns_cleanly(
tmp_path, monkeypatch, caplog: pytest.LogCaptureFixture
) -> None:
"""A malformed URL-ish dependency name is skipped with a warning, never
"""A malformed URL-ish dependency name falls to the drop warning, never
a RuntimeError out of the key parser."""
_patch_download_with_manifests(
monkeypatch,
tmp_path,
{"esphome/A": {"name": "A", "dependencies": [{"name": "file://"}]}},
)
top = convert_libraries([Library("esphome/A", None, None)], _backend())
assert top[0].dependencies == []
assert "Skipping dependency file:// of esphome/A" in caplog.text
def test_versionless_dependency_matching_resolved_manifest_name_is_wired(
tmp_path, monkeypatch, caplog: pytest.LogCaptureFixture
) -> None:
"""A bare name no package name matches is satisfied by a listed component's
manifest name (a fork published under another package name)."""
asked = _patch_download_with_manifests(
monkeypatch,
tmp_path,
{
"esphome/A": {"name": "A", "dependencies": [{"name": "B"}]},
"esphome/B-fork": {"name": "B"},
},
convert_libraries(
[Library("esphome/A", None, None)], _backend(provides=lambda name: False)
)
with caplog.at_level(logging.INFO):
top = convert_libraries(
[Library("esphome/A", None, None), Library("esphome/B-fork", None, None)],
_backend(),
)
assert [d.name for d in top[0].dependencies] == ["esphome/B-fork"]
assert asked == [("esphome", "A"), ("esphome", "B-fork")]
assert (
"Dependency B of esphome/A satisfied by esphome/B-fork through its manifest name"
"file:// of esphome/A has no version to resolve and nothing provides it"
in caplog.text
)
def test_dependency_naming_the_requester_itself_is_ignored(
tmp_path, monkeypatch
) -> None:
"""A manifest that lists its own name as a dependency neither fetches a
second copy of itself nor gains a self edge."""
asked = _patch_download_with_manifests(
monkeypatch,
tmp_path,
{
"esphome/A": "name=A\ndepends=a, esphome/A\n",
# No manifest name: the requested and resolved package names still
# identify it, with or without the owner
"esphome/C": {
"dependencies": [{"name": "C"}, {"name": "C", "owner": "esphome"}]
},
},
properties=("esphome/A",),
)
top = convert_libraries(
[Library("esphome/A", "1.0.0", None), Library("esphome/C", "1.0.0", None)],
_backend(),
)
assert [c.dependencies for c in top] == [[], []]
assert asked == [("esphome", "A"), ("esphome", "C")]
def test_versionless_dependency_settles_after_the_graph(
def test_versionless_dependency_matching_resolved_manifest_name_stays_quiet(
tmp_path, monkeypatch, caplog: pytest.LogCaptureFixture
) -> None:
"""The issue 21308 shape: a library.properties depends= naming a library the
config lists owner-qualified is wired to that one package, case-insensitively
and whatever constraint it carries, without a second node for the bare name
and ahead of a same-named bundled copy, as on the Arduino backend."""
asked = _patch_download_with_manifests(
monkeypatch,
tmp_path,
{
"adafruit/Adafruit GFX Library": (
"name=Adafruit GFX Library\ndepends=adafruit busio (>=1.0)\n"
),
"adafruit/Adafruit BusIO": {"name": "Adafruit BusIO"},
},
properties=("adafruit/Adafruit GFX Library",),
)
backend = _backend(provides=lambda name: name == "adafruit busio")
top = convert_libraries(
[
Library("adafruit/Adafruit GFX Library", "^1.13.0", None),
Library("adafruit/Adafruit BusIO", "^1.17.4", None),
],
backend,
)
assert [d.name for d in top[0].dependencies] == ["adafruit/Adafruit BusIO"]
assert len(asked) == 2
assert not backend.provided_requests
assert "multiple names" not in caplog.text
def test_by_name_node_folds_into_a_later_pin_of_the_same_package(
tmp_path, monkeypatch, caplog: pytest.LogCaptureFixture
) -> None:
"""A library fetched by name alone may pull in a manifest that pins the same
package owner-qualified; the by-name node folds into that one instead of
converting the package twice."""
# Manifests are keyed by the resolved name, so both B nodes read the same one:
# B pins C, and C pins owner/B, which arrives after B was fetched by name
manifests = {
"esphome/A": {"name": "A", "dependencies": [{"name": "B"}]},
"esphome/B": {"name": "B", "dependencies": {"esphome/C": "1.0.0"}},
"esphome/C": {"name": "C", "dependencies": {"esphome/B": "1.0.0"}},
}
_patch_download_with_manifests(monkeypatch, tmp_path, manifests, owner="esphome")
emitted: list[str] = []
top = convert_libraries(
[Library("esphome/A", None, None)],
_backend(emit=lambda c: emitted.append(c.name)),
)
assert [d.name for d in top[0].dependencies] == ["esphome/B"]
assert emitted.count("esphome/B") == 1
assert "multiple names" not in caplog.text
def test_two_spellings_of_a_name_in_one_round_share_a_node(
tmp_path, monkeypatch, caplog: pytest.LogCaptureFixture
) -> None:
"""Requesters spelling one unlisted name differently get a single by-name
fetch and the same node."""
asked = _patch_download_with_manifests(
monkeypatch,
tmp_path,
{
"esphome/A": {"name": "A", "dependencies": [{"name": "Foo"}]},
"esphome/B": {"name": "B", "dependencies": [{"name": "foo"}]},
"Foo": {"name": "Foo"},
},
)
top = convert_libraries(
[Library("esphome/A", None, None), Library("esphome/B", None, None)],
_backend(),
)
assert [d.name for d in top[0].dependencies] == ["Foo"]
assert [d.name for d in top[1].dependencies] == ["Foo"]
assert asked.count((None, "Foo")) + asked.count((None, "foo")) == 1
assert "multiple names" not in caplog.text
def test_parse_properties_depends_reads_names_and_drops_constraints(
caplog: pytest.LogCaptureFixture,
) -> None:
"""Arduino constraints are not PlatformIO syntax, so only the names survive;
an unclosed parenthesis loses nothing more than the constraint."""
with caplog.at_level(logging.DEBUG):
entries = lib.parse_properties_depends(
"Adafruit BusIO, Foo (>=1.2), ,Bar( <2 ), Baz (=1.0, Qux ()"
)
levels = {
r.message.removeprefix("Ignoring version constraint on depends="): r.levelno
for r in caplog.records
if "Ignoring version constraint" in r.message
}
# A lower bound is met by the current version; a cap or pin may not be
assert levels == {
"Foo (>=1.2)": logging.DEBUG,
"Bar( <2 )": logging.INFO,
"Baz (=1.0": logging.INFO,
"Qux ()": logging.INFO,
}
assert entries == [
{"name": "Adafruit BusIO"},
{"name": "Foo"},
{"name": "Bar"},
{"name": "Baz"},
{"name": "Qux"},
]
assert "Ignoring version constraint on depends=Foo (>=1.2)" in caplog.text
def test_by_name_node_pinned_later_reports_a_requirement_conflict(
tmp_path, monkeypatch
) -> None:
"""A node fetched by name that a later manifest pins under the same spelling
is resolved again against the pin; a conflict is an error, not a miss."""
"""A bare name satisfied by an owner-qualified component's manifest
name is not a drop."""
_patch_download_with_manifests(
monkeypatch,
tmp_path,
{
"esphome/A": {"name": "A", "dependencies": [{"name": "B"}]},
"B": {"name": "B", "dependencies": {"C": "1.0.0"}},
"C": {"name": "C", "dependencies": {"B": "^2.0"}},
"esphome/B": {"name": "B"},
},
)
def resolve(owner, pkgname, requirements):
if pkgname == "B" and requirements:
raise RuntimeError(f"No version of {pkgname} satisfies {requirements}")
return _resolution(owner, pkgname)
monkeypatch.setattr(lib, "_resolve_registry_version", resolve)
with pytest.raises(RuntimeError, match="No version of B satisfies"):
convert_libraries([Library("esphome/A", None, None)], _backend())
def test_by_name_fetch_yields_to_a_fork_pulled_in_later(
tmp_path, monkeypatch, caplog: pytest.LogCaptureFixture
) -> None:
"""A name fetched from the registry yields to a library another round
pins explicitly under the same manifest name, so only the fork is built."""
_patch_download_with_manifests(
monkeypatch,
tmp_path,
{
"esphome/A": {"name": "A", "dependencies": [{"name": "B"}]},
"esphome/X": {"name": "X", "dependencies": [{"name": "C"}]},
# The upstream B pulls in D, which nothing else needs; D asks for a
# name the registry does not know
"B": {"name": "B", "dependencies": {"esphome/D": "1.0.0"}},
"esphome/D": {"name": "D", "dependencies": [{"name": "Zed"}]},
"C": {"name": "C", "dependencies": {"custom/B-fork": "1.0.0"}},
"custom/B-fork": {"name": "B"},
},
unknown={"Zed"},
)
emitted: list[str] = []
top = convert_libraries(
[Library("esphome/A", None, None), Library("esphome/X", None, None)],
_backend(emit=lambda c: emitted.append(c.name)),
)
assert [d.name for d in top[0].dependencies] == ["custom/B-fork"]
assert sorted(emitted) == ["C", "custom/B-fork", "esphome/A", "esphome/X"]
# The miss belongs to a library that is not built, so it is not reported
assert "Zed" not in caplog.text
def test_git_source_replacing_a_registry_key_in_the_same_wave(
tmp_path, monkeypatch
) -> None:
"""A manifest in one wave naming a key of the same wave as a git source
defers that key to the next wave instead of tripping over its stale record."""
_patch_download_with_manifests(
monkeypatch,
tmp_path,
{
"o/A": {"name": "A", "dependencies": {"B": "https://github.com/o/B.git"}},
"o/B": {"name": "B"},
},
)
emitted: dict[str, ConvertedLibrary] = {}
top = convert_libraries(
[Library("o/A", "1.0.0", None), Library("o/B", "1.0.0", None)],
_backend(emit=lambda c: emitted.__setitem__(c.name, c)),
)
assert [c.name for c in top] == ["o/A", "o/B"]
assert isinstance(emitted["o/B"].source, GitSource)
assert [d.name for d in top[0].dependencies] == ["o/B"]
def test_owner_qualified_name_matches_the_resolved_owner(tmp_path, monkeypatch) -> None:
"""A version-less ``o/B`` is satisfied by a bare ``B`` pin the registry
resolved to owner ``o``, so the package is not fetched a second time."""
asked = _patch_download_with_manifests(
monkeypatch,
tmp_path,
{
"esphome/A": {
"name": "A",
"dependencies": [
{"name": "B", "version": "1.0.0"},
{"name": "B", "owner": "o"},
],
},
"o/B": {"name": "B"},
},
owner="o",
)
top = convert_libraries([Library("esphome/A", None, None)], _backend())
assert [d.name for d in top[0].dependencies] == ["o/B"]
assert asked.count((None, "B")) == 1
assert ("o", "B") not in asked
def test_two_spellings_fetched_by_name_fold_into_one_node(
tmp_path, monkeypatch, caplog: pytest.LogCaptureFixture
) -> None:
"""Version-less ``B`` and ``o/B`` from two requesters resolve to one
package; only one copy is built and both requesters depend on it."""
_patch_download_with_manifests(
monkeypatch,
tmp_path,
{
"esphome/A": {"name": "A", "dependencies": [{"name": "B"}]},
"esphome/X": {"name": "X", "dependencies": [{"name": "B", "owner": "o"}]},
"o/B": {"name": "B"},
},
owner="o",
)
emitted: list[str] = []
top = convert_libraries(
[Library("esphome/A", None, None), Library("esphome/X", None, None)],
_backend(emit=lambda c: emitted.append(c.name)),
)
assert emitted.count("o/B") == 1
kept = top[0].dependencies[0]
assert top[1].dependencies == [kept]
assert "multiple names" not in caplog.text
def test_registry_miss_resolves_again_as_a_git_source(tmp_path, monkeypatch) -> None:
"""A key the registry missed is resolved again when a later manifest names
the same key as a git source."""
_patch_download_with_manifests(
monkeypatch,
tmp_path,
{
"esphome/A": {"name": "A", "dependencies": [{"name": "B", "owner": "o"}]},
"esphome/X": {"name": "X", "dependencies": [{"name": "C"}]},
"C": {"name": "C", "dependencies": {"B": "https://github.com/o/B.git"}},
"o/B": {"name": "B"},
},
unknown={"B"},
)
emitted: dict[str, ConvertedLibrary] = {}
top = convert_libraries(
[Library("esphome/A", None, None), Library("esphome/X", None, None)],
_backend(emit=lambda c: emitted.__setitem__(c.name, c)),
)
assert [d.name for d in top[0].dependencies] == ["o/B"]
assert isinstance(emitted["o/B"].source, GitSource)
assert [d.name for d in emitted["C"].dependencies] == ["o/B"]
def test_normalize_dependencies_splits_an_embedded_owner() -> None:
"""``owner/Name`` as one string, in a list or a dict entry, normalizes like
the dict form; a URL or a deeper path is left alone."""
assert lib.normalize_dependencies(
["o/B", {"name": "o/C", "version": "1.0"}, "x://y", "a/b/c", "../escape"],
"libx",
) == [
{"name": "B", "owner": "o"},
{"name": "C", "owner": "o", "version": "1.0"},
{"name": "x://y"},
{"name": "a/b/c"},
{"name": "../escape"},
]
assert lib.normalize_dependencies("o/B, D", "libx") == [
{"name": "B", "owner": "o"},
{"name": "D"},
]
def test_owner_qualified_string_dependency_is_fetched_and_wired(
tmp_path, monkeypatch
) -> None:
"""``"dependencies": ["o/B"]`` fetches ``o/B`` and wires the requester to it."""
asked = _patch_download_with_manifests(
monkeypatch,
tmp_path,
{
"esphome/A": {"name": "A", "dependencies": ["o/B"]},
"o/B": {"name": "B"},
},
)
top = convert_libraries([Library("esphome/A", None, None)], _backend())
assert [d.name for d in top[0].dependencies] == ["o/B"]
assert asked == [("esphome", "A"), ("o", "B")]
def test_empty_version_is_settled_like_a_missing_one(
tmp_path, monkeypatch, caplog: pytest.LogCaptureFixture
) -> None:
"""``{"Foo": ""}`` carries no version either, so it is matched, fetched by
name or reported like a version-less entry rather than failing or vanishing."""
asked = _patch_download_with_manifests(
monkeypatch,
tmp_path,
{
"esphome/A": {"name": "A", "dependencies": {"Foo": "", "Bar": ""}},
"Bar": {"name": "Bar"},
},
unknown={"Foo"},
)
top = convert_libraries([Library("esphome/A", None, None)], _backend())
assert [d.name for d in top[0].dependencies] == ["Bar"]
assert (None, "Foo") in asked
assert "Dependency Foo of esphome/A was not resolved; skipping" in caplog.text
def test_bundled_name_is_satisfied_by_a_replacement_pulled_in_later(
tmp_path, monkeypatch
) -> None:
"""A name the framework bundles is still wired to an explicit replacement
of the same manifest name that a later round pulls in, as it would be had
the replacement been listed up front."""
_patch_download_with_manifests(
monkeypatch,
tmp_path,
{
"esphome/A": {"name": "A", "dependencies": [{"name": "Ethernet"}]},
"esphome/X": {"name": "X", "dependencies": [{"name": "C"}]},
"C": {"name": "C", "dependencies": {"vendor/Ethernet": "1.0.0"}},
"vendor/Ethernet": {"name": "Ethernet"},
},
)
backend = _backend(provides=lambda name: name == "Ethernet")
top = convert_libraries(
[Library("esphome/A", None, None), Library("esphome/X", None, None)],
backend,
)
assert [d.name for d in top[0].dependencies] == ["vendor/Ethernet"]
def test_missed_name_is_satisfied_by_a_library_pulled_in_later(
tmp_path, monkeypatch, caplog: pytest.LogCaptureFixture
) -> None:
"""A bare name the registry does not know is wired to a library a later
round pulls in under another package name but the same manifest name."""
asked = _patch_download_with_manifests(
monkeypatch,
tmp_path,
{
"esphome/A": {"name": "A", "dependencies": [{"name": "B"}]},
"esphome/X": {"name": "X", "dependencies": [{"name": "C"}]},
# Fetched by name in the same round as the miss; pins the fork
"C": {"name": "C", "dependencies": {"owner/B-fork": "1.0.0"}},
"owner/B-fork": {"name": "B"},
},
unknown={"B"},
)
with caplog.at_level(logging.INFO):
top = convert_libraries(
[Library("esphome/A", None, None), Library("esphome/X", None, None)],
_backend(),
)
assert [d.name for d in top[0].dependencies] == ["owner/B-fork"]
assert asked.count((None, "B")) == 1
assert (
"Dependency B of esphome/A satisfied by owner/B-fork through its manifest name"
in caplog.text
)
assert "skipping" not in caplog.text
def test_by_name_request_does_not_make_an_existing_node_optional(
tmp_path, monkeypatch, caplog: pytest.LogCaptureFixture
) -> None:
"""A version-less request for a node the walk already created (here one
skipped for its platform) neither re-fetches it nor turns a later miss
into a warning."""
asked = _patch_download_with_manifests(
monkeypatch,
tmp_path,
{
"esphome/A": {
"name": "A",
"dependencies": [{"name": "X", "version": "1.0"}, {"name": "X"}],
},
"X": {"name": "X", "platforms": "espressif8266"},
},
)
top = convert_libraries([Library("esphome/A", None, None)], _backend())
assert top[0].dependencies == []
assert asked == [("esphome", "A"), (None, "X")]
assert "not in the registry" not in caplog.text
def test_provided_dependency_asks_the_backend_once_per_name(
tmp_path, monkeypatch
) -> None:
"""A bundled name is recorded for the backend and never fetched; two
requesters cost one provides() call, so a backend warning there is not
repeated. A pinned request on the same name shares the memo."""
asked = _patch_download_with_manifests(
monkeypatch,
tmp_path,
{
"esphome/A": {"name": "A", "dependencies": [{"name": "Wire"}]},
"esphome/B": {"name": "B", "dependencies": {"Wire": "*"}},
},
)
calls: list[str] = []
def provides(name: str) -> bool:
calls.append(name)
return name == "Wire"
backend = _backend(provides=provides)
convert_libraries(
[Library("esphome/A", None, None), Library("esphome/B", None, None)],
backend,
_backend(),
)
assert calls == ["Wire"]
assert backend.provided_requests == {"Wire"}
assert (None, "Wire") not in asked
assert "has no version to resolve" not in caplog.text
def test_convert_libraries_symlink_url_resolves_as_local(setup_core: Path) -> None: