From 6bcbdd79c36f2e5a0e04e396c82c5c66c423fb93 Mon Sep 17 00:00:00 2001 From: "J. Nick Koston" Date: Wed, 5 Aug 2026 08:43:14 -0500 Subject: [PATCH] [core] Lift the log line processor into esphome/stacktrace.py (#18076) --- esphome/__main__.py | 15 +- esphome/api_client.py | 64 +------- esphome/platform_hooks.py | 7 +- esphome/platformio/toolchain.py | 53 ++++-- esphome/stacktrace.py | 105 ++++++++++++ .../analyze_memory/test_build_artifacts.py | 4 +- tests/unit_tests/test_api_client.py | 135 +--------------- tests/unit_tests/test_lazy_imports.py | 26 ++- tests/unit_tests/test_main.py | 32 ++++ tests/unit_tests/test_platformio_toolchain.py | 47 +++++- tests/unit_tests/test_stacktrace.py | 151 ++++++++++++++++++ 11 files changed, 414 insertions(+), 225 deletions(-) create mode 100644 esphome/stacktrace.py create mode 100644 tests/unit_tests/test_stacktrace.py diff --git a/esphome/__main__.py b/esphome/__main__.py index 4b380f1335..178546de68 100644 --- a/esphome/__main__.py +++ b/esphome/__main__.py @@ -58,6 +58,7 @@ from esphome.core import CORE, EsphomeError, coroutine from esphome.enum import StrEnum from esphome.helpers import get_bool_env, indent, is_ip_address from esphome.log import AnsiFore, color, setup_log +from esphome.stacktrace import LogLineProcessor from esphome.types import ConfigType from esphome.upload_targets import PortType, get_port_type from esphome.util import ( @@ -631,11 +632,9 @@ def run_miniterm(config: ConfigType, port: str, args) -> int: return 1 _LOGGER.info("Starting log output from %s with baud rate %s", port, baud_rate) - # Stacktrace analysis is optional; platform_hooks owns resolution - # and the user-facing messages. - process_stacktrace = platform_hooks.get_stacktrace_handler(CORE.target_platform) - - backtrace_state = False + # Decoder resolution, crash isolation, and disable-after-failure + # all live in LogLineProcessor, shared with the API log path. + processor = LogLineProcessor(config, CORE.target_platform) ser = serial.Serial() ser.baudrate = baud_rate ser.port = port @@ -675,11 +674,7 @@ def run_miniterm(config: ConfigType, port: str, args) -> int: "utf8", "backslashreplace" ) safe_print(parser.parse_line(line, time_str)) - - if process_stacktrace is not None: - backtrace_state = process_stacktrace( - config, line, backtrace_state - ) + processor.process_line(line) except serial.SerialException: _LOGGER.error("Serial port closed!") return 0 diff --git a/esphome/api_client.py b/esphome/api_client.py index 0ee2a7bed3..cd80a64fb9 100644 --- a/esphome/api_client.py +++ b/esphome/api_client.py @@ -3,7 +3,6 @@ from __future__ import annotations import asyncio from contextlib import suppress from datetime import datetime -import importlib import logging from typing import TYPE_CHECKING, Any import warnings @@ -18,6 +17,7 @@ with warnings.catch_warnings(): from esphome.const import CONF_ENCRYPTION, CONF_KEY, CONF_PORT, __version__ from esphome.core import CORE +from esphome.stacktrace import LogLineProcessor from esphome.util import safe_print if TYPE_CHECKING: @@ -29,50 +29,6 @@ if TYPE_CHECKING: _LOGGER = logging.getLogger(__name__) -class _LogLineProcessor: - """Feeds incoming log lines to the stack-trace decoder. - - Two responsibilities beyond just calling the decoder: - 1. Catch everything the decoder can raise. aioesphomeapi isolates - exceptions raised by log handlers, so an escaping one no longer - kills the session, but it does log a full traceback per line. A - crash dump carries a PC line plus one per backtrace frame, so the - tracebacks bury the dump the user is trying to read. Decoding is a - diagnostic nicety; nothing it raises is worth that noise. - 2. Disable decoding after the first failure. _decode_pc shells out to - the toolchain to resolve addr2line, which is expensive; a single - crash dump can contain many PC/BT lines and we don't want to retry - the failing subprocess for each one. This only works if every - failure is caught, which is why 1 is not narrowed to EsphomeError. - """ - - def __init__(self, config: dict[str, Any], platform_handler: Any | None) -> None: - self._config = config - self._platform_handler = platform_handler - self._decode_enabled = platform_handler is not None - self.backtrace_state = False - - def process_line(self, raw_line: str) -> None: - if not self._decode_enabled: - return - try: - self.backtrace_state = self._platform_handler( - self._config, raw_line, self.backtrace_state - ) - except Exception as exc: # noqa: BLE001 # pylint: disable=broad-except - self._decode_enabled = False - self.backtrace_state = False - # _run_idedata raises EsphomeError with no message; fall back - # to a generic explanation when str(exc) is empty. - detail = str(exc) or "build artifacts not found locally" - _LOGGER.debug("Stack-trace decoding failed", exc_info=True) - _LOGGER.warning( - "Crash trace decoding unavailable: %s. " - "Run 'esphome compile' for this device to enable PC decoding.", - detail, - ) - - async def async_run_logs( config: dict[str, Any], addresses: list[str], @@ -100,21 +56,9 @@ async def async_run_logs( provide_time=False, ) - # Try platform-specific stacktrace handler first, fall back to generic - platform_process_stacktrace = None - try: - module = importlib.import_module("esphome.components." + CORE.target_platform) - platform_process_stacktrace = module.process_stacktrace - except (AttributeError, ImportError): - # Distinguish "platform has no analyzer" from a genuinely broken - # platform package when debugging. - _LOGGER.debug("Stacktrace analyzer lookup failed", exc_info=True) - _LOGGER.info( - 'Stacktrace analysis is unavailable: no compatible analyzer found for target platform "%s".', - CORE.target_platform, - ) - - processor = _LogLineProcessor(config, platform_process_stacktrace) + # Decoder resolution, crash isolation, and disable-after-failure + # all live in LogLineProcessor, shared with the serial log path. + processor = LogLineProcessor(config, CORE.target_platform) def on_log(msg: SubscribeLogsResponse) -> None: """Handle a new log message.""" diff --git a/esphome/platform_hooks.py b/esphome/platform_hooks.py index 184644f29c..893500290b 100644 --- a/esphome/platform_hooks.py +++ b/esphome/platform_hooks.py @@ -10,10 +10,9 @@ imports each platform package and fails when they drift. The compile-path ``run_compile`` hook is deliberately not registered: compiling imports the platform package regardless, so its probe in -``__main__.py`` stays eager. The serial log path resolves -``process_stacktrace`` through get_stacktrace_handler below; the network -log client's probe in ``esphome/api_client.py`` still uses the old -importlib pattern and is converted separately. +``__main__.py`` stays eager. Both log paths resolve +``process_stacktrace`` through ``esphome.stacktrace.LogLineProcessor``, +which uses get_stacktrace_handler below. """ from __future__ import annotations diff --git a/esphome/platformio/toolchain.py b/esphome/platformio/toolchain.py index 0959cbfffb..32e30290ac 100644 --- a/esphome/platformio/toolchain.py +++ b/esphome/platformio/toolchain.py @@ -6,7 +6,7 @@ from pathlib import Path import re import shutil import sys -from typing import TYPE_CHECKING +from typing import TYPE_CHECKING, Any import platformdirs @@ -364,18 +364,24 @@ def run_compile(config, verbose): def _run_idedata(config): args = ["-t", "idedata"] stdout = run_platformio_cli_run(config, False, *args, capture_stdout=True) + if not isinstance(stdout, str): + # run_external_process returns 1 instead of captured output when + # launching platformio raised; see the error it logged above. + raise EsphomeError("Could not launch platformio to get idedata") match = re.search(r'{\s*".*}', stdout) if match is None: - _LOGGER.error("Could not match idedata, please report this error") + # A run that launches but fails emits its build error instead of + # idedata; the logged stdout is the useful part, not a bug report. + _LOGGER.error("Could not find idedata in the platformio output") _LOGGER.error("Stdout: %s", stdout) - raise EsphomeError + raise EsphomeError("PlatformIO did not report idedata") try: return json.loads(match.group()) - except ValueError: + except ValueError as err: _LOGGER.exception("Could not parse idedata") _LOGGER.error("Stdout: %s", stdout) - raise + raise EsphomeError("Could not parse idedata from platformio") from err def _load_idedata(config): @@ -419,9 +425,27 @@ class IDEData: def __init__(self, raw): self.raw = raw + def _require(self, *keys: str) -> Any: + """Read a nested key, classifying a miss as an environment error. + + A stale or truncated cached idedata JSON is the user's build + tree, not a bug; recompiling regenerates it. The message names + the key so a platformio schema change stays diagnosable. + """ + value = self.raw + # TypeError covers a key that is null instead of absent. + try: + for key in keys: + value = value[key] + except (KeyError, TypeError) as err: + raise EsphomeError( + f"Cached idedata is incomplete (missing {'.'.join(keys)})" + ) from err + return value + @property def firmware_elf_path(self) -> Path: - return Path(self.raw["prog_path"]) + return Path(self._require("prog_path")) @property def firmware_bin_path(self) -> Path: @@ -429,15 +453,22 @@ class IDEData: @property def extra_flash_images(self) -> list[FlashImage]: - return [ - FlashImage(path=Path(entry["path"]), offset=entry["offset"]) - for entry in self.raw["extra"]["flash_images"] - ] + try: + return [ + FlashImage(path=Path(entry["path"]), offset=entry["offset"]) + for entry in self._require("extra", "flash_images") + ] + except (KeyError, TypeError) as err: + # Covers entries missing path/offset and a null or non-list + # flash_images value alike. + raise EsphomeError( + "Cached idedata is incomplete (malformed extra.flash_images)" + ) from err @property def cc_path(self) -> str: # For example /Users//.platformio/packages/toolchain-xtensa32/bin/xtensa-esp32-elf-gcc - return self.raw["cc_path"] + return self._require("cc_path") @property def addr2line_path(self) -> str: diff --git a/esphome/stacktrace.py b/esphome/stacktrace.py new file mode 100644 index 0000000000..3fe3ef3cfe --- /dev/null +++ b/esphome/stacktrace.py @@ -0,0 +1,105 @@ +"""Stack-trace decoding for streamed device log lines. + +Shared by the serial (run_miniterm) and network (api_client) log paths. +Deliberately light: importing this module must not pull in aioesphomeapi +or any platform package. +""" + +from __future__ import annotations + +import logging +from typing import TYPE_CHECKING + +from esphome import platform_hooks +from esphome.core import EsphomeError +from esphome.types import ConfigType + +if TYPE_CHECKING: + from collections.abc import Callable + + # The contract every platform's process_stacktrace implements. + StacktraceHandler = Callable[[ConfigType, str, bool], bool] + +_LOGGER = logging.getLogger(__name__) + + +class LogLineProcessor: + """Feeds incoming log lines to the stack-trace decoder. + + Two responsibilities beyond just calling the decoder: + 1. Catch everything the decoder can raise. aioesphomeapi isolates + exceptions raised by log handlers, so an escaping one no longer + kills the session, but it does log a full traceback per line. A + crash dump carries a PC line plus one per backtrace frame, so the + tracebacks bury the dump the user is trying to read. Decoding is a + diagnostic nicety; nothing it raises is worth that noise. + 2. Disable decoding for the rest of the session after a failure. + _decode_pc shells out to the toolchain to resolve addr2line, + which is expensive; a single crash dump can contain many PC/BT + lines and we don't want to retry the failing subprocess for each + one. This only works if every failure is caught, which is why 1 + is not narrowed to EsphomeError. The latch is deliberately one + way: nothing a decode failure depends on heals by itself within + a session, the warning names the fix, and a fresh ``esphome + logs`` run picks it up; retrying mid-session would block the + stream with a failing subprocess instead. + """ + + def __init__(self, config: ConfigType, platform: str) -> None: + self._config = config + self._platform = platform + self._platform_handler: StacktraceHandler | None + try: + self._platform_handler = platform_hooks.get_stacktrace_handler(platform) + except Exception as exc: # noqa: BLE001 # pylint: disable=broad-except + # Total containment includes resolution: a platform package + # broken in an unanticipated way must not kill the session. + # Name the cause; the full traceback only exists at debug. + _LOGGER.debug("Stacktrace analyzer resolution failed", exc_info=True) + _LOGGER.warning( + 'Stacktrace analysis is unavailable: analyzer for target platform "%s" could not be loaded: %s', + platform, + f"{type(exc).__name__}: {exc}", + ) + self._platform_handler = None + self._decode_enabled = self._platform_handler is not None + self.backtrace_state = False + + def process_line(self, raw_line: str) -> None: + if not self._decode_enabled: + return + self._feed(raw_line) + + def _feed(self, raw_line: str) -> None: + try: + self.backtrace_state = self._platform_handler( + self._config, raw_line, self.backtrace_state + ) + except Exception as exc: # noqa: BLE001 # pylint: disable=broad-except + self._decode_enabled = False + self.backtrace_state = False + _LOGGER.debug("Stack-trace decoding failed", exc_info=True) + if isinstance(exc, (EsphomeError, OSError)): + # The environment branch: idedata and build tree failures + # get the remediation hint. The fallback string is + # defensive; the in-tree raise sites all carry a message + # now, but a bare EsphomeError must not render as parens. + _LOGGER.warning( + "Crash trace decoding unavailable: %s. " + "Run 'esphome compile' for this device to enable PC decoding.", + str(exc) or "build artifacts not found locally", + ) + else: + # A decoder bug is ESPHome's problem, not the user's; + # don't send them to recompile a healthy build. Always + # name the type: a bare KeyError message reads like a + # raised string in the paste a bug report needs. + detail = type(exc).__name__ + if msg := str(exc): + detail = f"{detail}: {msg}" + _LOGGER.warning( + 'Crash trace decoding disabled: decoder for "%s" raised %s ' + "(this is a bug; run with -v for the traceback)", + self._platform, + detail, + ) diff --git a/tests/unit_tests/analyze_memory/test_build_artifacts.py b/tests/unit_tests/analyze_memory/test_build_artifacts.py index ad2f8c2210..d97ee94e1c 100644 --- a/tests/unit_tests/analyze_memory/test_build_artifacts.py +++ b/tests/unit_tests/analyze_memory/test_build_artifacts.py @@ -143,8 +143,8 @@ def test_cc_path_from_cxx(cxx_path: str, expected: str) -> None: def test_native_idedata_resolves_toolchain_tools() -> None: """The binutils paths are derived from the native ESP-IDF cc_path. - Without cc_path, IDEData.objdump_path raises KeyError and the memory - analysis silently degrades to no component or symbol detail. + Without cc_path, IDEData.objdump_path raises EsphomeError and the + memory analysis silently degrades to no component or symbol detail. """ idedata = IDEData( { diff --git a/tests/unit_tests/test_api_client.py b/tests/unit_tests/test_api_client.py index 55907f282c..670557c16f 100644 --- a/tests/unit_tests/test_api_client.py +++ b/tests/unit_tests/test_api_client.py @@ -8,7 +8,6 @@ from unittest.mock import AsyncMock, Mock, patch import pytest from esphome import api_client -from esphome.components import esp32 from esphome.const import ( CONF_ENCRYPTION, CONF_KEY, @@ -16,7 +15,7 @@ from esphome.const import ( KEY_CORE, KEY_TARGET_PLATFORM, ) -from esphome.core import CORE, EsphomeError +from esphome.core import CORE def test_component_shim_reexports_runtime_client() -> None: @@ -29,135 +28,6 @@ def test_component_shim_reexports_runtime_client() -> None: assert api.CONF_ENCRYPTION is CONF_ENCRYPTION -def test_decoder_swallows_esphome_error() -> None: - """A failing stack-trace decode must not propagate. - - aioesphomeapi isolates exceptions raised by log handlers, so an - escaping one logs a full traceback for every line it fires on rather - than being reported once as an unavailable decoder. - """ - config = {"esphome": {"name": "test"}} - - with patch.object( - esp32, "process_stacktrace", side_effect=EsphomeError("no idedata") - ) as mock_process: - processor = api_client._LogLineProcessor(config, esp32.process_stacktrace) - processor.process_line("PC: 0x4010496e") - - assert mock_process.called - assert processor.backtrace_state is False - - -def test_decoder_swallows_platform_handler_error() -> None: - """The same protection must apply to the platform-specific handler.""" - config = {"esphome": {"name": "test"}} - - def platform_handler(_config, _line, _state): - raise EsphomeError("no idedata") - - processor = api_client._LogLineProcessor(config, platform_handler) - processor.process_line("PC: 0x4010496e") - - assert processor.backtrace_state is False - - -def test_decoder_swallows_non_esphome_error() -> None: - """Decoding failures that aren't EsphomeError must be contained too. - - A missing build directory surfaces as FileNotFoundError from the toolchain - subprocess. aioesphomeapi isolates it, so the session survives, but it logs - a traceback for every PC/BT line and decoding is never disabled, which - buries the crash dump the user is trying to read. - """ - config = {"esphome": {"name": "test"}} - - with patch.object( - esp32, - "process_stacktrace", - side_effect=FileNotFoundError( - 2, "No such file or directory", "/build/ol/build" - ), - ) as mock_process: - processor = api_client._LogLineProcessor(config, esp32.process_stacktrace) - processor.process_line("PC: 0x4010496e") - processor.process_line("BT0: 0x4010496e") - - # Disabled after the first failure rather than retried per backtrace line. - assert mock_process.call_count == 1 - assert processor.backtrace_state is False - - -def test_decoder_warning_uses_fallback_for_empty_error(caplog) -> None: - """_run_idedata raises EsphomeError with no message; the warning - must show a useful explanation rather than empty parens. - """ - config = {"esphome": {"name": "test"}} - - with patch.object(esp32, "process_stacktrace", side_effect=EsphomeError()): - processor = api_client._LogLineProcessor(config, esp32.process_stacktrace) - processor.process_line("PC: 0x4010496e") - - warnings = [r.message for r in caplog.records if r.levelname == "WARNING"] - assert any("build artifacts not found locally" in m for m in warnings) - assert not any("()" in m for m in warnings) - - -def test_decoder_short_circuits_after_failure() -> None: - """After one failure, subsequent lines must not retry the decoder. - - _decode_pc shells out to the toolchain; a crash dump can contain many - PC/BT lines and retrying the failing subprocess for each one would - stall log streaming. - """ - config = {"esphome": {"name": "test"}} - - with patch.object( - esp32, "process_stacktrace", side_effect=EsphomeError("no idedata") - ) as mock_process: - processor = api_client._LogLineProcessor(config, esp32.process_stacktrace) - processor.process_line("PC: 0x4010496e") - processor.process_line("BT0: 0x4010496e") - processor.process_line("BT1: 0x401049aa") - - assert mock_process.call_count == 1 - - -def test_decoder_threads_backtrace_state() -> None: - """When decoding succeeds, backtrace_state is threaded across calls.""" - config = {"esphome": {"name": "test"}} - - with patch.object( - esp32, "process_stacktrace", side_effect=[True, False] - ) as mock_process: - processor = api_client._LogLineProcessor(config, esp32.process_stacktrace) - processor.process_line(">>>stack>>>") - assert processor.backtrace_state is True - processor.process_line("<< None: - """The platform handler is preferred over the generic one.""" - config = {"esphome": {"name": "test"}} - calls: list[tuple[object, str, bool]] = [] - - def platform_handler(cfg, line, state): - calls.append((cfg, line, state)) - return True - - processor = api_client._LogLineProcessor(config, platform_handler) - - with patch.object(esp32, "process_stacktrace") as mock_generic: - processor.process_line("BT0: 0x4010496e") - - assert calls == [(config, "BT0: 0x4010496e", False)] - assert mock_generic.called is False - assert processor.backtrace_state is True - - @pytest.mark.asyncio @pytest.mark.parametrize( ("extra_config", "expected_deep_sleep"), @@ -194,6 +64,7 @@ async def test_async_run_logs_full_flow(caplog) -> None: stop() cleanup in the finally block. """ caplog.set_level("INFO", logger="esphome.api_client") + caplog.set_level("INFO", logger="esphome.platform_hooks") CORE.data[KEY_CORE] = {KEY_TARGET_PLATFORM: "host"} config = { "esphome": {"name": "test"}, @@ -233,7 +104,7 @@ async def test_async_run_logs_full_flow(caplog) -> None: assert mock_client.call_args.kwargs["noise_psk"] == "psk123" assert mock_client.call_args.kwargs["addresses"] == ["1.2.3.4", "5.6.7.8"] assert "1.2.3.4 or 5.6.7.8" in caplog.text - # host has no stacktrace analyzer; the fallback message is logged. + # host has no stacktrace analyzer; the notice fires at session start. assert "Stacktrace analysis is unavailable" in caplog.text # The log message was printed with a timestamp prefix. assert any("hello world" in line for line in printed) diff --git a/tests/unit_tests/test_lazy_imports.py b/tests/unit_tests/test_lazy_imports.py index 69202b1a72..74b2362949 100644 --- a/tests/unit_tests/test_lazy_imports.py +++ b/tests/unit_tests/test_lazy_imports.py @@ -37,16 +37,21 @@ HEAVY_MODULES = ( # existence guard and the leak check must watch the same list. FAST_PATH_HEAVY_MODULES = HEAVY_MODULES + ("esphome.components.esp32",) +# Heavy only for modules that must not know about the API transport; +# in the existence guard so a rename can't silently no-op its check. +API_HEAVY_MODULES = ("aioesphomeapi",) -def _leaked_heavy_modules(module: str) -> str: + +def _leaked_heavy_modules(module: str, extra: tuple[str, ...] = ()) -> str: """Import ``module`` in a subprocess and report the heavy modules it pulled. Any ``esphome.components.*`` package counts as heavy: executing a component package drags in codegen/validation machinery by design. + ``extra`` adds modules that are heavy for this caller specifically. """ check = ( f"import sys; import {module}; " - f"leaked = [m for m in {HEAVY_MODULES!r} if m in sys.modules]; " + f"leaked = [m for m in {HEAVY_MODULES + extra!r} if m in sys.modules]; " "leaked += [m for m in sys.modules if m.startswith('esphome.components.')]; " "print(','.join(leaked))" ) @@ -72,7 +77,7 @@ def test_main_module_does_not_import_heavy_modules() -> None: def test_watched_heavy_modules_exist() -> None: """A renamed heavy module would silently disable the leak checks.""" - for module in FAST_PATH_HEAVY_MODULES: + for module in FAST_PATH_HEAVY_MODULES + API_HEAVY_MODULES: assert importlib.util.find_spec(module) is not None, ( f"{module} no longer resolves; update the heavy-module lists" ) @@ -145,6 +150,21 @@ def test_api_client_does_not_import_heavy_modules() -> None: ) +def test_stacktrace_does_not_import_heavy_modules() -> None: + """``esphome.stacktrace`` guards its own docstring's contract. + + Both log paths construct a LogLineProcessor before streaming + starts; importing the module must not pull in aioesphomeapi or + any platform package. + """ + leaked = _leaked_heavy_modules("esphome.stacktrace", extra=API_HEAVY_MODULES) + assert not leaked, ( + f"esphome.stacktrace imports heavy modules at top level: {leaked}. " + "The logs fast path skips validation; importing the validation " + "stack anyway defeats the validated-config cache." + ) + + def test_espidf_toolchain_does_not_import_heavy_modules() -> None: """The esp-idf upload path must not pull the esp32 package back in. diff --git a/tests/unit_tests/test_main.py b/tests/unit_tests/test_main.py index 09c8d25249..16ab677ba6 100644 --- a/tests/unit_tests/test_main.py +++ b/tests/unit_tests/test_main.py @@ -5923,6 +5923,38 @@ def test_run_miniterm_backtrace_state_maintained() -> None: assert backtrace_states[3][1] is True +def test_run_miniterm_decoder_failure_keeps_streaming( + caplog: pytest.LogCaptureFixture, +) -> None: + """A decoder exception must not kill serial streaming. + + This is the serial path's gain from sharing LogLineProcessor: before + the lift a decoder exception propagated out of the read loop. + """ + chunk = b"PC: 0x4010496e\r\nBT0: 0x4010496e\r\nstill streaming\r\n" + mock_serial = MockSerial([chunk, MOCK_SERIAL_END]) + + CORE.data[KEY_CORE] = {KEY_TARGET_PLATFORM: PLATFORM_ESP32} + config = { + CONF_LOGGER: { + CONF_BAUD_RATE: 115200, + "deassert_rts_dtr": False, + } + } + args = MockArgs() + + decoder = Mock(side_effect=EsphomeError("no idedata")) + with ( + patch("serial.Serial", return_value=mock_serial), + patch.object(esp32, "process_stacktrace", decoder), + ): + run_miniterm(config, "/dev/ttyUSB0", args) + + # The failure is contained and latched; streaming continued to EOF. + assert decoder.call_count == 1 + assert "Crash trace decoding unavailable" in caplog.text + + def test_run_miniterm_handles_empty_reads( capfd: CaptureFixture[str], ) -> None: diff --git a/tests/unit_tests/test_platformio_toolchain.py b/tests/unit_tests/test_platformio_toolchain.py index 723646fbd6..9450e8e0e1 100644 --- a/tests/unit_tests/test_platformio_toolchain.py +++ b/tests/unit_tests/test_platformio_toolchain.py @@ -278,15 +278,56 @@ def test_run_idedata_raises_on_no_json( def test_run_idedata_raises_on_invalid_json( setup_core: Path, mock_run_platformio_cli_run: Mock ) -> None: - """Test _run_idedata raises on malformed JSON.""" + """Malformed JSON is the environment (garbage stdout), so it must + surface as EsphomeError and get the recompile hint downstream. + """ config = {"name": "test"} mock_run_platformio_cli_run.return_value = '{"invalid": json"}' - # The ValueError from json.loads is re-raised - with pytest.raises(ValueError): + with pytest.raises(EsphomeError): toolchain._run_idedata(config) +def test_run_idedata_raises_on_launch_failure( + setup_core: Path, mock_run_platformio_cli_run: Mock +) -> None: + """A failed platformio launch returns its exit code as an int; that + must surface as EsphomeError, not a TypeError from re.search. + """ + config = {"name": "test"} + mock_run_platformio_cli_run.return_value = 1 + + with pytest.raises(EsphomeError): + toolchain._run_idedata(config) + + +def test_idedata_missing_prog_path_raises_esphome_error(setup_core: Path) -> None: + """A stale cached idedata JSON without prog_path is the build tree's + fault; it must surface as EsphomeError, not a KeyError. + """ + with pytest.raises(EsphomeError): + _ = toolchain.IDEData({}).firmware_elf_path + + +def test_idedata_missing_flash_image_field_raises_esphome_error( + setup_core: Path, +) -> None: + """A cached idedata whose flash image entries lost a field must + classify as an environment error too, not a raw KeyError. + """ + idedata = toolchain.IDEData({"extra": {"flash_images": [{"offset": "0x1000"}]}}) + with pytest.raises(EsphomeError): + _ = idedata.extra_flash_images + + +def test_idedata_null_section_raises_esphome_error(setup_core: Path) -> None: + """A section that is null instead of absent must classify the same + as a missing key instead of escaping as TypeError. + """ + with pytest.raises(EsphomeError): + _ = toolchain.IDEData({"extra": None}).extra_flash_images + + def test_run_platformio_cli_sets_environment_variables( setup_core: Path, mock_run_external_process: Mock ) -> None: diff --git a/tests/unit_tests/test_stacktrace.py b/tests/unit_tests/test_stacktrace.py new file mode 100644 index 0000000000..fcc99ab587 --- /dev/null +++ b/tests/unit_tests/test_stacktrace.py @@ -0,0 +1,151 @@ +"""Tests for esphome.stacktrace.""" + +from __future__ import annotations + +from unittest.mock import Mock, patch + +from esphome import stacktrace +from esphome.const import PLATFORM_BK72XX, PLATFORM_ESP32, PLATFORM_ESP8266 +from esphome.core import EsphomeError + +CONFIG = {"esphome": {"name": "test"}} + + +def _run( + handler, + platform: str = PLATFORM_ESP32, + lines: tuple[str, ...] = ("PC: 0x4010496e",), +) -> stacktrace.LogLineProcessor: + """Processor with the resolver stubbed, fed the given lines.""" + with patch.object( + stacktrace.platform_hooks, "get_stacktrace_handler", return_value=handler + ): + processor = stacktrace.LogLineProcessor(CONFIG, platform) + for line in lines: + processor.process_line(line) + return processor + + +def _fed(handler) -> list[str]: + return [call.args[1] for call in handler.call_args_list] + + +def _warnings(caplog) -> list[str]: + return [r.message for r in caplog.records if r.levelname == "WARNING"] + + +def test_decoder_contains_failures_and_short_circuits() -> None: + """One decode failure is contained and never retried. + + aioesphomeapi isolates exceptions raised by log handlers, so an + escaping one logs a full traceback for every line it fires on; and + _decode_pc shells out to the toolchain, so retrying it per backtrace + line would stall streaming. + """ + handler = Mock(side_effect=EsphomeError("no idedata")) + processor = _run( + handler, lines=("PC: 0x4010496e", "BT0: 0x4010496e", "BT1: 0x401049aa") + ) + + assert handler.call_count == 1 + assert processor.backtrace_state is False + + +def test_resolution_failure_is_contained(caplog) -> None: + """A platform package broken in an unanticipated way must not kill + the session; decoding degrades with a warning like any other failure. + """ + with patch.object( + stacktrace.platform_hooks, + "get_stacktrace_handler", + side_effect=RuntimeError("boom"), + ): + processor = stacktrace.LogLineProcessor(CONFIG, PLATFORM_ESP32) + processor.process_line("PC: 0x4010496e") + + assert processor.backtrace_state is False + assert any("could not be loaded" in m for m in _warnings(caplog)) + + +def test_decoder_swallows_os_error_with_remediation_hint(caplog) -> None: + """Decoding failures that aren't EsphomeError must be contained too. + + A missing build directory surfaces as an OSError; that is the + user's environment, not a decoder bug, so it disables decoding + like an EsphomeError does and keeps the recompile hint. + """ + handler = Mock( + side_effect=FileNotFoundError(2, "No such file or directory", "/build") + ) + processor = _run(handler, lines=("PC: 0x4010496e", "BT0: 0x4010496e")) + + assert handler.call_count == 1 + assert processor.backtrace_state is False + warnings = _warnings(caplog) + assert any("esphome compile" in m for m in warnings) + assert not any("this is a bug" in m for m in warnings) + + +def test_decoder_warning_uses_fallback_for_empty_error(caplog) -> None: + """A message-less EsphomeError must show a useful explanation. + + Defensive: the in-tree idedata raise sites all carry a message now, + but a bare EsphomeError from elsewhere must not render as parens. + """ + _run(Mock(side_effect=EsphomeError())) + + warnings = _warnings(caplog) + assert any("build artifacts not found locally" in m for m in warnings) + assert not any("()" in m for m in warnings) + + +def test_decoder_bug_with_empty_message_names_the_type(caplog) -> None: + """A zero-message decoder bug must not masquerade as missing artifacts. + + The recompile hint is only right for EsphomeError from _run_idedata; + anything else is ESPHome's own bug and says so instead of sending + the user down a dead-end remediation path. + """ + _run(Mock(side_effect=IndexError())) + + warnings = _warnings(caplog) + assert any("IndexError" in m and "this is a bug" in m for m in warnings) + assert not any("esphome compile" in m for m in warnings) + + +def test_decoder_bug_warning_keeps_the_type_with_a_message(caplog) -> None: + """The type must survive a non-empty message; a bare KeyError message + like 'prog_path' reads as a raised string in a bug report paste. + """ + _run(Mock(side_effect=KeyError("prog_path"))) + + warnings = _warnings(caplog) + assert any("KeyError: 'prog_path'" in m for m in warnings) + + +def test_state_threads_between_lines() -> None: + """backtrace_state carries from one decoded line to the next.""" + handler = Mock(side_effect=[True, True]) + processor = _run( + handler, + platform=PLATFORM_ESP8266, + lines=(">>>stack>>>", "3ffffe10: 40201234 3ffe8410 00000000 40201000"), + ) + + assert _fed(handler) == [ + ">>>stack>>>", + "3ffffe10: 40201234 3ffe8410 00000000 40201000", + ] + assert handler.call_args_list[0].args[2] is False + assert handler.call_args_list[1].args[2] is True + assert processor.backtrace_state is True + + +def test_no_analyzer_disables_decoding(caplog) -> None: + """Platforms without an analyzer report at session start and stay quiet.""" + caplog.set_level("INFO", logger="esphome.platform_hooks") + processor = stacktrace.LogLineProcessor(CONFIG, PLATFORM_BK72XX) + processor.process_line("PC: 0x40104960") + + assert "Stacktrace analysis is unavailable" in caplog.text + assert processor.backtrace_state is False