diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 4dac899a41..d9507c5938 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -1226,8 +1226,20 @@ jobs: if [ -n "$compile_csv" ]; then # Run compilation with grouping and isolation python3 script/test_build_components.py -e compile -c "$compile_csv" -f --isolate "$directly_changed_csv" + if [[ "${{ matrix.batch.check_idf_py }}" == "true" ]]; then + # The real idf.py must find nothing to configure or build in a + # tree built above; catches drift on ESP-IDF bumps. + echo "Checking the native ESP-IDF build matches idf.py" + python3 script/check_idf_py_equivalence.py + fi else echo "All components in this batch are validate-only -- skipping compile stage." + if [[ "${{ matrix.batch.check_idf_py }}" == "true" ]]; then + # determine-jobs and this step disagree on what compiles; fail + # rather than let the check run nowhere. + echo "::error::This batch was picked for the idf.py check but compiled nothing" + exit 1 + fi fi - name: Print ccache statistics diff --git a/esphome/analyze_memory/toolchain.py b/esphome/analyze_memory/toolchain.py index b64af5ea5b..7d8195a17b 100644 --- a/esphome/analyze_memory/toolchain.py +++ b/esphome/analyze_memory/toolchain.py @@ -37,7 +37,7 @@ def find_elf_path(build_path: Path) -> Path | None: """ name = build_path.name for candidate in ( - # Native ESP-IDF: idf.py writes build/.elf, which ESPHome copies + # Native ESP-IDF: the build writes build/.elf, which ESPHome copies # to build/firmware.elf (see espidf.toolchain.create_elf_copy) build_path / "build" / "firmware.elf", # PlatformIO diff --git a/esphome/build_helpers/tool_runner.py b/esphome/build_helpers/tool_runner.py new file mode 100644 index 0000000000..52cd6c937a --- /dev/null +++ b/esphome/build_helpers/tool_runner.py @@ -0,0 +1,180 @@ +"""Run a native build tool (cmake, ninja) and relay its output. + +Output is read from a pipe so it can be filtered here: a child that inherits +our stdout writes straight to the file descriptor, past any Python wrapper. +""" + +from __future__ import annotations + +import codecs +from contextlib import suppress +import logging +import os +from pathlib import Path +import re +import shutil +import subprocess +import sys +from typing import Any, TextIO + +from esphome.util import ANSI_ESCAPE, RedirectText, shlex_quote + +_LOGGER = logging.getLogger(__name__) + +# Windows code page identifier for UTF-8, as used by ``chcp 65001``. +UTF8_CODEPAGE = 65001 + +# Same pattern idf.py uses to spot ninja status lines (``is_progression``). +_PROGRESS = re.compile(r"^\[\d+/\d+\]|.*\(\d+ \%\)$") +_READ_SIZE = 65536 + + +def _get_kernel32() -> Any | None: + """Return the Windows kernel32 module, or None on any other platform.""" + if sys.platform != "win32": + return None + import ctypes + + return ctypes.windll.kernel32 + + +class Utf8Console: + """Keep an attached Windows console on UTF-8 while a build tool runs. + + esp_idf_size draws its table with Unicode box characters, and CMake + re-decodes a child's output with the console code page, which garbles the + table on any page but UTF-8. A console already on UTF-8 is left alone so + an overlapping build never records UTF-8 as the page to go back to. + """ + + def __init__(self, kernel32: Any | None) -> None: + self._kernel32 = kernel32 + self._codepages: tuple[int, int] | None = None + + def __enter__(self) -> None: + kernel32 = self._kernel32 + if kernel32 is None: + return + old_in = kernel32.GetConsoleCP() + old_out = kernel32.GetConsoleOutputCP() + # Both calls return 0 when no console is attached. + if not old_in or not old_out: + return + if old_in == UTF8_CODEPAGE and old_out == UTF8_CODEPAGE: + return + # Record first so a switch that fails part way is still undone. + self._codepages = (old_in, old_out) + kernel32.SetConsoleCP(UTF8_CODEPAGE) + kernel32.SetConsoleOutputCP(UTF8_CODEPAGE) + + def __exit__(self, *exc_info: object) -> None: + if self._codepages is None: + return + old_in, old_out = self._codepages + self._codepages = None + self._kernel32.SetConsoleCP(old_in) + self._kernel32.SetConsoleOutputCP(old_out) + + +def _fit_terminal(text: str) -> str: + """Elide the middle of ``text`` to fit the terminal, as idf.py does. + + A width of 0 (a pipe, the dashboard) leaves the text whole. + """ + width = shutil.get_terminal_size((0, 0)).columns + if not width: + return text + if width <= 3: + return "." * width + if len(text) >= width: + keep = (width - 3) // 2 + return f"{text[:keep]}...{text[len(text) - keep :]}" + return text + + +class ToolOutput(RedirectText): + """RedirectText that can collapse ninja status lines into one line. + + With ``progress`` each ``[n/m]`` line overwrites the previous one, the + way idf.py shows a build. + """ + + def __init__( + self, out: TextIO, filter_lines: list[str] | None, progress: bool + ) -> None: + super().__init__(out, filter_lines=filter_lines) + self._progress = progress + self._on_progress_line = False + + def _splits_lines(self) -> bool: + return self._progress or super()._splits_lines() + + def _emit_line(self, line: str) -> None: + if self._progress and _PROGRESS.match(line): + if not self._is_filtered(line): + text = _fit_terminal(line.strip("\r\n")) + self._write_color_replace(f"\r{text}\x1b[K") + self._on_progress_line = True + return + self._end_progress_line() + super()._emit_line(line) + + def _end_progress_line(self) -> None: + if self._on_progress_line: + self._on_progress_line = False + self._write_color_replace(os.linesep) + + def drain(self) -> None: + super().drain() + # Called from cleanup, so a broken stream must not hide the exit code. + with suppress(OSError, ValueError): + self._end_progress_line() + self._out.flush() + + +def run_build_tool( + cmd: list[str], + *, + cwd: Path, + env: dict[str, str], + filter_lines: list[str] | None = None, + progress: bool = False, + log_path: Path | None = None, +) -> int: + """Run ``cmd`` and relay stdout and stderr, merged, to our stdout. + + ``log_path`` also gets the full, unfiltered output without color codes, as + idf.py wrote its logs (its hint patterns expect plain text). Returns the + exit code. + """ + _LOGGER.debug("Running: %s", " ".join(shlex_quote(arg) for arg in cmd)) + _LOGGER.debug(" in directory: %s", cwd) + output = ToolOutput(sys.stdout, filter_lines, progress) + decoder = codecs.getincrementaldecoder("utf-8")(errors="replace") + if log_path is not None: + log_path.parent.mkdir(parents=True, exist_ok=True) + with ( + Path(log_path or os.devnull).open("w", encoding="utf-8", newline="") as log, + Utf8Console(_get_kernel32()), + subprocess.Popen( + cmd, + cwd=cwd, + env=env, + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + close_fds=False, + ) as proc, + ): + try: + # read1 returns as soon as anything is available, so output + # streams while the tool runs. + while chunk := proc.stdout.read1(_READ_SIZE): + text = decoder.decode(chunk) + log.write(ANSI_ESCAPE.sub("", text)) + output.write(text) + if tail := decoder.decode(b"", final=True): + log.write(ANSI_ESCAPE.sub("", tail)) + output.write(tail) + finally: + output.drain() + return proc.returncode diff --git a/esphome/build_helpers/tools_cache.py b/esphome/build_helpers/tools_cache.py index e3d38e7a98..85036c166f 100644 --- a/esphome/build_helpers/tools_cache.py +++ b/esphome/build_helpers/tools_cache.py @@ -16,8 +16,8 @@ def tools_cache_path(env_var: str, subdir: str) -> Path: from esphome.helpers import get_str_env if prefix := get_str_env(env_var, "").strip(): - # resolve(): symlinked prefixes otherwise trip idf.py's - # venv-mismatch warning on every build + # resolve(): a symlinked prefix would otherwise record a second + # spelling of the same paths in the build tree return Path(prefix).expanduser().resolve() # appauthor=False keeps the Windows path short (no vendor segment); # deep IDF trees run into MAX_PATH otherwise diff --git a/esphome/espidf/clang_tidy.py b/esphome/espidf/clang_tidy.py index 9a70f1a99a..3e1a510dee 100644 --- a/esphome/espidf/clang_tidy.py +++ b/esphome/espidf/clang_tidy.py @@ -11,7 +11,7 @@ running codegen on a config, it generates a minimal ESP-IDF CMake project: component, so their public include dirs land on the translation unit; * the repo ``sdkconfig.defaults`` enables sdkconfig-gated components (bt, ...). -then runs ``idf.py reconfigure`` (configure only, no compile) and reads the +then runs the CMake configure (no compile) and reads the resulting ``build/compile_commands.json``. The IDF version is the esp32 component's recommended version. @@ -367,7 +367,7 @@ def _write_tidy_project( def _generate_compile_commands( work_dir: Path, settings: _Settings, platformio_ini: Path ) -> Path: - """Generate the tidy project and run ``idf.py reconfigure`` (no build). + """Generate the tidy project and run the CMake configure (no build). Two-phase, like a real ESPHome build: a first configure with no builtin requires discovers which components actually register for the target (e.g. @@ -398,7 +398,7 @@ def _generate_compile_commands( # Phase 1: discover the components available for this target. _write_tidy_project(work_dir, [], extra_deps, settings) if toolchain.run_reconfigure() != 0: - raise RuntimeError("idf.py reconfigure (discovery) failed") + raise RuntimeError("ESP-IDF CMake configure (discovery) failed") requires = sorted( set(get_available_components() or []) - _NON_REQUIRABLE_COMPONENTS @@ -407,7 +407,7 @@ def _generate_compile_commands( # Phase 2: require every available builtin component. _write_tidy_project(work_dir, requires, extra_deps, settings) if toolchain.run_reconfigure() != 0: - raise RuntimeError("idf.py reconfigure failed") + raise RuntimeError("ESP-IDF CMake configure failed") return work_dir / "build" / "compile_commands.json" diff --git a/esphome/espidf/framework.py b/esphome/espidf/framework.py index 8885da05cc..1a4b17efdb 100644 --- a/esphome/espidf/framework.py +++ b/esphome/espidf/framework.py @@ -1211,38 +1211,37 @@ def _ccache_env() -> dict[str, str]: # ESPHOME_CCACHE_ENABLE. idf_knob = parse_enable_env("IDF_CCACHE_ENABLE") if idf_knob is False: - # The raw value (e.g. "disable") is still inherited by idf.py via - # os.environ, where a non-false-constant string reads as truthy; - # export the canonical off spelling instead + # Replace the inherited raw value (e.g. "disable") with the canonical + # off spelling, so every reader of the env sees the same answer return {"IDF_CCACHE_ENABLE": "0"} if idf_knob is True: # Forced on ignores the runnability verdict, but the outcome is # worth saying out loud. Probed directly (not via the resolver, # whose failure message says "compiling without ccache" -- exactly # what forced-on does NOT do): only the truly-missing case means - # idf.py compiles without ccache; a broken binary is still used, - # since idf.py does its own PATH lookup. + # the build compiles without ccache; a broken binary is still used, + # since IDF's CMake does its own PATH lookup. if (ccache := shutil.which("ccache")) is None: _LOGGER.warning( "IDF_CCACHE_ENABLE=1 but no ccache binary is on PATH; " - "idf.py will compile without ccache" + "the build will compile without ccache" ) else: # The probe warns with this message iff the binary fails tool_version_runs( ccache, "IDF_CCACHE_ENABLE=1 forces on the ccache at %s even though " - "it failed to run; idf.py will use it anyway", + "it failed to run; the build will use it anyway", ) elif resolve_ccache_path() is None: # ESP-IDF silently skips ccache without the binary; export the # canonical off spelling so an unparsable inherited value (or a - # probe-rejected ccache idf.py would still find) cannot enable it + # probe-rejected ccache CMake would still find) cannot enable it return {"IDF_CCACHE_ENABLE": "0"} env = ccache_defaults_env(get_idf_tools_path() / "ccache") env.update(ccache_pch_env()) - # Exactly one canonical spelling ever reaches idf.py, whatever the + # Exactly one canonical spelling ever reaches the build, whatever the # accepted input spelling was ("enable", "yes", ...) env["IDF_CCACHE_ENABLE"] = "1" return env diff --git a/esphome/espidf/runner.py b/esphome/espidf/runner.py deleted file mode 100644 index bf563a4f43..0000000000 --- a/esphome/espidf/runner.py +++ /dev/null @@ -1,379 +0,0 @@ -r"""Subprocess entry point for running ``idf.py`` with stdio wrapping. - -Invoked as ``python runner.py [script args...]``. - -Wraps ``sys.stdout`` and ``sys.stderr`` with a ``_FilteringTTYStream`` -shim so that: - -1. ``isatty()`` unconditionally returns True. CMake, Ninja, and idf.py's - own progress-bar code all check ``stream.isatty()`` to decide between - TTY-format output (``\\r`` cursor moves, ANSI colors, fancy progress - bars) and a plain fallback. With the wrapper in place they always - emit TTY format, even when our real stdout is a pipe to the parent - process (e.g. running under the Home Assistant dashboard add-on). - Downstream consumers — local terminals and the HA dashboard log - viewer — render the TTY control sequences correctly. - -2. ``FILTER_IDF_LINES`` is applied inside the shim's ``write()`` so - noisy idf.py output is dropped before it leaves this subprocess. - Filtering is skipped when ``-v`` / ``--verbose`` appears in argv so - verbose mode still shows everything. - -ESP-IDF runs under its own Python virtual environment which does not -have the ``esphome`` package installed, so the runner is intentionally -self-contained: no imports from ``esphome`` at all. The line-filtering -wrapper is inlined below rather than imported from -``esphome.util.RedirectText`` for that reason. -""" - -import sys - -# Regex patterns matched against each line of idf.py / CMake / Ninja -# output. Lines that match are dropped before reaching the parent -# process. Patterns are anchored at the start of the line (the shim -# uses ``re.match``). Disabled when the user passes ``-v`` / -# ``--verbose`` to ``esphome compile``. -FILTER_IDF_LINES: list[str] = [ - # idf.py's "how to flash" block at the end of a successful build. - # ESPHome handles flashing itself, so these instructions just clutter - # the output. - r"Project build complete\.", - r" idf\.py ", - r" python -m esptool ", - r"or$", - r"or from the ", - # CMake dumps the full list of IDF component paths on one giant line. - # It's purely informational and bloats the log. - r"-- Component paths:", - # CMake lists every linker script it adds (dozens of lines) and the - # complete flat list of IDF components on one giant line. Neither - # has diagnostic value for end users. - r"-- Adding linker script ", - r"-- Components:", - # IDF component manager notices: emitted on first build (no lock), - # once per stubbed dependency, plus the final "Processing N - # dependencies" enumeration. Patterns allow a leading run of dots - # because the component manager prints progress dots on the same - # line, so a NOTICE often arrives prefixed with ".NOTICE:" or - # "...........NOTICE:". - r"\.*NOTICE: ", - # ``idf.py size`` prefaces its table with a centered banner; the - # per-region table below already makes the structure obvious. - r"\s*Memory Type Usage Summary", - # Prefix match for esp-idf-size's trailing "Note:" paragraph (no - # upstream flag suppresses it). - r"Note: The reported total sizes may be smaller than those in the", - # Drop the blank line rich emits after the note so the build log - # doesn't end with an orphan gap before ESPHome's own status lines. - r"\s*$", - # ESP-IDF shells out to ``git rev-parse`` to embed a commit hash; - # esphome-libs strips ``.git`` from the tarball so those probes fail - # noisily without affecting the build. - r"-- git rev-parse returned ", - r"fatal: not a git repository", - r"Stopping at filesystem boundary", -] - -# Windows code page identifier for UTF-8, as used by ``chcp 65001``. -UTF8_CODEPAGE = 65001 - - -def _get_kernel32(): - """Return the Windows kernel32 module, or None on any other platform.""" - if sys.platform != "win32": - return None - import ctypes - - return ctypes.windll.kernel32 - - -class _Utf8Console: - """Keep an attached Windows console on UTF-8 for the length of the build. - - The build tree runs in UTF-8 mode, so esp_idf_size draws its table with - Unicode box characters. ``idf.py size`` reaches it through ``cmake -P``, - and CMake re-decodes the child's output with the console code page before - printing it, which turns the table into mojibake on any code page but - UTF-8. Every process in the build shares this console, so switching it - here covers CMake too. The old code pages go back on exit so the user's - terminal is left as it was. - - A console that is already on UTF-8 is left alone. The code page belongs to - the console, not to this process, so a build that overlaps another one - must not save UTF-8 as the page to go back to. - """ - - def __init__(self, kernel32) -> None: - self._kernel32 = kernel32 - self._codepages: tuple[int, int] | None = None - - def __enter__(self) -> None: - kernel32 = self._kernel32 - if kernel32 is None: - return - old_in = kernel32.GetConsoleCP() - old_out = kernel32.GetConsoleOutputCP() - # Both calls return 0 when no console is attached. - if not old_in or not old_out: - return - if old_in == UTF8_CODEPAGE and old_out == UTF8_CODEPAGE: - return - # Record the old pages first so a switch that fails part way through - # still gets put back on exit. - self._codepages = (old_in, old_out) - kernel32.SetConsoleCP(UTF8_CODEPAGE) - kernel32.SetConsoleOutputCP(UTF8_CODEPAGE) - - def __exit__(self, *exc_info: object) -> None: - if self._codepages is None: - return - old_in, old_out = self._codepages - self._codepages = None - self._kernel32.SetConsoleCP(old_in) - self._kernel32.SetConsoleOutputCP(old_out) - - -def main() -> int: - # ---- sys.path fix-up --------------------------------------------------- - # - # When Python runs this file as ``python runner.py``, it prepends the - # script's directory — ``/esphome/espidf/`` — to - # ``sys.path[0]``. That directory is part of the esphome package whose - # sibling ``types.py`` (in ``esphome/``) collides with stdlib ``types``. - # Any subsequent import that transitively touches ``types`` (``runpy``, - # ``pathlib``, ``functools``, ``typing``, ...) could resolve the wrong - # module. Drop the entry pre-emptively. ``sys`` is a built-in so - # importing it at module level earlier did not trigger the shadow. - if sys.path and sys.path[0]: - sys.path.pop(0) - # ---- end sys.path fix-up ----------------------------------------------- - - import contextlib - import os - from pathlib import Path - import re - import runpy - - # Patch ``os.get_terminal_size`` to return a fallback size instead - # of raising ``OSError`` when the underlying fd isn't a real - # terminal. - # - # idf.py's ``fit_text_in_terminal`` (in ``idf_py_actions/tools.py``) - # unconditionally calls ``os.get_terminal_size()`` to format ninja - # progress lines. When that raises ``[Errno 25] Inappropriate - # ioctl for device`` on our pipe-backed stdout, idf.py catches the - # exception as ``EnvironmentError`` and silently exits its stdout - # reader coroutine — dropping all ninja build output from that - # point on. Returning a valid value keeps the coroutine alive so - # progress and error lines continue to flow through to the parent - # process. - # - # Honour the ``COLUMNS`` / ``LINES`` env vars if the caller set - # them explicitly. Otherwise fall back to ``(0, 0)``, which - # ``fit_text_in_terminal`` treats as "unknown width, don't - # truncate" (see the ``if not terminal_width: return out`` guard). - # Downstream log viewers (local terminals, the HA dashboard) wrap - # or scroll long lines themselves, so we'd rather emit the full - # file path than have idf.py elide its middle. - _orig_get_terminal_size = os.get_terminal_size - - def _get_terminal_size_fallback(fd: int = 1) -> os.terminal_size: - try: - return _orig_get_terminal_size(fd) - except OSError: - try: - columns = int(os.environ.get("COLUMNS", "0")) - except ValueError: - columns = 0 - try: - lines = int(os.environ.get("LINES", "0")) - except ValueError: - lines = 0 - return os.terminal_size((columns, lines)) - - os.get_terminal_size = _get_terminal_size_fallback # type: ignore[assignment] - - # Strip ANSI escape sequences before comparing a line against the filter - # patterns, so colorized lines still match plain-text patterns. - ansi_escape = re.compile(r"\033[@-_][0-?]*[ -/]*[@-~]") - - class _FilteringTTYStream: - r"""Minimal stdout/stderr wrapper. - - * ``isatty()`` unconditionally returns True, tricking downstream - code into emitting TTY-format output. - * Input is split with ``str.splitlines(keepends=True)``, which - breaks on more than ``\\n`` and ``\\r``; form feed and a few - other control characters count too. Any piece whose - ANSI-stripped, right-stripped form matches one of - ``filter_lines`` is dropped. - * Only the final piece can still be waiting for more text, so - that one is held until a ``\\n`` or ``\\r`` arrives. A piece - that ended on one of the other breaks goes out as it is. - - Mirrors the matching semantics of ``esphome.util.RedirectText`` - so filter patterns behave identically in both the PlatformIO - and IDF runner paths. - """ - - def __init__(self, stream, filter_lines: list[str] | None) -> None: - self._stream = stream - if filter_lines: - combined = r"|".join(r"(?:" + p + r")" for p in filter_lines) - self._filter_pattern: re.Pattern[str] | None = re.compile(combined) - else: - self._filter_pattern = None - self._line_buffer = "" - - def __getattr__(self, name: str): - # Hide ``buffer`` so consumers that use either - # ``getattr(stream, 'buffer', None)`` or - # ``hasattr(stream, 'buffer')`` see this as a text-only stream - # and skip writing raw bytes (which would bypass the filter). - if name == "buffer": - raise AttributeError(name) - return getattr(self._stream, name) - - def isatty(self) -> bool: - return True - - def flush(self) -> None: - self._stream.flush() - - def _emit(self, line: str) -> None: - if self._filter_pattern is not None: - stripped = ansi_escape.sub("", line).rstrip() - if self._filter_pattern.match(stripped) is not None: - return - self._stream.write(line) - - def drain(self) -> None: - """Write out a held-back line that never got its terminator. - - idf.py and CMake do not always end their last line with a - newline, and a build that dies part way through can stop mid - line. Without this the user is left staring at a build that - ended with no explanation. - """ - if not self._line_buffer: - return - line, self._line_buffer = self._line_buffer, "" - try: - # Add the terminator the line never got, so whatever ESPHome - # prints next does not run onto the same line. - self._emit(line + "\n") - self._stream.flush() - except (OSError, ValueError) as err: - # We are called from cleanup, so raising would replace the - # build's real exit code. Saying so must not raise either: - # under the dashboard our stdout and stderr are the same - # pipe, so whatever broke the write has most likely broken - # the report, and ``sys.__stderr__`` is None on some - # interpreters. Carry the line along; it is usually the - # message saying why the build failed. - if (real_stderr := sys.__stderr__) is not None: - with contextlib.suppress(OSError, ValueError): - print( - f"Could not write out remaining output ({err}): {line}", - file=real_stderr, - ) - - def write(self, data) -> int: - # Text streams normally hand us ``str``; decode in case - # somebody writes bytes directly. - if not isinstance(data, str): - data = data.decode(errors="replace") - - if self._filter_pattern is None: - # Nothing to match against, so no need to wait for a full line. - self._emit(data) - else: - lines = (self._line_buffer + data).splitlines(keepends=True) - # Every piece but the last ends with something - # ``str.splitlines`` treats as a break, so only the last one - # can still be waiting for more text. Hold that one, write - # out the rest. - # - # Some of those breaks are not line endings to us, a form - # feed for one, so a piece can go out without ending in a - # newline. That beats what we did before, which was to stop - # at the first such piece and drop every complete line - # behind it. - if lines and not lines[-1].endswith(("\n", "\r")): - self._line_buffer = lines.pop() - else: - self._line_buffer = "" - for line in lines: - self._emit(line) - - # We tell idf.py it is talking to a terminal, so it sends progress - # bars and cursor moves. Our own stdout is usually a pipe, which is - # block buffered, so without this the build looks frozen until - # 8 KiB of output piles up. - self._stream.flush() - return len(data) - - if len(sys.argv) < 2: - print( - "usage: runner.py [args...]", - file=sys.stderr, - ) - return 2 - - script_path = sys.argv[1] - - # Mirror the platformio runner behaviour: verbose mode disables the - # line filter so all output reaches the user. - is_verbose = any(arg in ("-v", "--verbose") for arg in sys.argv[2:]) - filter_lines = None if is_verbose else FILTER_IDF_LINES or None - - class _FilteredStreams: - """Route ``sys.stdout`` and ``sys.stderr`` through the filtering shims. - - On exit each shim releases a last line that never got its - terminator. The shims made here are drained rather than whatever - ``sys.stdout`` holds by then, which the script is free to replace. - """ - - def __init__(self, filter_lines: list[str] | None) -> None: - self._stdout = _FilteringTTYStream(sys.stdout, filter_lines) - self._stderr = _FilteringTTYStream(sys.stderr, filter_lines) - - def __enter__(self) -> None: - sys.stdout = self._stdout # type: ignore[assignment] - sys.stderr = self._stderr # type: ignore[assignment] - - def __exit__(self, *exc_info: object) -> None: - # Drain stderr from a finally so a surprise from the first one - # cannot strand the second. - try: - self._stdout.drain() - finally: - self._stderr.drain() - - # Shift argv so the target script sees its own path as argv[0] and - # its own arguments starting at argv[1]. runpy.run_path does not - # modify sys.argv itself. - sys.argv = [script_path] + sys.argv[2:] - - # Emulate Python's default behaviour of prepending the script's - # directory to sys.path[0] when running ``python script.py``. - # runpy.run_path does not do this automatically, but idf.py relies - # on it to import its sibling modules (python_version_checker, - # idf_py_actions, ...). - script_dir = str(Path(script_path).resolve().parent) - if script_dir not in sys.path: - sys.path.insert(0, script_dir) - - # If idf.py calls sys.exit(), SystemExit propagates out of run_path - # and carries the exit code back to our caller. For normal returns, - # fall through and exit with 0. Either way the context managers drain - # the streams and put the console back on the way out, and they report - # instead of raising so cleanup cannot bury the real exit code. - with _FilteredStreams(filter_lines), _Utf8Console(_get_kernel32()): - runpy.run_path(script_path, run_name="__main__") - return 0 - - -if __name__ == "__main__": - sys.exit(main()) diff --git a/esphome/espidf/size_summary.py b/esphome/espidf/size_summary.py index d98363dd67..ffe97ba618 100644 --- a/esphome/espidf/size_summary.py +++ b/esphome/espidf/size_summary.py @@ -1,6 +1,6 @@ """PlatformIO-format RAM/Flash one-liners after a native ESP-IDF build. -``idf.py size`` (chained onto ``idf.py build`` in +The ninja ``size`` target (run after ``all`` in ``toolchain.run_compile``) prints the per-region table inline as part of the build. This module adds two summary lines underneath, byte-identical to PlatformIO's output: diff --git a/esphome/espidf/toolchain.py b/esphome/espidf/toolchain.py index 3bb1f5b83d..d1aeae9603 100644 --- a/esphome/espidf/toolchain.py +++ b/esphome/espidf/toolchain.py @@ -10,6 +10,7 @@ import re import shutil import subprocess +from esphome.build_helpers.tool_runner import run_build_tool from esphome.const import ( CONF_COMPILE_PROCESS_LIMIT, CONF_ESPHOME, @@ -189,72 +190,235 @@ def _get_idf_tool(name: str) -> str: return executable -def run_idf_py( - *args, - cwd: Path | None = None, - capture_output: bool = False, - jobs: int | None = None, -) -> int | str: - """Run idf.py with the given arguments.""" - idf_path = _get_idf_path() - if idf_path is None: - raise EsphomeError("ESP-IDF not found") +# Lines dropped from cmake and ninja output unless ``-v`` is given; matched +# with ``re.match`` against the line without ANSI codes or trailing space. +FILTER_IDF_LINES: list[str] = [ + # Full component path and linker script lists, one giant line each. + r"-- Component paths:", + r"-- Adding linker script ", + r"-- Components:", + # Component manager notices; progress dots can precede them. + r"\.*NOTICE: ", + # esp_idf_size banner and trailing note around the size table. + r"\s*Memory Type Usage Summary", + r"Note: The reported total sizes may be smaller than those in the", + r"\s*$", + # esphome-libs tarballs have no .git, so IDF's commit probes fail noisily. + r"-- git rev-parse returned ", + r"fatal: not a git repository", + r"Stopping at filesystem boundary", +] - env = _get_idf_env() - if jobs is not None: - env = {**env, "IDF_PY_BUILD_JOBS": str(jobs)} - python_executable = _get_idf_tool("python") - idf_py = idf_path / "tools" / "idf.py" - # Dispatch idf.py through esphome.espidf.runner, which wraps - # sys.stdout/sys.stderr so ``isatty()`` reports True. This keeps CMake, - # Ninja, and idf.py's own progress-bar code emitting TTY-format output - # (``\r`` cursor moves, ANSI colors, fancy progress bars) even when our - # real stdout is a pipe — e.g. when esphome is running under the Home - # Assistant dashboard add-on. The runner is a plain script (not a - # ``python -m`` module) because IDF's Python venv does not have the - # esphome package installed. - runner_py = Path(__file__).parent / "runner.py" - - cmd = [python_executable, str(runner_py), str(idf_py)] + list(args) - - if cwd is None: - cwd = CORE.build_path - - _LOGGER.debug("Running: %s", " ".join(cmd)) - _LOGGER.debug(" in directory: %s", cwd) - - if capture_output: - result = subprocess.run( - cmd, - cwd=cwd, - env=env, - capture_output=True, - text=True, - check=False, - ) - if result.returncode != 0: - _LOGGER.error("idf.py failed:\n%s", result.stderr) - return result.stdout - result = subprocess.run( - cmd, - cwd=cwd, - env=env, - check=False, - ) - return result.returncode +# click's boolean spellings, which idf.py applies to IDF_CCACHE_ENABLE. +_CLICK_TRUE = frozenset({"1", "true", "t", "yes", "y", "on"}) +_CMAKECACHE_LINE = re.compile(r"^([^#/:=]+):([^:=]+)=(.*)$") -def _get_sdkconfig_args() -> list[str]: - """Get cmake -D flags for the sdkconfig file, if it exists.""" +@dataclass(frozen=True, kw_only=True) +class _IdfPyContract: + """How the pinned idf.py drives cmake and ninja (tools/idf_py_actions).""" + + binary_dir_arg: bool # cmake gets -B + ccache_as_bool: bool # CCACHE_ENABLE=True/False instead of 1/0 + size_ng: bool # size target gets ESP_IDF_SIZE_NG=1 + + +_IDF_PY_5 = _IdfPyContract( + binary_dir_arg=False, + ccache_as_bool=False, + size_ng=True, +) +_IDF_PY_6 = _IdfPyContract( + binary_dir_arg=True, + ccache_as_bool=True, + size_ng=False, +) + + +def _idf_py() -> _IdfPyContract: + from esphome.components.esp32 import idf_version + import esphome.config_validation as cv + + return _IDF_PY_6 if idf_version() >= cv.Version(6, 0, 0) else _IDF_PY_5 + + +def _build_dir() -> Path: + """The CMake binary dir; idf.py resolves the project dir the same way.""" + return Path(os.path.realpath(CORE.build_path)) / "build" + + +def _cache_entries() -> dict[str, str]: + """The ``-D`` entries idf.py passes to cmake, in idf.py's order.""" + entries = {} sdkconfig_path = CORE.relative_build_path(f"sdkconfig.{CORE.name}") if sdkconfig_path.is_file(): - return ["-D", f"SDKCONFIG={sdkconfig_path}"] - return [] + entries["SDKCONFIG"] = str(sdkconfig_path) + ccache = _get_idf_env().get("IDF_CCACHE_ENABLE", "").strip().lower() in _CLICK_TRUE + entries["CCACHE_ENABLE"] = str(ccache if _idf_py().ccache_as_bool else int(ccache)) + return entries -def run_reconfigure() -> int: - """Run cmake reconfigure only (no build).""" - return run_idf_py(*_get_sdkconfig_args(), "reconfigure") +def _parse_cmakecache(path: Path) -> dict[str, str]: + """Map each ``NAME:TYPE=VALUE`` line of a CMakeCache.txt to NAME: VALUE.""" + result = {} + for line in path.read_text(encoding="utf-8").splitlines(): + if m := _CMAKECACHE_LINE.match(line): + result[m.group(1)] = m.group(3) + return result + + +def _cache_entries_changed() -> bool: + """True when a ``-D`` entry is missing from or differs in CMakeCache.txt. + + idf.py reconfigures on this before every build; ESPHome's own staleness + check does not cover it (for example ccache switched on or off). ESPHome + also compares ``PYTHON``, so a moved IDF prefix reconfigures; idf.py + stopped with an error instead. + """ + cache_path = _build_dir() / "CMakeCache.txt" + if not cache_path.is_file(): + return True + cache = _parse_cmakecache(cache_path) + return any(cache.get(k) != v for k, v in _configure_defines().items()) + + +def _configure_defines() -> dict[str, str]: + """Every ``-D`` idf.py passes to cmake, in its order.""" + return { + "PYTHON_DEPS_CHECKED": "1", + "PYTHON": _get_idf_tool("python"), + "ESP_PLATFORM": "1", + **_cache_entries(), + } + + +def _tool_env() -> dict[str, str]: + """The IDF env plus color, as idf.py 6.x gives every tool. + + Also used on 5.x (which forced CLICOLOR_FORCE for ninja only); color + changes only what is printed, and this way NO_COLOR is respected. + """ + env = dict(_get_idf_env()) + if not env.get("NO_COLOR"): + env.setdefault("CLICOLOR_FORCE", "1") + env.setdefault("FORCE_COLOR", "1") + return env + + +def run_reconfigure(verbose: bool = False) -> int: + """Run the CMake configure, with the arguments idf.py uses.""" + build_dir = _build_dir() + build_dir.mkdir(parents=True, exist_ok=True) + cmd = [_get_idf_tool("cmake"), "-G", "Ninja"] + if _idf_py().binary_dir_arg: + cmd += ["-B", str(build_dir)] + cmd += [f"-D{name}={value}" for name, value in _configure_defines().items()] + cmd.append(str(build_dir.parent)) + log_path = build_dir / "log" / "cmake_output.log" + rc = run_build_tool( + cmd, + cwd=build_dir, + env=_tool_env(), + filter_lines=None if verbose else FILTER_IDF_LINES, + log_path=log_path, + ) + if rc != 0: + # As idf.py does: a partial cache must not look configured. + (build_dir / "CMakeCache.txt").unlink(missing_ok=True) + _LOGGER.error("CMake configure failed with exit code %d", rc) + _print_hints(log_path) + return rc + + +def _size_env() -> dict[str, str]: + """Environment idf.py gives the ``size`` target.""" + env = {"ESP_IDF_SIZE_FORCE_TERMINAL": "1", "SIZE_OUTPUT_FORMAT": "default"} + if _idf_py().size_ng: + env["ESP_IDF_SIZE_NG"] = "1" + return env + + +def _build_jobs(config) -> int | None: + """Ninja's -j: compile_process_limit, else IDF_PY_BUILD_JOBS as idf.py read it.""" + if (limit := config[CONF_ESPHOME].get(CONF_COMPILE_PROCESS_LIMIT)) is not None: + return limit + if not (value := os.environ.get("IDF_PY_BUILD_JOBS")): + return None + try: + jobs = int(value) + except ValueError: + jobs = 0 + if jobs <= 0: + raise EsphomeError("IDF_PY_BUILD_JOBS must be a positive integer") + return jobs + + +def _run_ninja( + target: str, + *, + verbose: bool, + jobs: int | None, + progress: bool = False, + extra_env: dict[str, str] | None = None, +) -> int: + """Build one ninja target, with the flags and env idf.py uses.""" + cmd = [_get_idf_tool("ninja")] + if jobs is not None: + cmd += ["-j", str(jobs)] + if verbose: + cmd.append("-v") + cmd.append(target) + log_path = _build_dir() / "log" / f"ninja_{Path(target).name}_output.log" + rc = run_build_tool( + cmd, + cwd=_build_dir(), + env={**_tool_env(), **(extra_env or {})}, + filter_lines=None if verbose else FILTER_IDF_LINES, + progress=progress and not verbose, + log_path=log_path, + ) + if rc != 0: + _LOGGER.error("ninja %s failed with exit code %d", target, rc) + _print_hints(log_path) + return rc + + +# Runs IDF's own hint matcher (hints.yml plus its hint modules) on a failed +# tool's output, as idf.py did; it only lives in the IDF venv. +_HINTS_SCRIPT = """ +import sys +sys.path.insert(0, sys.argv[1]) +from idf_py_actions.tools import generate_hints +for hint in generate_hints(sys.argv[2]): + print(hint) +""" + + +def _print_hints(log_path: Path) -> None: + """Print ESP-IDF's advice for a failed build; never fails the build itself.""" + try: + result = subprocess.run( + [ + _get_idf_tool("python"), + "-c", + _HINTS_SCRIPT, + str(_get_idf_path() / "tools"), + str(log_path), + ], + env=_get_idf_env(), + capture_output=True, + text=True, + timeout=60, + check=False, + ) + except (OSError, subprocess.SubprocessError, EsphomeError) as err: + _LOGGER.debug("Could not get ESP-IDF hints: %s", err) + return + if result.returncode != 0: + _LOGGER.debug("Could not get ESP-IDF hints:\n%s", result.stderr) + return + if hints := result.stdout.strip(): + _LOGGER.warning("%s", hints) def _builtin_component_cache_path() -> Path | None: @@ -313,7 +477,9 @@ def save_cached_builtin_components(components: list[str]) -> None: _LOGGER.warning("Could not write component list cache %s: %s", path, err) -def _write_project_and_reconfigure(builtin_components: list[str] | None) -> int: +def _write_project_and_reconfigure( + builtin_components: list[str] | None, verbose: bool = False +) -> int: """Write the full CMakeLists.txt and run the configure for it.""" from esphome.build_gen.espidf import write_project @@ -321,12 +487,12 @@ def _write_project_and_reconfigure(builtin_components: list[str] | None) -> int: write_project(minimal=False, builtin_components=builtin_components) # Explicit reconfigure: ninja only re-runs cmake when CMakeLists.txt # is strictly newer than build.ninja, which fails on coarse-mtime - # filesystems (#18682). Also keeps idf.py from regenerating memory.ld + # filesystems (#18682). Also keeps ninja from regenerating memory.ld # in testing mode. - return run_reconfigure() + return run_reconfigure(verbose) -def _configure_project() -> int: +def _configure_project(verbose: bool = False) -> int: """Configure the project, discovering the built-in components if needed. A cached component list skips the discovery configure. If the configure @@ -337,20 +503,20 @@ def _configure_project() -> int: if (cached := load_cached_builtin_components()) is not None: _LOGGER.info("Using cached ESP-IDF component list") - if _write_project_and_reconfigure(cached) == 0: + if _write_project_and_reconfigure(cached, verbose) == 0: return 0 _LOGGER.warning("Cached component list failed; rediscovering") _builtin_component_cache_path().unlink(missing_ok=True) _LOGGER.info("Discovering available ESP-IDF components...") write_project(minimal=True) - if (rc := run_reconfigure()) != 0: + if (rc := run_reconfigure(verbose)) != 0: _LOGGER.error("Component discovery failed") return rc discovered = get_available_components() if not discovered: _LOGGER.error("Component discovery found no built-in ESP-IDF components") return 1 - if (rc := _write_project_and_reconfigure(discovered)) != 0: + if (rc := _write_project_and_reconfigure(discovered, verbose)) != 0: _LOGGER.error("Reconfigure with discovered components failed") return rc save_cached_builtin_components(discovered) @@ -489,11 +655,9 @@ def run_compile(config, verbose: bool) -> int: 2. Regenerate CMakeLists.txt with discovered components 3. Run full build """ - # Check if we need to do discovery phase - if not need_reconfigure(): - _LOGGER.info("Build configuration is up to date") - else: - if (rc := _configure_project()) != 0: + jobs = _build_jobs(config) + if need_reconfigure(): + if (rc := _configure_project(verbose)) != 0: return rc # cmake does not rewrite CMakeCache.txt when only properties change, # so restamp it or every build repeats discovery. Only after success, @@ -504,45 +668,31 @@ def run_compile(config, verbose: bool) -> int: path = CORE.relative_build_path(name) if path.is_file(): os.utime(path) + elif _cache_entries_changed(): + _LOGGER.info("CMake cache options changed, reconfiguring") + if (rc := run_reconfigure(verbose)) != 0: + return rc + else: + _LOGGER.info("Build configuration is up to date") # In testing mode, generate the linker script first, patch DRAM/IRAM sizes, # then build. memory.ld is regenerated by ninja during the build phase, # so we must patch after it's generated but before linking (same timing # as iram_fix.py.script's AddPreAction hook in the PlatformIO path). if CORE.testing_mode: - memory_ld = CORE.relative_build_path( - "build", "esp-idf", "esp_system", "ld", "memory.ld" - ) - build_dir = CORE.relative_build_path("build") - # Build just the memory.ld target - ninja needs the path relative to build dir - memory_ld_target = os.path.relpath(str(memory_ld), str(build_dir)) - env = _get_idf_env() - ninja_executable = _get_idf_tool("ninja") - result = subprocess.run( - [ninja_executable, "-C", str(build_dir), memory_ld_target], - env=env, - check=False, - ) - if result.returncode != 0: - _LOGGER.error("Failed to generate linker script") - return result.returncode + memory_ld = str(Path("esp-idf", "esp_system", "ld", "memory.ld")) + if (rc := _run_ninja(memory_ld, verbose=verbose, jobs=jobs)) != 0: + return rc _patch_memory_segments() from esphome.build_gen.espidf import write_pch_checksum write_pch_checksum() - # Build - args = [] - - if verbose: - args.append("-v") - - args.extend(_get_sdkconfig_args()) - args.append("build") - args.append("size") - - rc = run_idf_py(*args, jobs=config[CONF_ESPHOME].get(CONF_COMPILE_PROCESS_LIMIT)) + # idf.py's ``build size``, minus the second ``ninja all`` it runs first. + rc = _run_ninja("all", verbose=verbose, jobs=jobs, progress=True) + if rc == 0: + rc = _run_ninja("size", verbose=verbose, jobs=jobs, extra_env=_size_env()) if rc == 0: size_json = CORE.relative_build_path("build", "esp_idf_size.json") partitions = CORE.relative_build_path("partitions.csv") @@ -553,7 +703,7 @@ def run_compile(config, verbose: bool) -> int: def get_firmware_path() -> Path: """Get the path to the compiled firmware binary. - This is the file idf.py writes directly (named after the project), + This is the file the build writes directly (named after the project), not the copy used for OTA/factory downloads below. """ build_dir = CORE.relative_build_path("build") @@ -584,7 +734,7 @@ def get_ota_firmware_path() -> Path: def get_built_elf_path() -> Path: - """Path to the ELF idf.py writes directly, ``/.elf``. + """Path to the ELF the build writes directly, ``/.elf``. Exists as soon as the build finishes, unlike the ``firmware.elf`` copy that ``create_elf_copy`` makes later. @@ -596,7 +746,7 @@ def get_built_elf_path() -> Path: def get_elf_path() -> Path: """Get the path to the firmware ELF file. - idf.py writes ``/.elf`` directly; this returns the + The build writes ``/.elf`` directly; this returns the ``/firmware.elf`` copy created by ``create_elf_copy`` so the dashboard's "download ELF" link can find it under the PlatformIO-convention name. @@ -723,7 +873,7 @@ def create_ota_bin() -> bool: def create_elf_copy() -> bool: """Copy the ELF binary to firmware.elf for dashboard compatibility. - idf.py writes the ELF at ``/.elf``; the dashboard's + The build writes the ELF at ``/.elf``; the dashboard's "download ELF" link requests the literal filename ``firmware.elf`` (PlatformIO convention), so copy it to that name. """ diff --git a/esphome/util.py b/esphome/util.py index b8ffa048ca..8aa321d908 100644 --- a/esphome/util.py +++ b/esphome/util.py @@ -179,15 +179,21 @@ class RedirectText: s = s.replace("\033", "\\033") self._out.write(s) - def _emit_line(self, line: str) -> None: - line_without_ansi = ANSI_ESCAPE.sub("", line) - line_without_end = line_without_ansi.rstrip() - if ( + def _is_filtered(self, line: str) -> bool: + return ( self._filter_pattern is not None - and self._filter_pattern.match(line_without_end) is not None - ): - # Filter pattern matched, ignore the line + and self._filter_pattern.match(ANSI_ESCAPE.sub("", line).rstrip()) + is not None + ) + + def _splits_lines(self) -> bool: + """Whether output is handled line by line rather than passed through.""" + return self._filter_pattern is not None or bool(self._line_callbacks) + + def _emit_line(self, line: str) -> None: + if self._is_filtered(line): return + line_without_end = ANSI_ESCAPE.sub("", line).rstrip() self._write_color_replace(line) # Check for flash size error and provide helpful guidance @@ -233,7 +239,7 @@ class RedirectText: if not isinstance(s, str): s = s.decode() - if self._filter_pattern is not None or self._line_callbacks: + if self._splits_lines(): lines = (self._line_buffer + s).splitlines(True) # Every piece but the last ends with something # ``str.splitlines`` treats as a break, so only the last one can diff --git a/script/check_idf_py_equivalence.py b/script/check_idf_py_equivalence.py new file mode 100755 index 0000000000..fa349e223a --- /dev/null +++ b/script/check_idf_py_equivalence.py @@ -0,0 +1,198 @@ +#!/usr/bin/env python3 +"""Check that a native ESP-IDF build tree is what idf.py itself would produce. + +ESPHome runs cmake and ninja directly with the arguments idf.py uses. This +runs the real ``idf.py reconfigure`` and ``idf.py build`` on a finished tree +(in place: CMake rejects a moved cache) and fails if either one changes the +cache, the generated build files or the firmware, or recompiles anything. +It catches drift when the pinned ESP-IDF version changes what idf.py does. +The color and ``size`` environment only change what is printed, so those +parts of the contract are pinned by the unit tests instead. +""" + +from __future__ import annotations + +import argparse +import hashlib +import json +from pathlib import Path +import re +import subprocess +import sys + +REPO_ROOT = Path(__file__).resolve().parent.parent +sys.path.insert(0, str(REPO_ROOT)) + +WATCHED = ( + "build/CMakeCache.txt", + "build/build.ninja", + "build/compile_commands.json", + "build/project_description.json", + "build/config/sdkconfig.h", + "build/bootloader/bootloader.bin", +) +# Ninja logs whose outputs mean real work when their recorded mtime changes. +# The top level re-logs the bootloader step's byproducts on every build, so +# the bootloader is judged by its own sub-build log instead. +TOP_NINJA_LOG = "build/.ninja_log" +NINJA_LOGS = (TOP_NINJA_LOG, "build/bootloader/.ninja_log") +BOOTLOADER_BYPRODUCT = re.compile(r"(^|/build/)bootloader/") +WORK_SUFFIXES = (".obj", ".o", ".a", ".elf", ".map", ".bin", ".ld") +DEFAULT_GLOB = "tests/test_build_components/build/.esphome/build/*" + + +def _digest(path: Path) -> str | None: + return hashlib.sha256(path.read_bytes()).hexdigest() if path.is_file() else None + + +def watched(name: str) -> list[str]: + """Files that change if idf.py configures or builds differently.""" + return [*WATCHED, f"build/{name}.elf", f"build/{name}.bin"] + + +def _snapshot(build_path: Path, name: str) -> dict[str, str | None]: + return {f: _digest(build_path / f) for f in watched(name)} + + +def _ninja_mtimes(build_path: Path) -> dict[tuple[str, str], str]: + """(log, output) -> recorded mtime; compaction-safe, unlike a line count.""" + mtimes = {} + for name in NINJA_LOGS: + log = build_path / name + lines = log.read_text(encoding="utf-8").splitlines() if log.is_file() else [] + for fields in (line.split("\t") for line in lines if not line.startswith("#")): + if len(fields) >= 4 and ( + name != TOP_NINJA_LOG or not BOOTLOADER_BYPRODUCT.search(fields[3]) + ): + mtimes[name, fields[3]] = fields[2] + return mtimes + + +def _log_problems(build_path: Path, mtimes: dict[tuple[str, str], str]) -> list[str]: + """A missing or unparsable ninja log would otherwise compare as unchanged.""" + problems = [] + for log in NINJA_LOGS: + if not (build_path / log).is_file(): + problems.append(f"missing {log}") + elif not any(k[0] == log and k[1].endswith(WORK_SUFFIXES) for k in mtimes): + problems.append(f"no build entries parsed from {log}") + return problems + + +def _setup_core(build_path: Path, description: dict) -> tuple[str, str]: + """Point CORE at the tree so ESPHome resolves the same IDF env as the build.""" + from esphome.components.esp32.const import KEY_ESP32, KEY_IDF_VERSION, KEY_VARIANT + import esphome.config_validation as cv + from esphome.core import CORE + + name = description["project_name"] + version = Path(description["idf_path"]).name + CORE.config_path = build_path.parents[2] / f"{name}.yaml" + CORE.build_path = build_path + CORE.name = name + CORE.data[KEY_ESP32] = { + KEY_IDF_VERSION: cv.Version.parse(version), + KEY_VARIANT: description["target"].upper(), + } + return name, version + + +def check(build_path: Path) -> list[str]: + """Return the problems found in one build tree.""" + # pylint: disable=protected-access + from esphome.espidf import toolchain + + description = json.loads( + (build_path / "build" / "project_description.json").read_text(encoding="utf-8") + ) + name, version = _setup_core(build_path, description) + env = toolchain._get_idf_env(version) + python = toolchain._get_idf_tool("python") + idf_py = toolchain._get_idf_path(version) / "tools" / "idf.py" + sdkconfig = build_path / f"sdkconfig.{name}" + sdkconfig_args = ["-D", f"SDKCONFIG={sdkconfig}"] if sdkconfig.is_file() else [] + + # CMake writes a different build.ninja on a tree's first configure than on + # a reconfigure, so the baseline is ESPHome's own reconfigure and build. + if (rc := toolchain.run_reconfigure()) != 0: + return [f"ESPHome's CMake configure failed with exit code {rc}"] + if (rc := toolchain._run_ninja("all", verbose=False, jobs=None)) != 0: + return [f"ESPHome's ninja build failed with exit code {rc}"] + before = _snapshot(build_path, name) + mtimes_before = _ninja_mtimes(build_path) + # A moved or renamed output would otherwise compare as "unchanged". + problems = [f"missing {f}" for f, digest in before.items() if digest is None] + if problems := problems + _log_problems(build_path, mtimes_before): + return problems + for action in ("reconfigure", "build"): + result = subprocess.run( + [python, str(idf_py), *sdkconfig_args, action], + cwd=build_path, + env=env, + capture_output=True, + text=True, + check=False, + ) + if result.returncode != 0: + return [f"idf.py {action} failed:\n{result.stdout}{result.stderr}"] + after = _snapshot(build_path, name) + mtimes_after = _ninja_mtimes(build_path) + problems = [f"idf.py changed {f}" for f in before if before[f] != after[f]] + problems += _log_problems(build_path, mtimes_after) + for key in sorted(mtimes_before.keys() | mtimes_after.keys()): + log, out = key + if not out.endswith(WORK_SUFFIXES): + continue + if key not in mtimes_after: + problems.append(f"idf.py dropped {out} from {log}") + elif mtimes_before.get(key) != mtimes_after[key]: + problems.append(f"idf.py rebuilt {out}") + return problems + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__.split("\n", 1)[0]) + parser.add_argument( + "build_paths", + nargs="*", + type=Path, + help=f"ESPHome build dirs (default: the first native ESP-IDF tree in {DEFAULT_GLOB})", + ) + args = parser.parse_args() + paths = args.build_paths or sorted(REPO_ROOT.glob(DEFAULT_GLOB)) + # Not resolved: SDKCONFIG must be spelled as the build spelled it. + trees = [ + p + for p in paths + if (p / "build" / "project_description.json").is_file() + and (p / "build" / "CMakeCache.txt").is_file() + ] + if rejected := [p for p in args.build_paths if p not in trees]: + for path in rejected: + print(f"{path}: not a configured native ESP-IDF build tree") + return 1 + if not trees: + print("No native ESP-IDF build tree found") + return 1 + if not args.build_paths: + # The contract does not depend on the target, so one tree is enough. + trees = trees[:1] + + failed = False + for tree in trees: + problems = check(tree) + print(f"{tree}: {'OK' if not problems else 'DIFFERS'}") + for problem in problems: + print(f" {problem}") + failed |= bool(problems) + if failed: + print( + "The direct cmake/ninja build no longer matches idf.py. Compare " + "_IdfPyContract and its users in esphome/espidf/toolchain.py with the " + "pinned ESP-IDF tools/idf_py_actions." + ) + return 1 if failed else 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/script/determine-jobs.py b/script/determine-jobs.py index e3a0ff688f..bd127a7a43 100755 --- a/script/determine-jobs.py +++ b/script/determine-jobs.py @@ -592,10 +592,17 @@ def _esp32_platformio_path_or_file_trigger(files: list[str]) -> bool: ) +# Checks the native ESP-IDF build in CI but does not shape it, so it is kept +# out of ESP_IDF_INFRA_TRIGGER_FILES (hashed into the clang-tidy cache key). +_ESP_IDF_CHECK_TRIGGER_FILES = frozenset({"script/check_idf_py_equivalence.py"}) + + def _esp_idf_infra_changed(files: list[str]) -> bool: """Whether any changed file is ESP-IDF build/runner infrastructure.""" return _path_or_file_trigger( - files, ESP_IDF_INFRA_TRIGGER_FILES, ESP_IDF_INFRA_TRIGGER_PATH_PREFIXES + files, + ESP_IDF_INFRA_TRIGGER_FILES | _ESP_IDF_CHECK_TRIGGER_FILES, + ESP_IDF_INFRA_TRIGGER_PATH_PREFIXES, ) @@ -1533,19 +1540,34 @@ def main() -> None: # Convert batches to CI matrix entries: the component list plus which # native toolchain installs the batch's test platforms need, so the # workflow only restores the matching multi-GB toolchain caches. + # The idf.py check does not depend on the components, so it runs once + # per workflow, in the first batch that compiles an esp32 test (a + # validate-only component is never compiled). + idf_py_check_assigned = False + skip_compile = set(validate_only_components) for batch in batches: platforms: set[str] = set() + compiled_platforms: set[str] = set() for component in batch: # Variants included: the compile stage builds them, so a # component tested only by test-..yaml # still needs that platform's toolchain - platforms.update( - get_component_test_platforms(component, base_only=False) + component_platforms = get_component_test_platforms( + component, base_only=False ) + platforms.update(component_platforms) + if component not in skip_compile: + compiled_platforms.update(component_platforms) + needs_idf = any(p.startswith("esp32") for p in platforms) + check_idf_py = not idf_py_check_assigned and any( + p.startswith("esp32") for p in compiled_platforms + ) + idf_py_check_assigned |= check_idf_py component_test_batches.append( { "components": " ".join(batch), - "needs_idf": any(p.startswith("esp32") for p in platforms), + "needs_idf": needs_idf, + "check_idf_py": check_idf_py, "needs_nrf": any(p.startswith("nrf52") for p in platforms), "needs_arduino8266": any( p.startswith("esp8266") for p in platforms diff --git a/tests/script/test_check_idf_py_equivalence.py b/tests/script/test_check_idf_py_equivalence.py new file mode 100644 index 0000000000..f8135772d0 --- /dev/null +++ b/tests/script/test_check_idf_py_equivalence.py @@ -0,0 +1,248 @@ +"""Tests for script/check_idf_py_equivalence.py.""" + +from collections.abc import Callable, Iterator +import json +from pathlib import Path +import subprocess +import sys +from unittest.mock import patch + +import pytest + +sys.path.insert(0, str(Path(__file__).parent.parent.parent / "script")) + +import check_idf_py_equivalence as guard # noqa: E402 + +from esphome.core import CORE # noqa: E402 +from esphome.espidf import toolchain # noqa: E402 + + +@pytest.fixture(autouse=True) +def _reset_core() -> Iterator[None]: + """check() points the global CORE at the tree it inspects.""" + yield + CORE.reset() + + +def _make_tree(tmp_path: Path) -> Path: + tree = tmp_path / "config" / ".esphome" / "build" / "dev" + build = tree / "build" + for name in (*guard.watched("dev"), *guard.NINJA_LOGS): + (tree / name).parent.mkdir(parents=True, exist_ok=True) + (tree / name).write_bytes(b"x") + (build / "project_description.json").write_text( + json.dumps( + { + "project_name": "dev", + "idf_path": "/idf/frameworks/5.5.5", + "target": "esp32", + } + ) + ) + (build / ".ninja_log").write_text( + "# ninja log v7\n1\t2\t10\tesp-idf/a.obj\t0\n" + "1\t2\t10\tbootloader/bootloader.bin\t0\n" + ) + (build / "bootloader" / ".ninja_log").write_text( + "# ninja log v7\n1\t2\t10\tbootloader.elf\t0\n" + ) + (tree / "sdkconfig.dev").write_text("") + return tree + + +def _run_check( + tree: Path, + side_effect: Callable[[list[str]], None] = lambda cmd: None, + rc: int = 0, + esphome_rcs: tuple[int, int] = (0, 0), +) -> tuple[list[str], list[list[str]]]: + """Run check() with idf.py replaced by ``side_effect``; return problems, calls.""" + calls: list[list[str]] = [] + + def run(cmd: list[str], **kwargs: object) -> subprocess.CompletedProcess: + calls.append(cmd) + side_effect(cmd) + return subprocess.CompletedProcess(cmd, rc, "out\n", "err\n") + + with ( + patch.object(toolchain, "_get_idf_env", return_value={}), + patch.object(toolchain, "_get_idf_tool", return_value="/py"), + patch.object(toolchain, "_get_idf_path", return_value=Path("/idf")), + patch.object(toolchain, "run_reconfigure", return_value=esphome_rcs[0]), + patch.object(toolchain, "_run_ninja", return_value=esphome_rcs[1]), + patch.object(guard.subprocess, "run", side_effect=run), + ): + return guard.check(tree), calls + + +def test_check_passes_when_idf_py_changes_nothing(tmp_path: Path) -> None: + tree = _make_tree(tmp_path) + problems, calls = _run_check(tree) + assert problems == [] + sdkconfig = f"SDKCONFIG={tree / 'sdkconfig.dev'}" + assert calls == [ + ["/py", str(Path("/idf/tools/idf.py")), "-D", sdkconfig, "reconfigure"], + ["/py", str(Path("/idf/tools/idf.py")), "-D", sdkconfig, "build"], + ] + + +def test_check_reports_changed_files_and_rebuilt_outputs(tmp_path: Path) -> None: + tree = _make_tree(tmp_path) + build = tree / "build" + + def drift(cmd: list[str]) -> None: + if cmd[-1] == "reconfigure": + (build / "CMakeCache.txt").write_text("changed") + return + # Compacted log. The re-logged bootloader byproduct and a stamp are + # not work; a new object mtime is. + (build / ".ninja_log").write_text( + "# ninja log v7\n3\t4\t20\tesp-idf/a.obj\t0\n" + "5\t6\t30\tbootloader/bootloader.bin\t0\n" + "5\t6\t30\tbootloader-stamp\t0\n" + ) + # The bootloader sub-build is judged by its own log. + (build / "bootloader" / ".ninja_log").write_text( + "# ninja log v7\n1\t2\t40\tbootloader.elf\t0\n" + ) + + problems, _ = _run_check(tree, drift) + assert problems == [ + "idf.py changed build/CMakeCache.txt", + "idf.py rebuilt esp-idf/a.obj", + "idf.py rebuilt bootloader.elf", + ] + + +@pytest.mark.parametrize( + ("after_build", "problem"), + [ + (lambda log: log.unlink(), "missing build/.ninja_log"), + ( + lambda log: log.write_text("# ninja log v7\n"), + "no build entries parsed from build/.ninja_log", + ), + ( + lambda log: log.write_text("# ninja log v7\n1\t2\t10\tesp-idf/b.obj\t0\n"), + "idf.py dropped esp-idf/a.obj from build/.ninja_log", + ), + ], + ids=["log-removed", "log-emptied", "entry-dropped"], +) +def test_check_reports_a_log_idf_py_left_unusable( + tmp_path: Path, after_build: Callable[[Path], None], problem: str +) -> None: + """The comparison side gets the same log checks as the baseline.""" + tree = _make_tree(tmp_path) + log = tree / guard.TOP_NINJA_LOG + + def run(cmd: list[str]) -> None: + if cmd[-1] == "build": + after_build(log) + + problems, _ = _run_check(tree, run) + assert problem in problems + + +def test_check_stops_when_idf_py_fails(tmp_path: Path) -> None: + problems, calls = _run_check(_make_tree(tmp_path), rc=2) + assert problems == ["idf.py reconfigure failed:\nout\nerr\n"] + assert len(calls) == 1 + + +@pytest.mark.parametrize( + "remove", ["build/build.ninja", "build/dev.bin", *guard.NINJA_LOGS] +) +def test_check_fails_when_an_input_is_missing(tmp_path: Path, remove: str) -> None: + """A moved or renamed output must not compare as unchanged.""" + tree = _make_tree(tmp_path) + (tree / remove).unlink() + problems, calls = _run_check(tree) + assert problems == [f"missing {remove}"] + assert calls == [] + + +@pytest.mark.parametrize( + ("esphome_rcs", "problem"), + [ + ((3, 0), "ESPHome's CMake configure failed with exit code 3"), + ((0, 4), "ESPHome's ninja build failed with exit code 4"), + ], + ids=["configure", "build"], +) +def test_check_stops_when_the_esphome_baseline_fails( + tmp_path: Path, esphome_rcs: tuple[int, int], problem: str +) -> None: + """The baseline is ESPHome's own reconfigure and build.""" + problems, calls = _run_check(_make_tree(tmp_path), esphome_rcs=esphome_rcs) + assert problems == [problem] + assert calls == [] + + +@pytest.mark.parametrize("log", guard.NINJA_LOGS) +def test_check_fails_when_a_ninja_log_has_no_entries(tmp_path: Path, log: str) -> None: + """A log format change must not leave the rebuild check with nothing to compare.""" + tree = _make_tree(tmp_path) + (tree / log).write_text("# ninja log v99\n1 2 3\n") + problems, calls = _run_check(tree) + assert problems == [f"no build entries parsed from {log}"] + assert calls == [] + + +def test_main_rejects_a_path_that_is_not_a_tree( + tmp_path: Path, capsys: pytest.CaptureFixture[str] +) -> None: + tree = _make_tree(tmp_path / "a") + stale = tmp_path / "stale" + with ( + patch.object(sys, "argv", ["check", str(tree), str(stale)]), + patch.object(guard, "check", return_value=[]) as mock_check, + ): + assert guard.main() == 1 + assert f"{stale}: not a configured native ESP-IDF build tree" in ( + capsys.readouterr().out + ) + mock_check.assert_not_called() + + +def test_main_without_build_trees( + tmp_path: Path, capsys: pytest.CaptureFixture[str] +) -> None: + with ( + patch.object(sys, "argv", ["check"]), + patch.object(guard, "REPO_ROOT", tmp_path), + ): + assert guard.main() == 1 + assert "No native ESP-IDF build tree found" in capsys.readouterr().out + + +@pytest.mark.parametrize(("problems", "rc"), [([], 0), (["idf.py changed x"], 1)]) +def test_main_reports_each_tree( + tmp_path: Path, + capsys: pytest.CaptureFixture[str], + problems: list[str], + rc: int, +) -> None: + tree = _make_tree(tmp_path) + with ( + patch.object(sys, "argv", ["check", str(tree)]), + patch.object(guard, "check", return_value=problems), + ): + assert guard.main() == rc + out = capsys.readouterr().out + assert f"{tree}: {'DIFFERS' if problems else 'OK'}" in out + assert ("no longer matches idf.py" in out) is bool(problems) + + +def test_main_checks_only_the_first_found_tree(tmp_path: Path) -> None: + """The contract does not depend on the target; one tree per batch is enough.""" + first = _make_tree(tmp_path / "a") + _make_tree(tmp_path / "b") + with ( + patch.object(sys, "argv", ["check"]), + patch.object(guard, "REPO_ROOT", tmp_path), + patch.object(guard, "DEFAULT_GLOB", "*/config/.esphome/build/*"), + patch.object(guard, "check", return_value=[]) as mock_check, + ): + assert guard.main() == 0 + mock_check.assert_called_once_with(first) diff --git a/tests/script/test_determine_jobs.py b/tests/script/test_determine_jobs.py index 49f101d679..cea2a2bfe6 100644 --- a/tests/script/test_determine_jobs.py +++ b/tests/script/test_determine_jobs.py @@ -249,11 +249,61 @@ def test_main_all_tests_should_run( # Should contain at least one component (no empty batches) assert len(batch["components"]) > 0 assert isinstance(batch["needs_idf"], bool) + assert isinstance(batch["check_idf_py"], bool) assert isinstance(batch["needs_nrf"], bool) assert isinstance(batch["needs_arduino8266"], bool) -def test_main_batch_flags_count_variant_tests( +def _batch(components: str, idf: bool, check: bool, a8266: bool) -> dict: + return { + "components": components, + "needs_idf": idf, + "check_idf_py": check, + "needs_nrf": False, + "needs_arduino8266": a8266, + } + + +@pytest.mark.parametrize( + ("batches", "component_platforms", "validate_only", "expected"), + [ + # The compile stage builds test-..yaml too, so a + # component tested on esp8266 only by a variant still needs the toolchain. + ( + [["safe_mode"]], + {"safe_mode": {"esp8266-ard"}}, + set(), + [_batch("safe_mode", idf=False, check=False, a8266=True)], + ), + # Only the first esp32 batch runs the idf.py equivalence check. + ( + [["a"], ["b"], ["c"]], + {"a": {"esp8266-ard"}, "b": {"esp32-idf"}, "c": {"esp32-c3-idf"}}, + set(), + [ + _batch("a", idf=False, check=False, a8266=True), + _batch("b", idf=True, check=True, a8266=False), + _batch("c", idf=True, check=False, a8266=False), + ], + ), + # A validate-only batch never compiles, so the next esp32 batch runs it. + ( + [["b"], ["c"]], + {"b": {"esp32-idf"}, "c": {"esp32-c3-idf"}}, + {"b"}, + [ + _batch("b", idf=True, check=False, a8266=False), + _batch("c", idf=True, check=True, a8266=False), + ], + ), + ], + ids=["variant", "idf-check-once", "idf-check-skips-validate-only"], +) +def test_main_batch_flags( + batches: list[list[str]], + component_platforms: dict[str, set[str]], + validate_only: set[str], + expected: list[dict], mock_determine_integration_tests: Mock, mock_should_run_clang_tidy: Mock, mock_should_run_clang_format: Mock, @@ -267,8 +317,7 @@ def test_main_batch_flags_count_variant_tests( capsys: pytest.CaptureFixture[str], monkeypatch: pytest.MonkeyPatch, ) -> None: - """The compile stage builds test-..yaml too, so a - component tested on esp8266 only by a variant still needs the toolchain.""" + """Each batch carries the toolchain flags its test platforms need.""" monkeypatch.delenv("GITHUB_ACTIONS", raising=False) mock_determine_integration_tests.return_value = (False, []) mock_should_run_clang_tidy.return_value = False @@ -282,13 +331,20 @@ def test_main_batch_flags_count_variant_tests( mock_changed_files.return_value = ["esphome/components/safe_mode/__init__.py"] def platforms(component: str, *, base_only: bool = True) -> set[str]: - return set() if base_only else {"esp8266-ard"} + return set() if base_only else component_platforms[component] with ( patch("sys.argv", ["determine-jobs.py"]), patch.object(determine_jobs, "_is_clang_tidy_full_scan", return_value=False), patch.object( - determine_jobs, "get_changed_components", return_value=["safe_mode"] + determine_jobs, + "get_changed_components", + return_value=[c for batch in batches for c in batch], + ), + patch.object( + determine_jobs, + "_component_change_is_validate_only", + side_effect=lambda component, changed: component in validate_only, ), patch.object( determine_jobs, @@ -298,7 +354,7 @@ def test_main_batch_flags_count_variant_tests( patch.object( determine_jobs, "get_components_with_dependencies", - return_value=["safe_mode"], + return_value=[c for batch in batches for c in batch], ), patch.object(determine_jobs, "_component_has_tests", return_value=True), patch.object( @@ -307,9 +363,7 @@ def test_main_batch_flags_count_variant_tests( return_value={"should_run": "false"}, ), patch.object( - determine_jobs, - "create_intelligent_batches", - return_value=([["safe_mode"]], {}), + determine_jobs, "create_intelligent_batches", return_value=(batches, {}) ), patch.object( determine_jobs, "get_component_test_platforms", side_effect=platforms @@ -318,14 +372,7 @@ def test_main_batch_flags_count_variant_tests( determine_jobs.main() output = json.loads(capsys.readouterr().out) - assert output["component_test_batches"] == [ - { - "components": "safe_mode", - "needs_idf": False, - "needs_nrf": False, - "needs_arduino8266": True, - } - ] + assert output["component_test_batches"] == expected def test_main_no_tests_should_run( @@ -1231,8 +1278,10 @@ def test_should_run_esp32_platformio_with_branch() -> None: @pytest.mark.parametrize( ("changed_files", "expected"), [ - # ESP-IDF runner / framework / build generator -> trigger - (["esphome/espidf/runner.py"], True), + # ESP-IDF toolchain / framework / build generator -> trigger + (["esphome/espidf/toolchain.py"], True), + (["esphome/build_helpers/tool_runner.py"], True), + (["script/check_idf_py_equivalence.py"], True), (["esphome/espidf/framework.py"], True), (["esphome/build_gen/espidf.py"], True), # Shared native-build modules the IDF build imports -> trigger @@ -1250,7 +1299,7 @@ def test_should_run_esp32_platformio_with_branch() -> None: ], ) def test_esp_idf_infra_changed(changed_files: list[str], expected: bool) -> None: - """ESP-IDF build/runner infra paths are detected; other paths are not.""" + """ESP-IDF build infra paths are detected; other paths are not.""" assert determine_jobs._esp_idf_infra_changed(changed_files) is expected diff --git a/tests/unit_tests/build_helpers/test_tool_runner.py b/tests/unit_tests/build_helpers/test_tool_runner.py new file mode 100644 index 0000000000..e2c7e44341 --- /dev/null +++ b/tests/unit_tests/build_helpers/test_tool_runner.py @@ -0,0 +1,338 @@ +"""Tests for esphome.build_helpers.tool_runner.""" + +# pylint: disable=protected-access + +import io +import os +from pathlib import Path +import sys +import threading +from types import SimpleNamespace +from unittest.mock import patch + +import pytest + +from esphome.build_helpers import tool_runner +from esphome.build_helpers.tool_runner import ( + UTF8_CODEPAGE, + ToolOutput, + Utf8Console, + run_build_tool, +) +from esphome.core import CORE + +FILTER = [r"-- Component paths:", r"\s*$"] + + +def _child(code: str) -> list[str]: + return [sys.executable, "-c", code] + + +def _run( + capsys: pytest.CaptureFixture[str], + tmp_path: Path, + code: str, + **kwargs: object, +) -> tuple[int, str]: + rc = run_build_tool(_child(code), cwd=tmp_path, env=dict(os.environ), **kwargs) + # The child's print() ends lines with \r\n on Windows. + return rc, capsys.readouterr().out.replace("\r\n", "\n") + + +def test_run_build_tool_filters_and_returns_exit_code( + capsys: pytest.CaptureFixture[str], tmp_path: Path +) -> None: + rc, out = _run( + capsys, + tmp_path, + "import sys\n" + "print('-- Component paths: /a /b')\n" + "print('')\n" + "print('Compiling main.cpp')\n" + "sys.exit(3)", + filter_lines=FILTER, + ) + assert rc == 3 + assert out == "Compiling main.cpp\n" + + +def test_run_build_tool_passes_everything_without_filter( + capsys: pytest.CaptureFixture[str], tmp_path: Path +) -> None: + rc, out = _run(capsys, tmp_path, "print('-- Component paths: /a')\nprint('')") + assert rc == 0 + assert out == "-- Component paths: /a\n\n" + + +def test_run_build_tool_merges_stderr_and_runs_in_cwd( + capsys: pytest.CaptureFixture[str], tmp_path: Path +) -> None: + _, out = _run( + capsys, + tmp_path, + "import os, sys\nprint(os.getcwd(), flush=True)\nprint('oops', file=sys.stderr)", + filter_lines=FILTER, + ) + assert out.splitlines() == [os.path.realpath(tmp_path), "oops"] + + +def test_run_build_tool_drains_a_partial_line( + capsys: pytest.CaptureFixture[str], tmp_path: Path +) -> None: + """A tool that dies mid line still shows that line, terminated.""" + rc, out = _run( + capsys, + tmp_path, + "import sys\nsys.stdout.write('ld returned 1 exit status')\nsys.exit(1)", + filter_lines=FILTER, + ) + assert rc == 1 + assert out == "ld returned 1 exit status\n" + + +def test_run_build_tool_replaces_invalid_utf8( + capsys: pytest.CaptureFixture[str], tmp_path: Path +) -> None: + _, out = _run( + capsys, + tmp_path, + # A multi-byte character split across writes must survive too. + "import sys\nb = sys.stdout.buffer\n" + "b.write(b'\\xc3'); b.flush(); b.write(b'\\xa9 ok\\n\\xff bad\\n')", + filter_lines=FILTER, + ) + assert out == "é ok\n� bad\n" + + +def test_run_build_tool_streams_before_the_tool_exits(tmp_path: Path) -> None: + """Output must reach the user while the tool runs, not when it ends.""" + seen = threading.Event() + lines: list[str] = [] + + class _Out(io.StringIO): + def write(self, s: str) -> int: + lines.append(s) + if "first" in s: + seen.set() + return len(s) + + release = tmp_path / "release" + code = ( + "import os, time\nprint('first', flush=True)\n" + f"while not os.path.exists({str(release)!r}): time.sleep(0.02)\n" + ) + with patch.object(tool_runner.sys, "stdout", _Out()): + thread = threading.Thread( + target=run_build_tool, + args=(_child(code),), + kwargs={"cwd": tmp_path, "env": dict(os.environ), "filter_lines": FILTER}, + ) + thread.start() + try: + assert seen.wait(30) + finally: + release.touch() + thread.join(30) + assert "first" in "".join(lines) + + +def _tool_output(progress: bool = True) -> tuple[ToolOutput, io.StringIO]: + out = io.StringIO() + return ToolOutput(out, FILTER, progress), out + + +def test_tool_output_collapses_progress_lines() -> None: + """Ninja status lines overwrite each other, as idf.py shows them.""" + output, out = _tool_output() + output.write("[1/2] Building a.o\n[2/2] Linking app\nwarning: x\n") + assert out.getvalue() == ( + "\r[1/2] Building a.o\x1b[K\r[2/2] Linking app\x1b[K" + + os.linesep + + "warning: x\n" + ) + + +def test_tool_output_drain_ends_a_progress_line() -> None: + output, out = _tool_output() + output.write("[1/1] Linking app\n") + output.drain() + assert out.getvalue() == "\r[1/1] Linking app\x1b[K" + os.linesep + # Nothing is pending any more. + output.drain() + assert out.getvalue().count(os.linesep) == 1 + + +def test_tool_output_drain_survives_a_broken_stream() -> None: + output, out = _tool_output() + output.write("[1/1] Linking app\n") + out.close() + output.drain() + + +def test_tool_output_filters_a_matching_progress_line() -> None: + out = io.StringIO() + output = ToolOutput(out, [r"\[1/2\]"], True) + output.write("[1/2] hidden\n[2/2] shown\n") + assert out.getvalue() == "\r[2/2] shown\x1b[K" + + +def test_tool_output_without_progress_keeps_status_lines() -> None: + output, out = _tool_output(progress=False) + output.write("[1/2] Building a.o\n") + assert out.getvalue() == "[1/2] Building a.o\n" + + +def test_tool_output_escapes_colors_for_the_dashboard() -> None: + CORE.dashboard = True + output, out = _tool_output() + output.write("[1/1] \x1b[1mLinking\x1b[0m\n") + assert "\x1b" not in out.getvalue() + assert "\\033[K" in out.getvalue() + + +@pytest.mark.parametrize( + ("width", "text", "expected"), + [ + (0, "x" * 50, "x" * 50), + (3, "abcdef", "..."), + (20, "short", "short"), + (11, "abcdefghijklmnop", "abcd...mnop"), + ], + ids=["unknown", "tiny", "fits", "elided"], +) +def test_fit_terminal(width: int, text: str, expected: str) -> None: + with patch.object( + tool_runner.shutil, + "get_terminal_size", + return_value=os.terminal_size((width, 24)), + ): + assert tool_runner._fit_terminal(text) == expected + + +class _FakeKernel32: + """Stand-in for the Windows kernel32 console code page calls.""" + + def __init__(self, input_cp: int, output_cp: int) -> None: + self.input_cp = input_cp + self.output_cp = output_cp + self.calls: list[tuple[str, int]] = [] + + def GetConsoleCP(self) -> int: # noqa: N802 + return self.input_cp + + def GetConsoleOutputCP(self) -> int: # noqa: N802 + return self.output_cp + + def SetConsoleCP(self, codepage: int) -> int: # noqa: N802 + self.calls.append(("SetConsoleCP", codepage)) + self.input_cp = codepage + return 1 + + def SetConsoleOutputCP(self, codepage: int) -> int: # noqa: N802 + self.calls.append(("SetConsoleOutputCP", codepage)) + self.output_cp = codepage + return 1 + + +def test_run_build_tool_switches_the_console_to_utf8( + capsys: pytest.CaptureFixture[str], tmp_path: Path +) -> None: + """An attached console runs the tool on UTF-8 and is then put back.""" + kernel32 = _FakeKernel32(850, 850) + with patch.object(tool_runner, "_get_kernel32", return_value=kernel32): + _run(capsys, tmp_path, "print('x')") + assert kernel32.calls == [ + ("SetConsoleCP", UTF8_CODEPAGE), + ("SetConsoleOutputCP", UTF8_CODEPAGE), + ("SetConsoleCP", 850), + ("SetConsoleOutputCP", 850), + ] + + +def test_utf8_console_restores_after_an_error() -> None: + kernel32 = _FakeKernel32(437, 437) + with pytest.raises(RuntimeError), Utf8Console(kernel32): + raise RuntimeError + assert (kernel32.input_cp, kernel32.output_cp) == (437, 437) + + +def test_utf8_console_restores_when_the_switch_fails_part_way() -> None: + kernel32 = _FakeKernel32(850, 850) + + def _refuse(codepage: int) -> int: + kernel32.calls.append(("SetConsoleOutputCP", codepage)) + return 0 + + kernel32.SetConsoleOutputCP = _refuse # type: ignore[method-assign] + with Utf8Console(kernel32): + pass + assert kernel32.input_cp == 850 + assert kernel32.calls[-2:] == [("SetConsoleCP", 850), ("SetConsoleOutputCP", 850)] + + +@pytest.mark.parametrize( + "codepages", [(0, 0), (UTF8_CODEPAGE, UTF8_CODEPAGE)], ids=["none", "utf8"] +) +def test_utf8_console_leaves_the_console_alone(codepages: tuple[int, int]) -> None: + """No console, or one already on UTF-8 (an overlapping build), is untouched.""" + kernel32 = _FakeKernel32(*codepages) + with Utf8Console(kernel32): + pass + assert kernel32.calls == [] + + +def test_utf8_console_without_kernel32() -> None: + with Utf8Console(None): + pass + + +@pytest.mark.skipif(sys.platform == "win32", reason="kernel32 exists on Windows") +def test_get_kernel32_is_none_off_windows() -> None: + assert tool_runner._get_kernel32() is None + + +def test_tool_output_collapses_progress_without_a_filter() -> None: + """Progress mode splits lines on its own; it does not need a filter.""" + out = io.StringIO() + output = ToolOutput(out, None, True) + output.write("[1/1] Linking app\ndone\n") + assert out.getvalue() == "\r[1/1] Linking app\x1b[K" + os.linesep + "done\n" + + +def test_run_build_tool_flushes_a_truncated_character( + capsys: pytest.CaptureFixture[str], tmp_path: Path +) -> None: + """Output that ends inside a multi-byte character still shows up.""" + _, out = _run( + capsys, + tmp_path, + "import sys\nsys.stdout.buffer.write(b'end \\xc3')", + filter_lines=FILTER, + ) + assert out == "end �\n" + + +def test_get_kernel32_loads_it_on_windows(monkeypatch: pytest.MonkeyPatch) -> None: + kernel32 = object() + fake_ctypes = SimpleNamespace(windll=SimpleNamespace(kernel32=kernel32)) + monkeypatch.setattr(tool_runner.sys, "platform", "win32") + monkeypatch.setitem(sys.modules, "ctypes", fake_ctypes) + assert tool_runner._get_kernel32() is kernel32 + + +def test_run_build_tool_logs_the_unfiltered_output( + capsys: pytest.CaptureFixture[str], tmp_path: Path +) -> None: + """The log gets every line, filtered or not, for idf.py's hint matcher.""" + log = tmp_path / "log" / "ninja_all_output.log" + _, out = _run( + capsys, + tmp_path, + "print('-- Component paths: /a')\nprint('\\x1b[1merror:\\x1b[0m boom')", + filter_lines=FILTER, + log_path=log, + ) + assert out == "\x1b[1merror:\x1b[0m boom\n" + assert log.read_text(encoding="utf-8").replace("\r\n", "\n") == ( + "-- Component paths: /a\nerror: boom\n" + ) diff --git a/tests/unit_tests/fixtures/espidf/closing_probe.py b/tests/unit_tests/fixtures/espidf/closing_probe.py deleted file mode 100644 index a77d5c8f28..0000000000 --- a/tests/unit_tests/fixtures/espidf/closing_probe.py +++ /dev/null @@ -1,11 +0,0 @@ -"""Leave a partial line behind and then close the stream under the runner. - -Run through ``esphome/espidf/runner.py`` by test_espidf_runner.py. Draining -cannot work here; the point is that the failure is reported rather than -raised out of the runner's cleanup, where it would bury the exit code. -""" - -import sys - -sys.stdout.write("partial before close") -sys.stdout.close() diff --git a/tests/unit_tests/fixtures/espidf/crashing_probe.py b/tests/unit_tests/fixtures/espidf/crashing_probe.py deleted file mode 100644 index bf434cc24e..0000000000 --- a/tests/unit_tests/fixtures/espidf/crashing_probe.py +++ /dev/null @@ -1,11 +0,0 @@ -"""Die part way through a line, the way a build that blows up does. - -Run through ``esphome/espidf/runner.py`` by test_espidf_runner.py. The -message has no trailing newline, so the runner's shim is holding it when -the process exits; nothing else will ever come to release it. -""" - -import sys - -sys.stdout.write("FATAL: ld returned 1 exit status") -sys.exit(2) diff --git a/tests/unit_tests/fixtures/espidf/filtering_probe.py b/tests/unit_tests/fixtures/espidf/filtering_probe.py deleted file mode 100644 index 04c2b2ed8c..0000000000 --- a/tests/unit_tests/fixtures/espidf/filtering_probe.py +++ /dev/null @@ -1,15 +0,0 @@ -"""Write a mix of noisy and useful build lines, without flushing. - -Run through ``esphome/espidf/runner.py`` by test_espidf_runner.py. The -runner's shim owns both the filtering and the flushing, so this script -only writes. -""" - -import sys - -sys.stdout.write("Project build complete.\n") -sys.stdout.write("Compiling main.cpp\n") -sys.stdout.write("-- Component paths: /a /b /c\n") -sys.stdout.write("[2/9] Building C object\n") -# No terminator, so the shim has to hold this one back. -sys.stdout.write("still going") diff --git a/tests/unit_tests/fixtures/espidf/formfeed_probe.py b/tests/unit_tests/fixtures/espidf/formfeed_probe.py deleted file mode 100644 index 727cda25ce..0000000000 --- a/tests/unit_tests/fixtures/espidf/formfeed_probe.py +++ /dev/null @@ -1,12 +0,0 @@ -"""Write a form feed part way through the output. - -Run through ``esphome/espidf/runner.py`` by test_espidf_runner.py. A form -feed is not a line terminator here, so everything written must still come -out, including the complete lines that follow it. -""" - -import sys - -sys.stdout.write("Compiling main.cpp\n") -sys.stdout.write("page one\x0cpage two\n") -sys.stdout.write("[2/9] Building C object\n") diff --git a/tests/unit_tests/fixtures/espidf/partial_noise_probe.py b/tests/unit_tests/fixtures/espidf/partial_noise_probe.py deleted file mode 100644 index 9c81f8eb7b..0000000000 --- a/tests/unit_tests/fixtures/espidf/partial_noise_probe.py +++ /dev/null @@ -1,10 +0,0 @@ -"""End on an unterminated line that the filter is supposed to drop. - -Run through ``esphome/espidf/runner.py`` by test_espidf_runner.py, to -check that releasing a held-back line still applies the filter. -""" - -import sys - -sys.stdout.write("Compiling main.cpp\n") -sys.stdout.write("Project build complete.") diff --git a/tests/unit_tests/fixtures/espidf/streaming_probe.py b/tests/unit_tests/fixtures/espidf/streaming_probe.py deleted file mode 100644 index c05741e311..0000000000 --- a/tests/unit_tests/fixtures/espidf/streaming_probe.py +++ /dev/null @@ -1,14 +0,0 @@ -"""Print one line, then stay alive so the caller can prove it streamed. - -Run through ``esphome/espidf/runner.py`` by test_espidf_runner.py. The -runner wraps stdout in its filtering shim, so this script deliberately -does not flush: the shim has to do it. The long sleep keeps the process -running, so anything the caller reads must have arrived while the build -was still going rather than at exit. -""" - -import sys -import time - -sys.stdout.write("Compiling main.cpp\n") -time.sleep(60) diff --git a/tests/unit_tests/test_espidf_clang_tidy.py b/tests/unit_tests/test_espidf_clang_tidy.py index 4cc445e29b..e0cfec88c8 100644 --- a/tests/unit_tests/test_espidf_clang_tidy.py +++ b/tests/unit_tests/test_espidf_clang_tidy.py @@ -1,5 +1,7 @@ """Tests for esphome.espidf.clang_tidy tidy-project generation.""" +# pylint: disable=protected-access + import json import os from pathlib import Path @@ -99,3 +101,39 @@ def test_idedata_from_tidy_project_missing_tu_raises(tmp_path) -> None: compile_commands.write_text(json.dumps([])) with pytest.raises(RuntimeError, match="tidy.cpp not found"): clang_tidy._idedata_from_tidy_project(compile_commands) + + +@pytest.mark.parametrize( + ("reconfigure_rcs", "error"), + [ + ((1,), "ESP-IDF CMake configure \\(discovery\\) failed"), + ((0, 1), "ESP-IDF CMake configure failed"), + ((0, 0), None), + ], + ids=["discovery", "full", "ok"], +) +def test_generate_compile_commands_configures_twice( + tmp_path: Path, reconfigure_rcs: tuple[int, ...], error: str | None +) -> None: + """Discovery configure, then a configure requiring what it found.""" + with ( + patch.object(clang_tidy, "_setup_core"), + patch.object(clang_tidy, "_convert_pio_libs", return_value={}), + patch.object(clang_tidy, "_write_tidy_project") as mock_write, + patch("esphome.espidf.toolchain.run_reconfigure", side_effect=reconfigure_rcs), + patch( + "esphome.build_gen.espidf.get_available_components", + return_value=["lwip", "esp_timer"], + ), + ): + if error: + with pytest.raises(RuntimeError, match=error): + clang_tidy._generate_compile_commands( + tmp_path, _settings(), tmp_path / "platformio.ini" + ) + return + result = clang_tidy._generate_compile_commands( + tmp_path, _settings(), tmp_path / "platformio.ini" + ) + assert result == tmp_path / "build" / "compile_commands.json" + assert mock_write.call_args_list[1].args[1] == ["esp_timer", "lwip"] diff --git a/tests/unit_tests/test_espidf_framework.py b/tests/unit_tests/test_espidf_framework.py index 9654c7cfc5..c785d56fa4 100644 --- a/tests/unit_tests/test_espidf_framework.py +++ b/tests/unit_tests/test_espidf_framework.py @@ -1627,8 +1627,8 @@ def test_ccache_env_opt_out_via_env(tmp_path: Path) -> None: # short-circuits before build_path is needed. p1, p2, p3 = _ccache_patches(tmp_path, "/usr/bin/ccache", None) with patch.dict("os.environ", {"IDF_CCACHE_ENABLE": "0"}, clear=True), p1, p2, p3: - # The canonical off spelling is exported: the raw value is inherited - # by idf.py, where a spelling like "disable" would read as truthy + # The canonical off spelling is exported, so every reader of the + # env sees the same answer assert _ccache_env() == {"IDF_CCACHE_ENABLE": "0"} @@ -1636,7 +1636,7 @@ def test_ccache_env_opt_in_without_binary( tmp_path: Path, caplog: pytest.LogCaptureFixture ) -> None: # Explicit IDF_CCACHE_ENABLE=1 forces it on; without a usable binary - # idf.py silently skips ccache, so this branch must say so out loud. + # IDF's CMake silently skips ccache, so this branch must say so out loud. p1, p2, p3 = _ccache_patches(tmp_path, None, tmp_path / "build") with patch.dict("os.environ", {"IDF_CCACHE_ENABLE": "1"}, clear=True), p1, p2, p3: env = _ccache_env() @@ -1669,7 +1669,7 @@ def test_ccache_env_opt_in_with_working_binary( def test_ccache_env_opt_in_with_rejected_binary( tmp_path: Path, caplog: pytest.LogCaptureFixture ) -> None: - # Forced on with a present-but-rejected binary: idf.py does its own + # Forced on with a present-but-rejected binary: IDF's CMake does its own # PATH lookup and uses it anyway; the warning must say so, not claim # the build runs without ccache. # A present but non-executable file: the real probe fails and logs @@ -1686,7 +1686,7 @@ def test_ccache_env_opt_in_with_rejected_binary( ): env = _ccache_env() assert env["IDF_CCACHE_ENABLE"] == "1" - assert "idf.py will use it anyway" in caplog.text + assert "the build will use it anyway" in caplog.text # Exactly one story: the resolver's contradictory "compiling without # ccache" must not precede it assert "compiling without ccache" not in caplog.text @@ -1715,7 +1715,7 @@ def test_ccache_env_idf_knob_unrecognized_warns_and_defers( tmp_path: Path, caplog: pytest.LogCaptureFixture ) -> None: """An unparsable IDF_CCACHE_ENABLE warns, defers to the shared resolver, - and is not forwarded to idf.py as truthy.""" + and is not forwarded to the build as truthy.""" p1, p2, p3 = _ccache_patches(tmp_path, "/usr/bin/ccache", tmp_path / "build") env_vars = {"IDF_CCACHE_ENABLE": "enabled"} with patch.dict("os.environ", env_vars, clear=True), p1, p2, p3: diff --git a/tests/unit_tests/test_espidf_runner.py b/tests/unit_tests/test_espidf_runner.py deleted file mode 100644 index 71ccfee1f2..0000000000 --- a/tests/unit_tests/test_espidf_runner.py +++ /dev/null @@ -1,315 +0,0 @@ -"""Tests for esphome.espidf.runner.""" - -from __future__ import annotations - -import io -import os -from pathlib import Path -import subprocess -import sys -import threading - -import pytest - -from esphome.espidf import runner - -# A flushing runner delivers the first line in well under a second; this is -# only ever waited out when the shim has gone back to buffering, so keep it -# just long enough to cover interpreter startup on a loaded CI machine. -FIRST_LINE_TIMEOUT = 10.0 - - -def _prepare_main( - monkeypatch: pytest.MonkeyPatch, probe: Path, *args: str -) -> tuple[io.BytesIO, io.TextIOWrapper]: - """Point ``runner.main()`` at *probe* with a buffered fake stdout. - - ``main`` rewrites ``sys.path``, ``sys.argv``, both std streams and - ``os.get_terminal_size``; every one of those is monkeypatched so it is - put back afterwards. The fake stdout is block buffered like a pipe, so - the caller can tell whether the shim flushed. The wrapper comes back with - the buffer because dropping it would close the buffer underneath us. - """ - buf = io.BytesIO() - stream = io.TextIOWrapper(buf, encoding="utf-8", newline="\n", line_buffering=False) - - monkeypatch.setattr(sys, "path", list(sys.path)) - monkeypatch.setattr(sys, "argv", ["runner.py", str(probe), *args]) - monkeypatch.setattr(sys, "stdout", stream) - monkeypatch.setattr(sys, "stderr", stream) - monkeypatch.setattr(os, "get_terminal_size", os.get_terminal_size) - - return buf, stream - - -def _run_main( - monkeypatch: pytest.MonkeyPatch, probe: Path, *args: str -) -> tuple[io.BytesIO, io.TextIOWrapper]: - """Run ``runner.main()`` against *probe* and expect a clean exit.""" - buf, stream = _prepare_main(monkeypatch, probe, *args) - assert runner.main() == 0 - return buf, stream - - -def test_main_filters_noise_and_flushes_each_write( - monkeypatch: pytest.MonkeyPatch, fixture_path: Path -) -> None: - """Useful lines reach the stream right away; noisy ones are dropped.""" - buf, _stream = _run_main( - monkeypatch, fixture_path / "espidf" / "filtering_probe.py" - ) - - # Read before any flush of our own: the shim has to have flushed. - output = buf.getvalue().decode("utf-8") - - assert "Compiling main.cpp\n" in output - assert "[2/9] Building C object\n" in output - # Matched by FILTER_IDF_LINES, so they never leave the runner. - assert "Project build complete." not in output - assert "-- Component paths:" not in output - # Held back until the end because no terminator arrived. - assert output.endswith("still going\n") - - -def test_main_keeps_output_after_a_form_feed( - monkeypatch: pytest.MonkeyPatch, fixture_path: Path -) -> None: - """A form feed is text, not a line break, so nothing after it is lost.""" - buf, _stream = _run_main(monkeypatch, fixture_path / "espidf" / "formfeed_probe.py") - - assert buf.getvalue().decode("utf-8") == ( - "Compiling main.cpp\npage one\x0cpage two\n[2/9] Building C object\n" - ) - - -def test_main_drains_a_partial_line_when_the_build_dies( - monkeypatch: pytest.MonkeyPatch, fixture_path: Path -) -> None: - """A build that stops mid line must still show that line. - - This is the whole point of draining: the message explaining why the - build failed is exactly the one most likely to arrive without a - trailing newline. - """ - buf, _stream = _prepare_main( - monkeypatch, fixture_path / "espidf" / "crashing_probe.py" - ) - - with pytest.raises(SystemExit) as excinfo: - runner.main() - - assert excinfo.value.code == 2 - assert buf.getvalue().decode("utf-8") == "FATAL: ld returned 1 exit status\n" - - -def test_main_reports_rather_than_raises_when_draining_fails( - monkeypatch: pytest.MonkeyPatch, - fixture_path: Path, - capfd: pytest.CaptureFixture[str], -) -> None: - """A stream that closed under us must not crash the runner's cleanup. - - The drain runs from a ``finally``, so an exception there would replace - whatever exit code the build was carrying back. - """ - _prepare_main(monkeypatch, fixture_path / "espidf" / "closing_probe.py") - - assert runner.main() == 0 - reported = capfd.readouterr().err - assert "Could not write out remaining output" in reported - # The held line has to come along; the stream it was meant for is gone. - assert "partial before close" in reported - - -def test_main_survives_a_drain_failure_with_nowhere_to_report_it( - monkeypatch: pytest.MonkeyPatch, fixture_path: Path -) -> None: - """With no real stderr to report to, cleanup still must not raise. - - ``sys.__stderr__`` is None on some interpreters, and ``print(file=None)`` - falls back to ``sys.stdout``, which here is the shim wrapping the stream - that just failed. - """ - monkeypatch.setattr(sys, "__stderr__", None) - _prepare_main(monkeypatch, fixture_path / "espidf" / "closing_probe.py") - - assert runner.main() == 0 - - -def test_main_still_filters_a_drained_partial_line( - monkeypatch: pytest.MonkeyPatch, fixture_path: Path -) -> None: - """Releasing a held line does not smuggle noise past the filter.""" - buf, _stream = _run_main( - monkeypatch, fixture_path / "espidf" / "partial_noise_probe.py" - ) - - assert buf.getvalue().decode("utf-8") == "Compiling main.cpp\n" - - -def test_main_keeps_everything_in_verbose_mode( - monkeypatch: pytest.MonkeyPatch, fixture_path: Path -) -> None: - """``-v`` turns the filter off so the noisy lines survive.""" - buf, _stream = _run_main( - monkeypatch, fixture_path / "espidf" / "filtering_probe.py", "-v" - ) - - output = buf.getvalue().decode("utf-8") - - assert "Project build complete.\n" in output - assert "-- Component paths: /a /b /c\n" in output - # With no filter there is no line buffering, so the partial line goes - # straight through as well. - assert output.endswith("still going") - - -def test_runner_streams_output_before_the_build_finishes( - fixture_path: Path, probe_env: dict[str, str] -) -> None: - """The runner must flush, or a dashboard build looks frozen. - - ``toolchain.py`` spawns the runner as a plain script with no ``-u``, and - hands it a pipe when esphome itself is running under the dashboard. A - pipe is block buffered, so without a flush in the shim's ``write()`` the - output sits in the child until 8 KiB piles up or the build ends. - """ - runner_py = Path(runner.__file__) - probe = fixture_path / "espidf" / "streaming_probe.py" - - with subprocess.Popen( - [sys.executable, str(runner_py), str(probe)], - stdout=subprocess.PIPE, - # Keep stderr: if the runner dies on startup, its traceback is the - # only clue about why no line showed up. - stderr=subprocess.PIPE, - env=probe_env, - text=True, - ) as proc: - assert proc.stdout is not None - assert proc.stderr is not None - first_line: list[str] = [] - reader = threading.Thread( - target=lambda: first_line.append(proc.stdout.readline()), daemon=True - ) - try: - reader.start() - reader.join(FIRST_LINE_TIMEOUT) - still_running = proc.poll() is None - - # The probe sleeps for a minute after writing, so reaching us at - # all means the line was flushed rather than released at exit. - assert first_line == ["Compiling main.cpp\n"], ( - f"runner stderr: {'' if still_running else proc.stderr.read()}" - ) - assert still_running - finally: - proc.kill() - proc.wait() - # Join before leaving the block, so the reader is done rather than - # racing ``Popen`` closing the pipe under it. - reader.join(1.0) - - -class _FakeKernel32: - """Stand-in for the Windows kernel32 console code page calls.""" - - def __init__(self, input_cp: int, output_cp: int) -> None: - self.input_cp = input_cp - self.output_cp = output_cp - self.calls: list[tuple[str, int]] = [] - - def GetConsoleCP(self) -> int: # noqa: N802 - return self.input_cp - - def GetConsoleOutputCP(self) -> int: # noqa: N802 - return self.output_cp - - def SetConsoleCP(self, codepage: int) -> int: # noqa: N802 - self.calls.append(("SetConsoleCP", codepage)) - self.input_cp = codepage - return 1 - - def SetConsoleOutputCP(self, codepage: int) -> int: # noqa: N802 - self.calls.append(("SetConsoleOutputCP", codepage)) - self.output_cp = codepage - return 1 - - -def test_main_runs_the_build_with_a_utf8_console( - monkeypatch: pytest.MonkeyPatch, fixture_path: Path -) -> None: - """An attached console is switched to UTF-8 and then put back.""" - kernel32 = _FakeKernel32(850, 850) - monkeypatch.setattr(runner, "_get_kernel32", lambda: kernel32) - - _run_main(monkeypatch, fixture_path / "espidf" / "filtering_probe.py") - - assert kernel32.calls == [ - ("SetConsoleCP", runner.UTF8_CODEPAGE), - ("SetConsoleOutputCP", runner.UTF8_CODEPAGE), - ("SetConsoleCP", 850), - ("SetConsoleOutputCP", 850), - ] - - -def test_main_restores_the_console_when_the_build_dies( - monkeypatch: pytest.MonkeyPatch, fixture_path: Path -) -> None: - """A failing build must not leave the user's console on UTF-8.""" - kernel32 = _FakeKernel32(437, 437) - monkeypatch.setattr(runner, "_get_kernel32", lambda: kernel32) - _prepare_main(monkeypatch, fixture_path / "espidf" / "crashing_probe.py") - - with pytest.raises(SystemExit): - runner.main() - - assert (kernel32.input_cp, kernel32.output_cp) == (437, 437) - - -def test_main_restores_the_console_when_the_switch_fails_part_way( - monkeypatch: pytest.MonkeyPatch, fixture_path: Path -) -> None: - """A failed output page switch must not strand the changed input page.""" - kernel32 = _FakeKernel32(850, 850) - - def _refuse(codepage: int) -> int: - kernel32.calls.append(("SetConsoleOutputCP", codepage)) - return 0 - - kernel32.SetConsoleOutputCP = _refuse # type: ignore[method-assign] - monkeypatch.setattr(runner, "_get_kernel32", lambda: kernel32) - - _run_main(monkeypatch, fixture_path / "espidf" / "filtering_probe.py") - - assert kernel32.input_cp == 850 - assert kernel32.calls[-2:] == [("SetConsoleCP", 850), ("SetConsoleOutputCP", 850)] - - -def test_main_leaves_the_console_alone_when_there_is_none( - monkeypatch: pytest.MonkeyPatch, fixture_path: Path -) -> None: - """Without a console the code page calls return 0 and nothing is set.""" - kernel32 = _FakeKernel32(0, 0) - monkeypatch.setattr(runner, "_get_kernel32", lambda: kernel32) - - _run_main(monkeypatch, fixture_path / "espidf" / "filtering_probe.py") - - assert kernel32.calls == [] - - -def test_main_leaves_a_console_already_on_utf8_alone( - monkeypatch: pytest.MonkeyPatch, fixture_path: Path -) -> None: - """An overlapping build must not save UTF-8 as the page to restore.""" - kernel32 = _FakeKernel32(runner.UTF8_CODEPAGE, runner.UTF8_CODEPAGE) - monkeypatch.setattr(runner, "_get_kernel32", lambda: kernel32) - - _run_main(monkeypatch, fixture_path / "espidf" / "filtering_probe.py") - - assert kernel32.calls == [] - - -@pytest.mark.skipif(sys.platform == "win32", reason="kernel32 exists on Windows") -def test_get_kernel32_is_none_off_windows() -> None: - assert runner._get_kernel32() is None diff --git a/tests/unit_tests/test_espidf_toolchain.py b/tests/unit_tests/test_espidf_toolchain.py index 3e1e88d66d..ee21cd77a9 100644 --- a/tests/unit_tests/test_espidf_toolchain.py +++ b/tests/unit_tests/test_espidf_toolchain.py @@ -8,11 +8,12 @@ import json import os from pathlib import Path import subprocess -from unittest.mock import patch +from unittest.mock import call, patch import pytest -from esphome.components.esp32.const import KEY_ESP32, KEY_VARIANT +from esphome.components.esp32.const import KEY_ESP32, KEY_IDF_VERSION, KEY_VARIANT +import esphome.config_validation as cv from esphome.const import ( CONF_COMPILE_PROCESS_LIMIT, CONF_ESPHOME, @@ -104,11 +105,27 @@ def _setup_build(setup_core: Path) -> tuple[Path, Path]: """Point CORE at a build dir; return (compile_commands, idedata cache) paths.""" CORE.name = "test" CORE.build_path = setup_core / "build" / "test" + CORE.data.setdefault(KEY_ESP32, {})[KEY_IDF_VERSION] = cv.Version(5, 5, 5) compile_commands = CORE.relative_build_path("build", "compile_commands.json") cache = CORE.relative_internal_path("idedata", "test.json") return compile_commands, cache +@contextmanager +def _up_to_date_compile(ninja_side_effect=None) -> Iterator[tuple]: + """Patch run_compile's staleness checks to "up to date"; yield the ninja + and print_summary mocks.""" + with ( + patch.object(toolchain, "need_reconfigure", return_value=False), + patch.object(toolchain, "_cache_entries_changed", return_value=False), + patch.object( + toolchain, "_run_ninja", return_value=0, side_effect=ninja_side_effect + ) as mock_ninja, + patch.object(toolchain, "print_summary") as mock_summary, + ): + yield mock_ninja, mock_summary + + def test_has_outdated_files_detects_exclusion_change(setup_core: Path) -> None: """A newer exclude_components.esphomeinternal stamp forces a reconfigure so components that leave the exclusion set get rediscovered.""" @@ -279,28 +296,6 @@ def test_get_cmake_output_missing_build_does_not_resolve_idf_env( mock_run.assert_not_called() -def test_run_idf_py_jobs_sets_build_jobs_env(setup_core: Path) -> None: - """The jobs argument is exported to idf.py as IDF_PY_BUILD_JOBS.""" - _setup_build(setup_core) - - with ( - patch.object(toolchain, "_get_idf_path", return_value=Path("/idf")), - patch.object(toolchain, "_get_idf_env", return_value={"PATH": "/bin"}), - patch.object(toolchain, "_get_idf_tool", return_value="python"), - patch.object(toolchain.subprocess, "run") as mock_run, - ): - mock_run.return_value.returncode = 0 - - toolchain.run_idf_py("build", jobs=2) - env = mock_run.call_args.kwargs["env"] - assert env["IDF_PY_BUILD_JOBS"] == "2" - assert env["PATH"] == "/bin" - - toolchain.run_idf_py("build") - env = mock_run.call_args.kwargs["env"] - assert "IDF_PY_BUILD_JOBS" not in env - - def test_run_compile_restamps_cmakecache_after_discovery(setup_core: Path) -> None: """After a successful discovery reconfigure the reference CMakeCache.txt is restamped; cmake does not rewrite it when only properties or plain @@ -325,7 +320,7 @@ def test_run_compile_restamps_cmakecache_after_discovery(setup_core: Path) -> No ), patch("esphome.build_gen.espidf.write_project"), patch.object(toolchain, "run_reconfigure", return_value=0), - patch.object(toolchain, "run_idf_py", return_value=0), + patch.object(toolchain, "_run_ninja", return_value=0), patch.object(toolchain, "print_summary"), ): assert toolchain.run_compile(config, verbose=False) == 0 @@ -350,7 +345,7 @@ def test_run_compile_discovery_without_cmakecache(setup_core: Path) -> None: ), patch("esphome.build_gen.espidf.write_project"), patch.object(toolchain, "run_reconfigure", return_value=0), - patch.object(toolchain, "run_idf_py", return_value=0), + patch.object(toolchain, "_run_ninja", return_value=0), patch.object(toolchain, "print_summary"), ): assert toolchain.run_compile(config, verbose=False) == 0 @@ -376,7 +371,7 @@ def test_run_compile_reconfigures_after_full_write_outside_testing_mode( def record_write(minimal: bool = False, builtin_components=None) -> None: calls.append(("write_project", minimal)) - def record_reconfigure() -> int: + def record_reconfigure(verbose: bool = False) -> int: nonlocal reconfigures reconfigures += 1 calls.append(("run_reconfigure",)) @@ -391,7 +386,7 @@ def test_run_compile_reconfigures_after_full_write_outside_testing_mode( ), patch("esphome.build_gen.espidf.write_project", side_effect=record_write), patch.object(toolchain, "run_reconfigure", side_effect=record_reconfigure), - patch.object(toolchain, "run_idf_py", return_value=0) as mock_build, + patch.object(toolchain, "_run_ninja", return_value=0) as mock_build, patch.object(toolchain, "print_summary"), ): assert not CORE.testing_mode @@ -421,7 +416,7 @@ def _record_compile_calls( calls: list[tuple] = [] rcs = iter(reconfigure_rcs) - def record_reconfigure() -> int: + def record_reconfigure(verbose: bool = False) -> int: calls.append(("run_reconfigure",)) return next(rcs, 0) @@ -431,6 +426,11 @@ def _record_compile_calls( def record_save(components: list[str]) -> None: calls.append(("save", components)) + def record_ninja(target: str, **kwargs: object) -> int: + if target == "all": + calls.append(("build",)) + return 0 + with ( patch.object(toolchain, "need_reconfigure", return_value=True), patch.object(toolchain, "load_cached_builtin_components", return_value=cached), @@ -443,11 +443,7 @@ def _record_compile_calls( patch.object( toolchain, "_builtin_component_cache_path", return_value=cache_file ), - patch.object( - toolchain, - "run_idf_py", - side_effect=lambda *a, **kw: calls.append(("build",)) or 0, - ), + patch.object(toolchain, "_run_ninja", side_effect=record_ninja), patch.object(toolchain, "print_summary"), ): rc = toolchain.run_compile({CONF_ESPHOME: {}}, verbose=False) @@ -630,19 +626,21 @@ def test_component_cache_ignores_corrupt_file(setup_core: Path, tmp_path: Path) assert toolchain.load_cached_builtin_components() is None -def test_run_compile_passes_compile_process_limit(setup_core: Path) -> None: - """compile_process_limit is forwarded to run_idf_py as the job limit.""" +@pytest.mark.parametrize("limit", [1, None]) +def test_run_compile_passes_compile_process_limit( + setup_core: Path, limit: int | None +) -> None: + """compile_process_limit is the job limit for both ninja runs.""" _setup_build(setup_core) - config = {CONF_ESPHOME: {CONF_COMPILE_PROCESS_LIMIT: 1}} + esphome = {} if limit is None else {CONF_COMPILE_PROCESS_LIMIT: limit} - with ( - patch.object(toolchain, "need_reconfigure", return_value=False), - patch.object(toolchain, "run_idf_py", return_value=0) as mock_run, - patch.object(toolchain, "print_summary"), - ): - assert toolchain.run_compile(config, verbose=False) == 0 + with _up_to_date_compile() as (mock_run, _): + assert toolchain.run_compile({CONF_ESPHOME: esphome}, verbose=False) == 0 - mock_run.assert_called_once_with("build", "size", jobs=1) + assert mock_run.call_args_list == [ + call("all", verbose=False, jobs=limit, progress=True), + call("size", verbose=False, jobs=limit, extra_env=toolchain._size_env()), + ] def test_run_compile_passes_size_summary_paths(setup_core: Path) -> None: @@ -650,14 +648,8 @@ def test_run_compile_passes_size_summary_paths(setup_core: Path) -> None: ELF from get_built_elf_path, which must stay in lockstep with the project() name in the generated CMakeLists.""" _setup_build(setup_core) - config = {CONF_ESPHOME: {}} - - with ( - patch.object(toolchain, "need_reconfigure", return_value=False), - patch.object(toolchain, "run_idf_py", return_value=0), - patch.object(toolchain, "print_summary") as mock_summary, - ): - assert toolchain.run_compile(config, verbose=False) == 0 + with _up_to_date_compile() as (_, mock_summary): + assert toolchain.run_compile({CONF_ESPHOME: {}}, verbose=False) == 0 mock_summary.assert_called_once_with( CORE.relative_build_path("build", "esp_idf_size.json"), @@ -682,21 +674,6 @@ def test_create_elf_copy_missing_source(setup_core: Path) -> None: assert toolchain.create_elf_copy() is False -def test_run_compile_without_compile_process_limit(setup_core: Path) -> None: - """When no compile_process_limit is set, no job limit is passed to idf.py.""" - _setup_build(setup_core) - config = {CONF_ESPHOME: {}} - - with ( - patch.object(toolchain, "need_reconfigure", return_value=False), - patch.object(toolchain, "run_idf_py", return_value=0) as mock_run, - patch.object(toolchain, "print_summary"), - ): - assert toolchain.run_compile(config, verbose=False) == 0 - - mock_run.assert_called_once_with("build", "size", jobs=None) - - def test_run_compile_writes_the_pch_checksum_before_the_build( setup_core: Path, ) -> None: @@ -704,11 +681,7 @@ def test_run_compile_writes_the_pch_checksum_before_the_build( order: list[str] = [] with ( - patch.object(toolchain, "need_reconfigure", return_value=False), - patch.object( - toolchain, "run_idf_py", side_effect=lambda *a, **k: order.append("build") - ), - patch.object(toolchain, "print_summary"), + _up_to_date_compile(lambda *a, **k: order.append("build") or 0), patch( "esphome.build_gen.espidf.write_pch_checksum", side_effect=lambda: order.append("checksum"), @@ -727,6 +700,404 @@ def test_get_core_framework_version_from_core_data(): assert toolchain._get_core_framework_version() == "5.5.4" +@contextmanager +def _fake_tools(env: dict[str, str] | None = None) -> Iterator: + """Stub the IDF env and tool lookup; yield the run_build_tool mock.""" + with ( + patch.object( + toolchain, + "_get_idf_env", + return_value={"PATH": "/bin", "IDF_CCACHE_ENABLE": "0", **(env or {})}, + ), + patch.object(toolchain, "_get_idf_tool", side_effect=lambda n: f"/tools/{n}"), + patch.object(toolchain, "run_build_tool", return_value=0) as mock_run, + ): + yield mock_run + + +def test_run_reconfigure_cmake_argv_matches_idf_py(setup_core: Path) -> None: + """The configure is the command idf.py runs. + + IDF 5.5.5 tools/idf_py_actions/tools.py ensure_build_directory: cmake + -G Ninja -DPYTHON_DEPS_CHECKED=1 -DPYTHON= -DESP_PLATFORM=1, + then the -D entries (SDKCONFIG, CCACHE_ENABLE), then the real project + path, in the build dir. IDF 6.1 adds -B , the same as the cwd. + """ + _setup_build(setup_core) + sdkconfig = CORE.relative_build_path("sdkconfig.test") + sdkconfig.parent.mkdir(parents=True) + sdkconfig.write_text("") + project = os.path.realpath(CORE.build_path) + + with _fake_tools() as mock_run: + assert toolchain.run_reconfigure() == 0 + + assert mock_run.call_args.args[0] == [ + "/tools/cmake", + "-G", + "Ninja", + "-DPYTHON_DEPS_CHECKED=1", + "-DPYTHON=/tools/python", + "-DESP_PLATFORM=1", + f"-DSDKCONFIG={sdkconfig}", + "-DCCACHE_ENABLE=0", + project, + ] + kwargs = mock_run.call_args.kwargs + assert kwargs["cwd"] == Path(project) / "build" + assert kwargs["cwd"].is_dir() + assert kwargs["env"]["CLICOLOR_FORCE"] == "1" + assert kwargs["filter_lines"] is toolchain.FILTER_IDF_LINES + + +@pytest.mark.parametrize(("ccache", "expected"), [("1", "True"), ("0", "False")]) +def test_run_reconfigure_cmake_argv_matches_idf6_py( + setup_core: Path, ccache: str, expected: str +) -> None: + """IDF 6 idf.py adds -B and formats CCACHE_ENABLE as True/False. + + IDF 6.1 tools/idf_py_actions/tools.py ensure_build_directory. + """ + _setup_build(setup_core) + CORE.data[KEY_ESP32][KEY_IDF_VERSION] = cv.Version(6, 1, 0) + build_dir = Path(os.path.realpath(CORE.build_path)) / "build" + + with _fake_tools({"IDF_CCACHE_ENABLE": ccache}) as mock_run: + assert toolchain.run_reconfigure() == 0 + + assert mock_run.call_args.args[0] == [ + "/tools/cmake", + "-G", + "Ninja", + "-B", + str(build_dir), + "-DPYTHON_DEPS_CHECKED=1", + "-DPYTHON=/tools/python", + "-DESP_PLATFORM=1", + f"-DCCACHE_ENABLE={expected}", + str(build_dir.parent), + ] + + +def test_run_reconfigure_without_sdkconfig_or_filter(setup_core: Path) -> None: + """No sdkconfig file means no SDKCONFIG entry; -v shows every line.""" + _setup_build(setup_core) + with _fake_tools() as mock_run: + assert toolchain.run_reconfigure(verbose=True) == 0 + cmd = mock_run.call_args.args[0] + assert not any(arg.startswith("-DSDKCONFIG=") for arg in cmd) + assert cmd[-2] == "-DCCACHE_ENABLE=0" + assert mock_run.call_args.kwargs["filter_lines"] is None + + +def test_run_reconfigure_failure_removes_cmakecache( + setup_core: Path, caplog: pytest.LogCaptureFixture +) -> None: + """Like idf.py, a failed configure must not leave a cache that looks valid.""" + _setup_build(setup_core) + cache = Path(os.path.realpath(CORE.build_path)) / "build" / "CMakeCache.txt" + cache.parent.mkdir(parents=True) + cache.write_text("") + with ( + _fake_tools() as mock_run, + patch.object(toolchain, "_print_hints") as mock_hints, + ): + mock_run.return_value = 4 + assert toolchain.run_reconfigure() == 4 + assert not cache.exists() + assert "CMake configure failed with exit code 4" in caplog.text + log_path = mock_run.call_args.kwargs["log_path"] + assert log_path == cache.parent / "log" / "cmake_output.log" + mock_hints.assert_called_once_with(log_path) + + +@pytest.mark.parametrize( + ("value", "expected"), + [ + ("1", "1"), + ("on", "1"), + (" Yes ", "1"), + ("0", "0"), + ("", "0"), + ("disable", "0"), + ], +) +def test_cache_entries_ccache_follows_click_booleans( + setup_core: Path, value: str, expected: str +) -> None: + """IDF_CCACHE_ENABLE maps to CCACHE_ENABLE the way idf.py's click flag does.""" + _setup_build(setup_core) + with _fake_tools({"IDF_CCACHE_ENABLE": value}): + assert toolchain._cache_entries() == {"CCACHE_ENABLE": expected} + + +def test_parse_cmakecache(tmp_path: Path) -> None: + cache = tmp_path / "CMakeCache.txt" + cache.write_text( + "# comment\n" + "// help text\n" + "SDKCONFIG:UNINITIALIZED=/a/sdkconfig.test\n" + "CCACHE_ENABLE:UNINITIALIZED=1\n" + "FLAGS:STRING=-DX=1\n" + "\n" + ) + assert toolchain._parse_cmakecache(cache) == { + "SDKCONFIG": "/a/sdkconfig.test", + "CCACHE_ENABLE": "1", + "FLAGS": "-DX=1", + } + + +# The constant -D entries every configure writes (see _configure_defines). +_CONFIGURED = ( + "PYTHON_DEPS_CHECKED:UNINITIALIZED=1\n" + "PYTHON:UNINITIALIZED=/tools/python\n" + "ESP_PLATFORM:UNINITIALIZED=1\n" +) + + +@pytest.mark.parametrize( + ("cache_text", "expected"), + [ + (None, True), + (_CONFIGURED + "CCACHE_ENABLE:UNINITIALIZED=0\n", False), + (_CONFIGURED + "CCACHE_ENABLE:UNINITIALIZED=1\n", True), + ( + _CONFIGURED.replace("/tools/python", "/old/python") + + "CCACHE_ENABLE:UNINITIALIZED=0\n", + True, + ), + (_CONFIGURED, True), + ], + ids=["no_cache", "same", "changed", "python_moved", "missing"], +) +def test_cache_entries_changed( + setup_core: Path, cache_text: str | None, expected: bool +) -> None: + """Mirrors idf.py's _new_cmakecache_entries, plus the cached PYTHON.""" + _setup_build(setup_core) + if cache_text is not None: + cache = Path(os.path.realpath(CORE.build_path)) / "build" / "CMakeCache.txt" + cache.parent.mkdir(parents=True) + cache.write_text(cache_text) + with _fake_tools(): + assert toolchain._cache_entries_changed() is expected + + +@pytest.mark.parametrize( + ("version", "size_ng"), [(cv.Version(5, 5, 5), True), (cv.Version(6, 1, 0), False)] +) +def test_size_env(setup_core: Path, version: cv.Version, size_ng: bool) -> None: + """IDF 5.x idf.py sets ESP_IDF_SIZE_NG; 6.x dropped it (core_ext.py size_target).""" + CORE.data[KEY_ESP32] = {KEY_IDF_VERSION: version} + env = toolchain._size_env() + assert env["ESP_IDF_SIZE_FORCE_TERMINAL"] == "1" + assert env["SIZE_OUTPUT_FORMAT"] == "default" + assert ("ESP_IDF_SIZE_NG" in env) is size_ng + + +def test_run_ninja_matches_idf_py_run_target(setup_core: Path) -> None: + """The command is ninja [-j N] [-v] in the build dir.""" + _setup_build(setup_core) + with _fake_tools() as mock_run: + assert ( + toolchain._run_ninja( + "size", verbose=True, jobs=2, progress=True, extra_env={"A": "b"} + ) + == 0 + ) + assert mock_run.call_args.args[0] == ["/tools/ninja", "-j", "2", "-v", "size"] + kwargs = mock_run.call_args.kwargs + assert kwargs["cwd"] == Path(os.path.realpath(CORE.build_path)) / "build" + assert kwargs["env"]["CLICOLOR_FORCE"] == "1" + assert kwargs["env"]["A"] == "b" + # -v shows every line as it comes, as idf.py does + assert kwargs["filter_lines"] is None + assert kwargs["progress"] is False + + +def test_run_ninja_filters_and_reports_failure( + setup_core: Path, caplog: pytest.LogCaptureFixture +) -> None: + _setup_build(setup_core) + with ( + _fake_tools() as mock_run, + patch.object(toolchain, "_print_hints") as mock_hints, + ): + mock_run.return_value = 1 + assert toolchain._run_ninja("all", verbose=False, jobs=None, progress=True) == 1 + log_path = mock_run.call_args.kwargs["log_path"] + assert log_path.name == "ninja_all_output.log" + mock_hints.assert_called_once_with(log_path) + assert mock_run.call_args.args[0] == ["/tools/ninja", "all"] + assert mock_run.call_args.kwargs["filter_lines"] is toolchain.FILTER_IDF_LINES + assert mock_run.call_args.kwargs["progress"] is True + assert "ninja all failed with exit code 1" in caplog.text + + +@pytest.mark.parametrize("reconfigure_rc", [0, 5]) +def test_run_compile_reconfigures_when_cache_entries_change( + setup_core: Path, reconfigure_rc: int +) -> None: + """A changed -D entry (ccache toggled) reconfigures, as idf.py build did.""" + _setup_build(setup_core) + with ( + patch.object(toolchain, "need_reconfigure", return_value=False), + patch.object(toolchain, "_cache_entries_changed", return_value=True), + patch.object( + toolchain, "run_reconfigure", return_value=reconfigure_rc + ) as mock_reconfigure, + patch.object(toolchain, "_run_ninja", return_value=0) as mock_ninja, + patch.object(toolchain, "print_summary"), + ): + assert toolchain.run_compile({CONF_ESPHOME: {}}, verbose=True) == reconfigure_rc + mock_reconfigure.assert_called_once_with(True) + assert mock_ninja.called is (reconfigure_rc == 0) + + +@pytest.mark.parametrize("failing", ["all", "size"]) +def test_run_compile_stops_on_ninja_failure(setup_core: Path, failing: str) -> None: + """A failed build skips size; either failure skips the summary.""" + _setup_build(setup_core) + with _up_to_date_compile(lambda target, **kw: 7 if target == failing else 0) as ( + mock_ninja, + mock_summary, + ): + assert toolchain.run_compile({CONF_ESPHOME: {}}, verbose=False) == 7 + targets = [c.args[0] for c in mock_ninja.call_args_list] + assert targets == (["all"] if failing == "all" else ["all", "size"]) + mock_summary.assert_not_called() + + +@pytest.mark.parametrize("memory_ld_rc", [0, 3]) +def test_run_compile_testing_mode_builds_memory_ld_first( + setup_core: Path, memory_ld_rc: int +) -> None: + """Testing mode builds and patches memory.ld before the main build.""" + _setup_build(setup_core) + CORE.testing_mode = True + targets: list[str] = [] + + def record(target: str, **kwargs: object) -> int: + targets.append(target) + return memory_ld_rc if target.endswith("memory.ld") else 0 + + with ( + _up_to_date_compile(record), + patch.object(toolchain, "_patch_memory_segments") as mock_patch, + ): + assert toolchain.run_compile({CONF_ESPHOME: {}}, verbose=False) == memory_ld_rc + memory_ld = str(Path("esp-idf", "esp_system", "ld", "memory.ld")) + if memory_ld_rc: + assert targets == [memory_ld] + mock_patch.assert_not_called() + else: + assert targets == [memory_ld, "all", "size"] + mock_patch.assert_called_once() + + +@pytest.mark.parametrize("version", [cv.Version(5, 5, 5), cv.Version(6, 1, 0)]) +@pytest.mark.parametrize( + ("env", "expected"), + [ + ({}, {"CLICOLOR_FORCE": "1", "FORCE_COLOR": "1"}), + ({"CLICOLOR_FORCE": "0"}, {"CLICOLOR_FORCE": "0", "FORCE_COLOR": "1"}), + ({"NO_COLOR": "1"}, {}), + ], + ids=["default", "user-value", "no-color"], +) +def test_tool_env_colors( + setup_core: Path, version: cv.Version, env: dict[str, str], expected: dict[str, str] +) -> None: + """IDF 6 RunTool.__call__ color handling, used for every IDF version.""" + _setup_build(setup_core) + CORE.data[KEY_ESP32][KEY_IDF_VERSION] = version + with _fake_tools(env): + result = toolchain._tool_env() + colors = {k: v for k, v in result.items() if k in ("CLICOLOR_FORCE", "FORCE_COLOR")} + assert colors == expected + + +@pytest.mark.parametrize( + ("limit", "env_value", "expected"), + [ + (2, "8", 2), + (None, "8", 8), + (None, "", None), + (None, None, None), + ], + ids=["limit-wins", "env", "empty-env", "unset"], +) +def test_build_jobs_honors_idf_py_build_jobs( + monkeypatch: pytest.MonkeyPatch, + limit: int | None, + env_value: str | None, + expected: int | None, +) -> None: + """compile_process_limit first, else IDF_PY_BUILD_JOBS as idf.py read it.""" + if env_value is None: + monkeypatch.delenv("IDF_PY_BUILD_JOBS", raising=False) + else: + monkeypatch.setenv("IDF_PY_BUILD_JOBS", env_value) + esphome = {} if limit is None else {CONF_COMPILE_PROCESS_LIMIT: limit} + assert toolchain._build_jobs({CONF_ESPHOME: esphome}) == expected + + +@pytest.mark.parametrize("value", ["0", "-1", "many"]) +def test_build_jobs_rejects_invalid_idf_py_build_jobs( + monkeypatch: pytest.MonkeyPatch, value: str +) -> None: + """Like idf.py, a value that is not a positive integer is an error.""" + monkeypatch.setenv("IDF_PY_BUILD_JOBS", value) + with pytest.raises(EsphomeError, match="positive integer"): + toolchain._build_jobs({CONF_ESPHOME: {}}) + + +@contextmanager +def _hint_env(tmp_path: Path, **run_kwargs: object) -> Iterator: + with ( + patch.object(toolchain, "_get_idf_tool", return_value="/py"), + patch.object(toolchain, "_get_idf_path", return_value=tmp_path / "idf"), + patch.object(toolchain, "_get_idf_env", return_value={}), + patch.object(toolchain.subprocess, "run", **run_kwargs) as mock_run, + ): + yield mock_run + + +def test_print_hints_shows_idf_advice( + tmp_path: Path, caplog: pytest.LogCaptureFixture +) -> None: + """The hint step runs IDF's own generate_hints on the failed tool's log.""" + log = tmp_path / "ninja_all_output.log" + done = subprocess.CompletedProcess([], 0, "HINT: the binary is too big\n", "") + with _hint_env(tmp_path, return_value=done) as mock_run: + toolchain._print_hints(log) + cmd = mock_run.call_args.args[0] + assert cmd[0] == "/py" + assert "generate_hints" in cmd[2] + assert cmd[3:] == [str(tmp_path / "idf" / "tools"), str(log)] + assert "HINT: the binary is too big" in caplog.text + + +@pytest.mark.parametrize( + "run_kwargs", + [ + {"return_value": subprocess.CompletedProcess([], 1, "", "ImportError")}, + {"return_value": subprocess.CompletedProcess([], 0, " \n", "")}, + {"side_effect": subprocess.TimeoutExpired("py", 60)}, + {"side_effect": OSError("gone")}, + ], + ids=["script-fails", "no-hints", "timeout", "oserror"], +) +def test_print_hints_never_fails_the_build( + tmp_path: Path, caplog: pytest.LogCaptureFixture, run_kwargs: dict +) -> None: + """No hints is fine; the hint step must not raise or add warnings.""" + with caplog.at_level("WARNING"), _hint_env(tmp_path, **run_kwargs): + toolchain._print_hints(tmp_path / "log") + assert caplog.records == [] + + def test_get_cmake_cache_value_reads_the_configured_cache(setup_core: Path) -> None: _setup_build(setup_core) output = "CMAKE_OBJDUMP:FILEPATH=/tools/objdump\nESPHOME_PCH:BOOL=OFF\n"