From 253ecdd145880aa5e521592a899d0f2d46d9e1d8 Mon Sep 17 00:00:00 2001 From: "J. Nick Koston" Date: Wed, 5 Aug 2026 14:34:16 -0500 Subject: [PATCH] [core] Resolve the stacktrace decoder lazily behind an address gate (#18048) --- esphome/api_client.py | 3 +- esphome/components/nrf52/__init__.py | 16 +- esphome/platform_hooks.py | 62 +++- esphome/stacktrace.py | 80 +++- tests/unit_tests/test_api_client.py | 97 +++-- tests/unit_tests/test_main.py | 16 +- tests/unit_tests/test_stacktrace.py | 534 +++++++++++++++++++++++++-- 7 files changed, 716 insertions(+), 92 deletions(-) diff --git a/esphome/api_client.py b/esphome/api_client.py index cd80a64fb9..b9a71a3ff7 100644 --- a/esphome/api_client.py +++ b/esphome/api_client.py @@ -56,8 +56,7 @@ async def async_run_logs( provide_time=False, ) - # Decoder resolution, crash isolation, and disable-after-failure - # all live in LogLineProcessor, shared with the serial log path. + # Decoder resolution policy lives in LogLineProcessor. processor = LogLineProcessor(config, CORE.target_platform) def on_log(msg: SubscribeLogsResponse) -> None: diff --git a/esphome/components/nrf52/__init__.py b/esphome/components/nrf52/__init__.py index 4002d1cc04..27d7b5fd35 100644 --- a/esphome/components/nrf52/__init__.py +++ b/esphome/components/nrf52/__init__.py @@ -722,11 +722,25 @@ def _addr2line(addr2line: str, elf: Path, addr: str) -> str: return "" +# Module-level so tests can pin the samples that gate lazy decoding +# (tests/unit_tests/test_stacktrace.py) against the real pattern. +# The PC group is bounded to 3+ hex digits to agree with the log gate +# in platform_hooks.STACKTRACE_GATES by construction; the logger prints +# both registers with %08x, so a real PC is always 8 digits and even a +# vector-table address is zero-padded past the bound. The LR bound is +# left wide so a nonsensical LR still satisfies the combined match; the +# regex is all or nothing, so a failed LR half would drop the PC decode +# with it. Widening the PC bound without the gate fails the generative +# superset test in tests/unit_tests/test_stacktrace.py, so the two +# cannot drift. +STACKTRACE_NRF52_PC_LR_RE = re.compile(r"PC=(0x[0-9a-fA-F]{3,})\s+LR=(0x[0-9a-fA-F]+)") + + def process_stacktrace(config: ConfigType, line: str, backtrace_state: bool) -> bool: if "Last crash:" in line: return True if backtrace_state: - match = re.search(r"PC=(0x[0-9a-fA-F]+)\s+LR=(0x[0-9a-fA-F]+)", line) + match = STACKTRACE_NRF52_PC_LR_RE.search(line) if match: pc = match.group(1) lr = match.group(2) diff --git a/esphome/platform_hooks.py b/esphome/platform_hooks.py index 893500290b..b58e3f570c 100644 --- a/esphome/platform_hooks.py +++ b/esphome/platform_hooks.py @@ -37,12 +37,54 @@ _LOGGER = logging.getLogger(__name__) # (upload method, log transport) warns. A new hook is loud by default. COSMETIC_HOOKS: Final = frozenset({"process_stacktrace"}) +# Trigger gates for the stacktrace decoders, keyed by platform. A log +# session knows its target platform, so each session only pays for its +# own platform's trigger language; a line matching the gate is what +# lazily imports the platform package to resolve the decoder, and a +# false trigger costs that import on the event loop mid-stream. That is +# why 8-digit decimals (uptime counters) and ESP-IDF's decimal log +# timestamps must stay out of the esp8266 bare-hex branch. +# +# Declaring a gate here is what registers a platform's +# process_stacktrace hook; deriving the registry entry from the keys +# keeps the two in sync by construction. Each gate must stay a superset +# of its decoder patterns' trigger language; both directions are pinned +# in tests/unit_tests/test_stacktrace.py, including a generative test +# that derives inputs from the decoder regexes themselves. The keyword +# branches exist because (?:0x)? register forms can glue a pointer to +# trailing word characters that defeat the pointer branch's \b, and the +# markers are the address-free lines that open the state-gated +# decoders' dump regions. The esp32/esp8266/rp2 crash handlers all +# announce a stored dump with the CRASH DETECTED banner as its first +# line; their decoders key on the 0x-bearing lines that follow, but +# gating on the banner resolves the decoder at the dump's first line +# and can only be a true positive. +# +# Stored as strings: this module is imported by every CLI invocation, +# and only a log session needs a gate, so the session compiles exactly +# its own platform's entry instead of import time compiling all four. +STACKTRACE_GATES: Final[dict[str, str]] = { + PLATFORM_ESP32: ( + r"0x[0-9a-fA-F]{3,}\b" + r"|(?:PC|RA|MEPC|MTVAL|EXCVADDR|call)\s*[:=]\s*(?:0x)?4[0-9a-fA-F]{7}" + r"|CRASH DETECTED ON PREVIOUS BOOT" + ), + PLATFORM_ESP8266: ( + r"0x[0-9a-fA-F]{3,}\b" + r"|\b(?![0-9]{8}\b)[0-9a-fA-F]{8}\b" + r"|(?:PC|EXCVADDR|call)\s*[:=]\s*(?:0x)?4[0-9a-fA-F]{7}" + r"|[eE]xception \(\d+\):" + r"|>>>stack>>>" + r"|CRASH DETECTED ON PREVIOUS BOOT" + ), + PLATFORM_RP2: r"0x[0-9a-fA-F]{3,}\b|CRASH DETECTED ON PREVIOUS BOOT", + PLATFORM_NRF52: r"0x[0-9a-fA-F]{3,}\b|Last crash:", +} + PLATFORM_HOOKS: Final[dict[str, frozenset[str]]] = { "show_logs": frozenset({PLATFORM_NRF52}), "upload_program": frozenset({PLATFORM_NRF52}), - "process_stacktrace": frozenset( - {PLATFORM_ESP32, PLATFORM_ESP8266, PLATFORM_NRF52, PLATFORM_RP2} - ), + "process_stacktrace": frozenset(STACKTRACE_GATES), } @@ -55,6 +97,18 @@ PLATFORM_HOOKS: Final[dict[str, frozenset[str]]] = { _IN_TREE_PLATFORMS: Final = frozenset(Platform) +def has_registered_hook(platform: str, hook: str) -> bool: + """True when *platform* declares *hook* in ``PLATFORM_HOOKS``. + + Callers that defer imports key off this: a registered hook is known + to exist, so ``get_platform_hook`` can wait until it is needed; + anything else must be probed up front so availability is reported + at session start. Keeping the predicate here keeps the resolution + rule in one module. + """ + return platform in PLATFORM_HOOKS[hook] + + def get_platform_hook(platform: str, hook: str) -> Callable[..., Any] | None: """Return ``esphome.components..`` or None. @@ -63,7 +117,7 @@ def get_platform_hook(platform: str, hook: str) -> Callable[..., Any] | None: hook also returns None, so a stale registry degrades to the generic path instead of raising. """ - registered = platform in PLATFORM_HOOKS[hook] + registered = has_registered_hook(platform, hook) if not registered and platform in _IN_TREE_PLATFORMS: return None # For external platforms this probes the imported package like the diff --git a/esphome/stacktrace.py b/esphome/stacktrace.py index 3fe3ef3cfe..93b1a73ea9 100644 --- a/esphome/stacktrace.py +++ b/esphome/stacktrace.py @@ -1,4 +1,4 @@ -"""Stack-trace decoding for streamed device log lines. +"""Lazy 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 @@ -8,6 +8,7 @@ or any platform package. from __future__ import annotations import logging +import re from typing import TYPE_CHECKING from esphome import platform_hooks @@ -26,18 +27,28 @@ _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 + Three responsibilities beyond just calling the decoder: + 1. Resolve the platform decoder through the registry: lazily for + in-tree platforms with a registered decoder, where nothing is + imported until a line matches the platform's own gate in + platform_hooks.STACKTRACE_GATES, and eagerly otherwise. A + registry-proven miss reports its unavailable notice at session + start without importing anything; an external platform resolves + up front because the gates' grammar derives from the in-tree + decoders and its import cannot be avoided anyway - resolving + early keeps it out of the streaming callback, where a blocking + import would stall delivery mid-stream. + 2. 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. + 3. 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 + one. This only works if every failure is caught, which is why 2 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 @@ -48,28 +59,57 @@ class LogLineProcessor: 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._platform_handler: StacktraceHandler | None = None + self._decode_enabled = True + # None only for platforms resolved eagerly below, which never + # consult the gate: a registered platform always declares one. + # Compiled here rather than in the registry so only a log + # session pays for its own platform's gate. + gate = platform_hooks.STACKTRACE_GATES.get(platform) + self._gate: re.Pattern[str] | None = None if gate is None else re.compile(gate) self.backtrace_state = False + if not platform_hooks.has_registered_hook(platform, "process_stacktrace"): + self._resolve_handler() def process_line(self, raw_line: str) -> None: if not self._decode_enabled: return + if self._platform_handler is None: + if not self._gate.search(raw_line): + return + # Deliberate trade: the platform import (~300 ms, seconds on + # small hosts) blocks the streaming callback here, once per + # session, instead of every session paying it at startup. + if not self._resolve_handler(): + return + # The only runtime breadcrumb for the gate: with -v this + # distinguishes "gate never fired" from "no crash occurred". + _LOGGER.debug( + "Stacktrace gate fired for %s; decoder resolved", self._platform + ) self._feed(raw_line) + def _resolve_handler(self) -> bool: + try: + handler = platform_hooks.get_stacktrace_handler(self._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 or + # retry on every address-bearing line. Name the cause like + # _feed does; 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', + self._platform, + f"{type(exc).__name__}: {exc}", + ) + handler = None + if handler is None: + self._decode_enabled = False + return False + self._platform_handler = handler + return True + def _feed(self, raw_line: str) -> None: try: self.backtrace_state = self._platform_handler( diff --git a/tests/unit_tests/test_api_client.py b/tests/unit_tests/test_api_client.py index 670557c16f..19ed83abe1 100644 --- a/tests/unit_tests/test_api_client.py +++ b/tests/unit_tests/test_api_client.py @@ -28,40 +28,13 @@ def test_component_shim_reexports_runtime_client() -> None: assert api.CONF_ENCRYPTION is CONF_ENCRYPTION -@pytest.mark.asyncio -@pytest.mark.parametrize( - ("extra_config", "expected_deep_sleep"), - [({"deep_sleep": {}}, True), ({}, False)], -) -async def test_async_run_logs_passes_deep_sleep( - extra_config: dict, expected_deep_sleep: bool -) -> None: - """async_run_logs tells async_run whether the device deep sleeps, from the config.""" - CORE.data[KEY_CORE] = {KEY_TARGET_PLATFORM: "esp32"} - config = {"esphome": {"name": "test"}, "api": {CONF_PORT: 6053}, **extra_config} - # async_run blocks forever after connecting; raise to unwind async_run_logs - # once we have captured how it was called. - sentinel = RuntimeError("stop the wait") - - with ( - patch.object( - api_client, "async_run", AsyncMock(side_effect=sentinel) - ) as mock_run, - patch.object(api_client, "APIClient"), - pytest.raises(RuntimeError, match="stop the wait"), - ): - await api_client.async_run_logs(config, ["1.2.3.4"]) - - assert mock_run.call_args.kwargs["deep_sleep"] is expected_deep_sleep - - @pytest.mark.asyncio async def test_async_run_logs_full_flow(caplog) -> None: """Drive async_run_logs end to end with a fake connection. Covers the encryption key extraction, the multi-address banner, the - missing-stacktrace-analyzer fallback, the on_log handler, and the - stop() cleanup in the finally block. + registry-miss unavailable notice at session start, the on_log + handler, and the stop() cleanup in the finally block. """ caplog.set_level("INFO", logger="esphome.api_client") caplog.set_level("INFO", logger="esphome.platform_hooks") @@ -112,6 +85,41 @@ async def test_async_run_logs_full_flow(caplog) -> None: stop.assert_awaited_once() +@pytest.mark.asyncio +async def test_async_run_logs_never_resolves_without_crash_lines() -> None: + """The headline claim: an ordinary session imports no platform code.""" + CORE.data[KEY_CORE] = {KEY_TARGET_PLATFORM: "esp32"} + config = {"esphome": {"name": "test"}, "api": {CONF_PORT: 6053}} + + stop = AsyncMock() + run_started = asyncio.Event() + + async def fake_async_run(*args, **kwargs): + run_started.set() + return stop + + mock_run = AsyncMock(side_effect=fake_async_run) + + with ( + patch.object(api_client, "async_run", mock_run), + patch.object(api_client, "APIClient"), + patch.object(api_client, "safe_print"), + patch("esphome.platform_hooks.get_stacktrace_handler") as mock_resolve, + ): + task = asyncio.get_running_loop().create_task( + api_client.async_run_logs(config, ["1.2.3.4"]) + ) + async with asyncio.timeout(1): + await run_started.wait() + on_log = mock_run.call_args.args[1] + on_log(Mock(message=b"[I][app:100] hello\n[C][wifi:200] connected")) + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + + mock_resolve.assert_not_called() + + def test_run_logs_suppresses_keyboard_interrupt() -> None: """Ctrl-C during log streaming exits cleanly instead of tracebacking.""" with patch.object( @@ -124,3 +132,34 @@ def test_run_logs_suppresses_keyboard_interrupt() -> None: ) assert mock_run.call_args.kwargs["subscribe_states"] is False + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("extra_config", "expected_deep_sleep"), + [({"deep_sleep": {}}, True), ({}, False)], +) +async def test_async_run_logs_passes_deep_sleep( + extra_config: dict, expected_deep_sleep: bool +) -> None: + """async_run_logs tells async_run whether the device deep sleeps. + + That flag is the only thing capping reconnect backoff for a device + that is only briefly awake; dropping it means missed wake windows. + """ + CORE.data[KEY_CORE] = {KEY_TARGET_PLATFORM: "esp32"} + config = {"esphome": {"name": "test"}, "api": {CONF_PORT: 6053}, **extra_config} + # async_run blocks forever after connecting; raise to unwind + # async_run_logs once we have captured how it was called. + sentinel = RuntimeError("stop the wait") + + with ( + patch.object( + api_client, "async_run", AsyncMock(side_effect=sentinel) + ) as mock_run, + patch.object(api_client, "APIClient"), + pytest.raises(RuntimeError, match="stop the wait"), + ): + await api_client.async_run_logs(config, ["1.2.3.4"]) + + assert mock_run.call_args.kwargs["deep_sleep"] is expected_deep_sleep diff --git a/tests/unit_tests/test_main.py b/tests/unit_tests/test_main.py index 16ab677ba6..fc7bb9ada3 100644 --- a/tests/unit_tests/test_main.py +++ b/tests/unit_tests/test_main.py @@ -63,7 +63,7 @@ from esphome.__main__ import ( ) from esphome.address_cache import AddressCache from esphome.bundle import BUNDLE_EXTENSION, BundleFile, BundleResult -from esphome.components import esp32 +from esphome.components import esp32, esp8266 from esphome.components.esp32 import ( KEY_ESP32, KEY_VARIANT, @@ -5738,8 +5738,12 @@ def test_run_miniterm_batches_lines_with_same_timestamp( def test_run_miniterm_analyzer_import_failure_keeps_streaming( caplog: pytest.LogCaptureFixture, ) -> None: - """A broken platform import must not stop serial log streaming.""" - mock_serial = MockSerial([b"[I][app:100]: Line 1\r\n", MOCK_SERIAL_END]) + """A broken platform import must not stop serial log streaming. + + The decoder resolves lazily, so a crash-shaped line has to arrive + before the import is attempted at all. + """ + mock_serial = MockSerial([b"PC: 0x40104960\r\n", MOCK_SERIAL_END]) CORE.data[KEY_CORE] = {KEY_TARGET_PLATFORM: PLATFORM_ESP32} config = { @@ -5870,7 +5874,9 @@ def test_run_miniterm_backtrace_state_maintained() -> None: mock_serial = MockSerial([backtrace_chunk, MOCK_SERIAL_END]) - CORE.data[KEY_CORE] = {KEY_TARGET_PLATFORM: PLATFORM_ESP32} + # An esp8266 dump on an esp8266 session; the platform-scoped gate + # would rightly never resolve esp32's decoder for these lines. + CORE.data[KEY_CORE] = {KEY_TARGET_PLATFORM: PLATFORM_ESP8266} config = { CONF_LOGGER: { CONF_BAUD_RATE: 115200, @@ -5896,7 +5902,7 @@ def test_run_miniterm_backtrace_state_maintained() -> None: with ( patch("serial.Serial", return_value=mock_serial), patch.object( - esp32, + esp8266, "process_stacktrace", side_effect=track_backtrace_state, ), diff --git a/tests/unit_tests/test_stacktrace.py b/tests/unit_tests/test_stacktrace.py index fcc99ab587..ff317e55f8 100644 --- a/tests/unit_tests/test_stacktrace.py +++ b/tests/unit_tests/test_stacktrace.py @@ -2,14 +2,381 @@ from __future__ import annotations +import importlib +import inspect +from pathlib import Path +import re from unittest.mock import Mock, patch +from hypothesis import given, settings +from hypothesis.strategies import data as st_data, from_regex +import pytest + from esphome import stacktrace -from esphome.const import PLATFORM_BK72XX, PLATFORM_ESP32, PLATFORM_ESP8266 +from esphome.const import ( + PLATFORM_BK72XX, + PLATFORM_ESP32, + PLATFORM_ESP8266, + PLATFORM_NRF52, + PLATFORM_RP2, +) from esphome.core import EsphomeError CONFIG = {"esphome": {"name": "test"}} +# Real dump lines per registered platform. "addresses" are gate-firing +# dump lines (registers, backtraces, the exception header); the gate +# must fire on each or that platform's decoding silently never starts. +# "state_markers" are the address-free lines that set backtrace_state; +# the gate matches them directly so their decoders never miss the line +# that opens their dump region. "extra_triggers" are gate-firing lines +# no decoder pattern consumes, like the stored-dump banner that lets +# the decoder resolve at a dump's first line. A new decoder must +# declare its lines here so all are pinned instead of discovered in +# the field. +CRASH_SAMPLES: dict[str, dict[str, list[str]]] = { + PLATFORM_ESP32: { + "state_markers": [], + "extra_triggers": ["*** CRASH DETECTED ON PREVIOUS BOOT ***"], + "addresses": [ + "Backtrace: 0x400d1a2c:0x3ffb1f60 0x400d2a3c:0x3ffb1f80", + "PC : 0x400d1a2c PS : 0x00060330", + "EXCVADDR: 0x40001234", + "MEPC : 0x40380abc RA : 0x40380def", + "MTVAL : 0x40000123", + "last failed alloc call: 40201234(512)", + "BT0: 0x40104960", + ], + }, + PLATFORM_ESP8266: { + "state_markers": [">>>stack>>>"], + "extra_triggers": ["*** CRASH DETECTED ON PREVIOUS BOOT ***"], + "addresses": [ + "epc1=0x40201234 epc2=0x00000000 excvaddr=0x40001234", + "3ffffe10: 40201234 3ffe8410 00000000 40201000", + "PC : 40201234", + "EXCVADDR: 0x40001234", + "BT0: 0x40201234", + "last failed alloc call: 40201234(512)", + "Exception (28):", + ], + }, + PLATFORM_RP2: { + "state_markers": ["CRASH DETECTED ON PREVIOUS BOOT"], + "addresses": ["PC: 0x10001234 (fault location)"], + }, + PLATFORM_NRF52: { + "state_markers": ["Last crash:"], + "addresses": [ + # The zephyr logger prints both registers with %08x, so even + # a vector-table PC pads past the decoder's {3,} bound. + "PC=0x00000050 LR=0x00000000", + # Synthetic short form; keeps the bound's lower edge pinned. + "PC=0x27a1c LR=0x1e33", + ], + }, +} + +BENIGN_LINES = [ + "[I][app:100] hello world", + "[C][wifi:400] BSSID: AA:BB:CC:DD:EE:FF", + "[19:26:11.966][I][main:151]: version 2026.7.0-dev", + "[I][app:102]: Uptime: 12345678 ms", + "[I][app:102]: Uptime: 41234567 ms", + "[V][esp-idf:000]: I (40219876) wifi: connected", + "[D][api:102]: Client connected (40123456)", + "[D][sensor:093]: 'Water meter': Sending state 12345678.00000 L", + # A 32-hex hash has no internal word boundary, so the exactly-8 + # bare-hex branch must not fire anywhere inside it. + "[I][ota:117]: MD5 of binary: d41d8cd98f00b204e9800998ecf8427e", + # Short 0x tokens are everywhere (BLE handles, flags); the pointer + # branch's 3-digit minimum exists to keep them out. + "[D][ble:200]: Connection handle 0x1F, MTU 23", + "[C][network:600]: IPv6: fe80::1a2b:3c4d:5e6f:7a8b", + "[C][ota:097]: Version: 2026.7.0", +] + +GATE_PARAMS = [ + pytest.param(platform, line, True, id=f"{platform}-{kind}-{n}") + for platform, samples in CRASH_SAMPLES.items() + for kind in ("addresses", "state_markers", "extra_triggers") + for n, line in enumerate(samples.get(kind, [])) +] + [ + pytest.param(platform, line, False, id=f"benign-{platform}-{n}") + for platform in CRASH_SAMPLES + for n, line in enumerate(BENIGN_LINES) +] + + +@pytest.mark.parametrize(("platform", "line", "should_fire"), GATE_PARAMS) +def test_platform_gate(platform: str, line: str, should_fire: bool) -> None: + gate = re.compile(stacktrace.platform_hooks.STACKTRACE_GATES[platform]) + assert bool(gate.search(line)) is should_fire + + +def test_gates_are_platform_scoped() -> None: + """A session only pays for its own platform's trigger language. + + Another platform's marker on an esp32 session must not cost the + one-time import; the platform is known when the session starts. + """ + esp32_gate = re.compile(stacktrace.platform_hooks.STACKTRACE_GATES[PLATFORM_ESP32]) + for line in ( + ">>>stack>>>", + "Last crash:", + "Exception (28):", + "3ffffe10: 40201234 3ffe8410 00000000 40201000", + ): + assert not esp32_gate.search(line) + + +def _top_level_branches(pattern: str) -> list[str]: + """Split a regex source on alternations outside groups and classes.""" + branches: list[str] = [] + depth = 0 + in_class = False + esc = False + start = 0 + for i, ch in enumerate(pattern): + if esc: + esc = False + elif ch == "\\": + esc = True + elif in_class: + in_class = ch != "]" + elif ch == "[": + in_class = True + elif ch == "(": + depth += 1 + elif ch == ")": + depth -= 1 + elif ch == "|" and depth == 0: + branches.append(pattern[start:i]) + start = i + 1 + branches.append(pattern[start:]) + return branches + + +@pytest.mark.parametrize("platform", sorted(CRASH_SAMPLES)) +def test_every_gate_branch_is_exercised(platform: str) -> None: + """A typoed or dead gate branch cannot hide behind the others. + + The superset checks stay green when a branch matches nothing, so a + broken alternation would silently stop resolving the decoder on the + lines it was added for; every branch must be hit by a sample. + """ + samples = CRASH_SAMPLES[platform] + lines = [line for kind in samples for line in samples[kind]] + branches = _top_level_branches(stacktrace.platform_hooks.STACKTRACE_GATES[platform]) + assert len(branches) > 1 + for branch in branches: + assert any(re.search(branch, line) for line in lines), ( + f"no {platform} sample exercises gate branch {branch!r}; add one " + "or drop the dead branch" + ) + + +# The in-tree sources that print each marker literal the gates key on. +# esp8266's >>>stack>>> comes from the Arduino core's postmortem +# handler, outside this tree; its decoder source is the nearest pin. +FIRMWARE_MARKER_SOURCES = { + "CRASH DETECTED ON PREVIOUS BOOT": ( + "esphome/components/esp32/crash_handler.cpp", + "esphome/components/esp8266/crash_handler.cpp", + "esphome/components/rp2/crash_handler.cpp", + ), + "Last crash:": ("esphome/components/logger/logger_zephyr.cpp",), +} + + +def test_gate_markers_match_firmware_output() -> None: + """The marker literals must stay what the firmware prints. + + A reworded crash banner would keep every regex-level guard green + while the gate silently stops resolving the decoder at a stored + dump's first line; pin the literals to the sources that print them. + """ + root = Path(__file__).parents[2] + for marker, sources in FIRMWARE_MARKER_SOURCES.items(): + for source in sources: + text = (root / source).read_text(encoding="utf-8") + assert marker in text, ( + f"{source} no longer prints {marker!r}; update the gates and " + "samples to the new banner" + ) + + +def test_crash_samples_cover_registry() -> None: + """A newly registered decoder must come with a non-empty gate sample. + + The gate table and the hook registry cannot drift; the registry + entry is derived from the gate table's keys. + """ + assert set(CRASH_SAMPLES) == set(stacktrace.platform_hooks.STACKTRACE_GATES) + assert set(stacktrace.platform_hooks.STACKTRACE_GATES) == set( + stacktrace.platform_hooks.PLATFORM_HOOKS["process_stacktrace"] + ) + assert all(samples["addresses"] for samples in CRASH_SAMPLES.values()) + + +# The stacktrace pattern constants each decoder module exports. The +# samples and these patterns must cover each other, so an edit on either +# side fails the guards below instead of quietly widening the gap +# between the gate and the decoders. +DECODER_PATTERNS: dict[str, list[str]] = { + PLATFORM_ESP32: [ + "STACKTRACE_ESP32_PC_RE", + "STACKTRACE_ESP32_EXCVADDR_RE", + "STACKTRACE_ESP32_C3_PC_RE", + "STACKTRACE_ESP32_C3_RA_RE", + "STACKTRACE_ESP32_C3_MTVAL_RE", + "STACKTRACE_BAD_ALLOC_RE", + "STACKTRACE_ESP32_BACKTRACE_RE", + "STACKTRACE_ESP32_BACKTRACE_PC_RE", + "STACKTRACE_ESP32_CRASH_BT_RE", + ], + PLATFORM_ESP8266: [ + "STACKTRACE_ESP8266_EXCEPTION_TYPE_RE", + "STACKTRACE_ESP8266_PC_RE", + "STACKTRACE_ESP8266_EXCVADDR_RE", + "STACKTRACE_ESP8266_CRASH_PC_RE", + "STACKTRACE_ESP8266_CRASH_EXCVADDR_RE", + "STACKTRACE_ESP8266_CRASH_BT_RE", + "STACKTRACE_BAD_ALLOC_RE", + "STACKTRACE_ESP8266_BACKTRACE_PC_RE", + ], + PLATFORM_RP2: ["_CRASH_RE", "_CRASH_ADDR_RE"], + PLATFORM_NRF52: ["STACKTRACE_NRF52_PC_LR_RE"], +} + +# Declared decoder patterns whose language the gate deliberately does +# not cover: bare stack-dump words, where the gate keys on the dump +# line's 3ff... stack address instead and a lone letter-free word never +# appears outside a dump region whose other lines already fired. +GATE_EXEMPT_PATTERNS = { + "STACKTRACE_ESP32_BACKTRACE_PC_RE", + "STACKTRACE_ESP8266_BACKTRACE_PC_RE", +} + + +@pytest.mark.parametrize("platform", sorted(CRASH_SAMPLES)) +def test_platform_declarations_match_decoder(platform: str) -> None: + r"""Samples, declared patterns, and the decoder must agree. + + Directions checked: every declared pattern exists; every address + sample matches a declared pattern; every declared pattern is + exercised by a sample; no stacktrace pattern exists undeclared; each + declared state marker behaviourally opens the decoder's dump region; + and a decoder that sets state must declare a marker. + + Known blind spots: the esp32/esp8266 catch-all backtrace patterns + can satisfy the sample-matches-a-pattern direction on their own; the + undeclared-pattern sweep keys off naming, so a differently-named + constant or a function-local re.search literal is invisible to it; + the state-gated detection rests on a textual heuristic (every + such decoder today spells it as ``return True`` or + ``backtrace_state = True``, and the declared-markers direction + pins the heuristic against a silent respelling); a decoder that + gains a second opening marker alongside a declared one passes + unnoticed, since the declared marker already satisfies both the + marker-opens-region check and the non-empty ``state_markers`` + requirement; and the generative guard draws full matches only, so + a decoder match glued to trailing word characters that defeat the + pointer branch's ``\b`` is invisible to it (today's crash + handlers always delimit addresses). + """ + module = importlib.import_module(f"esphome.components.{platform}") + patterns: dict[str, re.Pattern] = {} + for name in DECODER_PATTERNS[platform]: + pattern = getattr(module, name, None) + if pattern is None: + pytest.fail( + f"{platform} no longer defines {name}; update DECODER_PATTERNS " + "and CRASH_SAMPLES together" + ) + patterns[name] = pattern + + lines = ( + CRASH_SAMPLES[platform]["state_markers"] + CRASH_SAMPLES[platform]["addresses"] + ) + for line in CRASH_SAMPLES[platform]["addresses"]: + assert any(p.search(line) for p in patterns.values()), ( + f"{line!r} no longer matches any {platform} decoder pattern; " + "update CRASH_SAMPLES and re-derive the gate" + ) + for name, pattern in patterns.items(): + assert any(pattern.search(line) for line in lines), ( + f"no sample exercises {platform}.{name}; add one so the gate " + "provably covers it" + ) + undeclared = [ + name + for name, value in vars(module).items() + if isinstance(value, re.Pattern) + and ("STACKTRACE" in name or name.startswith("_CRASH")) + and name not in DECODER_PATTERNS[platform] + ] + assert not undeclared, ( + f"{platform} gained stacktrace patterns {undeclared}; declare them in " + "DECODER_PATTERNS with samples" + ) + + for marker in CRASH_SAMPLES[platform]["state_markers"]: + assert module.process_stacktrace(CONFIG, marker, False) is True, ( + f"{marker!r} no longer opens {platform}'s dump region; update " + "state_markers to the line the decoder actually keys on" + ) + # Textual heuristic, deliberately one-directional: a state-gated + # decoder must declare a marker. The reverse (a stateless decoder + # declaring none) is not asserted; an unrelated "return True" added + # to a decoder would turn it into a false failure. + source = inspect.getsource(module.process_stacktrace) + sets_state = "return True" in source or "backtrace_state = True" in source + if CRASH_SAMPLES[platform]["state_markers"]: + # The heuristic fails open on a respelling (return bool(...)); + # pinning it against the decoders known to be state-gated today + # turns a silent disarm into a failure that names the fix. + assert sets_state, ( + f"{platform}.process_stacktrace declares state_markers but the " + "state-gating heuristic no longer recognises it; update the " + "spelling list in this test" + ) + if sets_state: + assert CRASH_SAMPLES[platform]["state_markers"], ( + f"{platform}.process_stacktrace is state-gated but declares no " + "state_markers; the gate cannot promise to open its dump region" + ) + + +@pytest.mark.parametrize( + ("platform", "name"), + [ + (platform, name) + for platform, names in DECODER_PATTERNS.items() + for name in names + if name not in GATE_EXEMPT_PATTERNS + ], +) +@given(data=st_data()) +@settings(max_examples=25, deadline=None) +def test_address_gate_covers_decoder_pattern_languages( + platform: str, name: str, data +) -> None: + """Each platform's gate must be a superset of its decoder patterns. + + The sample table only pins finite literals; a decoder regex that + widens would keep every sample green while the gate misses the new + form. Generating inputs from the decoder regex itself closes that + direction. + """ + pattern = getattr(importlib.import_module(f"esphome.components.{platform}"), name) + example = data.draw(from_regex(pattern, fullmatch=True)) + gate = re.compile(stacktrace.platform_hooks.STACKTRACE_GATES[platform]) + assert gate.search(example), ( + f"{platform}.{name} accepts {example!r} but the {platform} gate does " + "not fire; decoding would silently never start on that form" + ) + def _run( handler, @@ -21,8 +388,8 @@ def _run( stacktrace.platform_hooks, "get_stacktrace_handler", return_value=handler ): processor = stacktrace.LogLineProcessor(CONFIG, platform) - for line in lines: - processor.process_line(line) + for line in lines: + processor.process_line(line) return processor @@ -30,7 +397,7 @@ def _fed(handler) -> list[str]: return [call.args[1] for call in handler.call_args_list] -def _warnings(caplog) -> list[str]: +def _warnings(caplog: pytest.LogCaptureFixture) -> list[str]: return [r.message for r in caplog.records if r.levelname == "WARNING"] @@ -51,23 +418,9 @@ def test_decoder_contains_failures_and_short_circuits() -> None: 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: +def test_decoder_swallows_os_error_with_remediation_hint( + caplog: pytest.LogCaptureFixture, +) -> None: """Decoding failures that aren't EsphomeError must be contained too. A missing build directory surfaces as an OSError; that is the @@ -86,7 +439,9 @@ def test_decoder_swallows_os_error_with_remediation_hint(caplog) -> None: assert not any("this is a bug" in m for m in warnings) -def test_decoder_warning_uses_fallback_for_empty_error(caplog) -> None: +def test_decoder_warning_uses_fallback_for_empty_error( + caplog: pytest.LogCaptureFixture, +) -> None: """A message-less EsphomeError must show a useful explanation. Defensive: the in-tree idedata raise sites all carry a message now, @@ -99,7 +454,9 @@ def test_decoder_warning_uses_fallback_for_empty_error(caplog) -> None: assert not any("()" in m for m in warnings) -def test_decoder_bug_with_empty_message_names_the_type(caplog) -> None: +def test_decoder_bug_with_empty_message_names_the_type( + caplog: pytest.LogCaptureFixture, +) -> None: """A zero-message decoder bug must not masquerade as missing artifacts. The recompile hint is only right for EsphomeError from _run_idedata; @@ -113,7 +470,9 @@ def test_decoder_bug_with_empty_message_names_the_type(caplog) -> None: assert not any("esphome compile" in m for m in warnings) -def test_decoder_bug_warning_keeps_the_type_with_a_message(caplog) -> None: +def test_decoder_bug_warning_keeps_the_type_with_a_message( + caplog: pytest.LogCaptureFixture, +) -> 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. """ @@ -123,8 +482,12 @@ def test_decoder_bug_warning_keeps_the_type_with_a_message(caplog) -> None: 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.""" +def test_marker_then_address_threads_state() -> None: + """A state marker resolves the decoder and threads state onward. + + esp8266's ``>>>stack>>>`` fires the gate itself, so the decoder sees + it live and the following stack words decode inside the region. + """ handler = Mock(side_effect=[True, True]) processor = _run( handler, @@ -141,11 +504,120 @@ def test_state_threads_between_lines() -> None: 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") +def test_lines_before_the_gate_never_reach_the_decoder() -> None: + """Benign lines are dropped, not buffered: the gate is a superset of + the decoder languages, so a line that fails it cannot decode. + """ + handler = Mock(return_value=False) + quiet = tuple(f"quiet line {n}" for n in range(12)) + _run(handler, lines=quiet + ("PC: 0x4010496e",)) + assert _fed(handler) == ["PC: 0x4010496e"] + + +def test_processor_resolves_lazily_on_address_token() -> None: + """No resolution attempt until a line carries an address token.""" + handler = Mock(return_value=False) + + with patch.object( + stacktrace.platform_hooks, "get_stacktrace_handler", return_value=handler + ) as mock_resolve: + processor = stacktrace.LogLineProcessor(CONFIG, PLATFORM_ESP32) + processor.process_line("[I][app:100] hello world") + mock_resolve.assert_not_called() + + processor.process_line("PC: 0x40104960") + mock_resolve.assert_called_once_with(PLATFORM_ESP32) + + # Later lines feed the resolved handler directly, no re-resolution. + processor.process_line("[I][app:101] back to normal") + mock_resolve.assert_called_once() + + assert _fed(handler) == ["PC: 0x40104960", "[I][app:101] back to normal"] + + +def test_processor_unexpected_resolution_error_disables_decoding( + caplog: pytest.LogCaptureFixture, +) -> None: + """Resolution is inside the containment guarantee like everything else.""" + with patch.object( + stacktrace.platform_hooks, + "get_stacktrace_handler", + side_effect=OSError("filesystem went away"), + ) as mock_resolve: + processor = stacktrace.LogLineProcessor(CONFIG, PLATFORM_ESP32) + processor.process_line("PC: 0x40104960") + processor.process_line("BT0: 0x40104960") + + mock_resolve.assert_called_once() + warnings = _warnings(caplog) + assert len(warnings) == 1 + assert "could not be loaded" in warnings[0] + assert processor.backtrace_state is False + + +def test_processor_import_failure_disables_decoding( + caplog: pytest.LogCaptureFixture, +) -> None: + """A broken platform package degrades once instead of raising.""" + caplog.set_level("INFO", logger="esphome.platform_hooks") + + with patch.object( + stacktrace.platform_hooks, + "import_module", + Mock(side_effect=ImportError("broken install")), + ) as mock_import: + processor = stacktrace.LogLineProcessor(CONFIG, PLATFORM_ESP32) + processor.process_line("PC: 0x40104960") + processor.process_line("BT0: 0x40104960") + + mock_import.assert_called_once() + assert "Stacktrace analysis is unavailable" in caplog.text + assert "broken install" in caplog.text + assert processor.backtrace_state is False + + +def test_processor_registry_miss_disables_at_construction( + caplog: pytest.LogCaptureFixture, +) -> None: + """Platforms the registry proves have no analyzer disable up front. + + The unavailable notice fires at session start (as it always did) and + the per-line gate never runs. + """ + caplog.set_level("INFO", logger="esphome.platform_hooks") + + with patch.object(stacktrace.platform_hooks, "import_module") as mock_import: + processor = stacktrace.LogLineProcessor(CONFIG, PLATFORM_BK72XX) + processor.process_line("PC: 0x40104960") + + mock_import.assert_not_called() assert "Stacktrace analysis is unavailable" in caplog.text assert processor.backtrace_state is False + + +def test_external_platform_resolves_at_construction( + caplog: pytest.LogCaptureFixture, +) -> None: + """External platforms resolve eagerly, like before the registry. + + The address gate's grammar derives from the in-tree decoders, so it + cannot speak for an external decoder; resolving up front keeps the + import off the streaming callback and the notice at session start. + """ + caplog.set_level("INFO", logger="esphome.platform_hooks") + module = type("ExternalPlatform", (), {}) # no process_stacktrace + + with patch.object( + stacktrace.platform_hooks, + "import_module", + Mock(return_value=module), + ) as mock_import: + processor = stacktrace.LogLineProcessor(CONFIG, "my_external_chip") + mock_import.assert_called_once() + assert "Stacktrace analysis is unavailable" in caplog.text + + processor.process_line("PC: 0x40104960") + + mock_import.assert_called_once() + assert processor.backtrace_state is False