mirror of
https://github.com/esphome/esphome.git
synced 2026-08-23 06:36:23 +00:00
[core] Resolve the stacktrace decoder lazily behind an address gate (#18048)
This commit is contained in:
@@ -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
|
||||
|
||||
@@ -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,
|
||||
),
|
||||
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user