260 lines
11 KiB
Python
260 lines
11 KiB
Python
"""Diagnostics engine: real captures (fixtures) + synthetic scenarios built
|
|
from the firmware's exact log strings for failures not captured yet."""
|
|
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from kota_tech.core.logfile import read_log_file
|
|
from kota_tech.core.logline import LineKind, LogLine, parse_line
|
|
from kota_tech.diag.engine import DiagnosticsEngine, Severity, load_rules
|
|
|
|
FIX = Path(__file__).parent / "fixtures"
|
|
|
|
|
|
def run(lines, engine=None):
|
|
e = engine or DiagnosticsEngine()
|
|
t = 0.0
|
|
for item in lines:
|
|
t += 0.01
|
|
if isinstance(item, LogLine):
|
|
e.feed(item)
|
|
else:
|
|
e.feed(parse_line(item, host_time=t))
|
|
return e, e.report()
|
|
|
|
|
|
def replay(name):
|
|
e = DiagnosticsEngine()
|
|
for ln in read_log_file(FIX / name):
|
|
e.feed(ln)
|
|
return e.report()
|
|
|
|
|
|
def roots(report):
|
|
return {rc.finding.rule_id: rc for rc in report.roots}
|
|
|
|
|
|
BOOT = ["ESP-ROM:esp32s3-20210327"]
|
|
APP = ["I (1299) app_init: Project name: kota-sensorboard",
|
|
"I (1317) app_init: ELF file SHA256: 45fa9d5b9..."]
|
|
|
|
|
|
# ------------------------------------------------------------------ rules
|
|
|
|
def test_rules_load_and_have_no_cycles():
|
|
from kota_tech.diag import kota
|
|
graph = {r.id: r.explained_by for r in load_rules()}
|
|
for st in ("ERROR", "OUT_OF_RESOURCES", "CONNECTION_ERROR", "RTC_ERROR", "SD_ERROR", "RESETING", "HTTP_SERVER"):
|
|
for code in range(7):
|
|
if sf := kota.status_finding(st, code):
|
|
graph[sf[0]] = sf[5]
|
|
for start, explained_by in graph.items():
|
|
seen, todo = set(), list(explained_by)
|
|
while todo:
|
|
rid = todo.pop()
|
|
assert rid != start, f"explained_by cycle through {start}"
|
|
if rid not in seen:
|
|
seen.add(rid)
|
|
todo += graph.get(rid, ())
|
|
|
|
|
|
# --------------------------------------------------------- real captures
|
|
|
|
def test_live_bench_box_uncommissioned():
|
|
r = replay("live_boot_A4CB8FC946F8_fw1.0.txt")
|
|
rs = roots(r)
|
|
assert r.firmware.known and "new_system.bin" in r.firmware.describe()
|
|
assert r.firmware.supports_sd_wifi is False
|
|
assert r.led == "PURPLE" # confirmed on the bench
|
|
assert "Wi-Fi" in r.verdict and r.verdict.startswith("Problem")
|
|
grouped = {c.rule_id for c in rs["matter.not_commissioned"].consequences}
|
|
assert {"net.dns_fail", "influx.buffer_fail", "ota.registration_failed"} <= grouped
|
|
assert "rtc.clock_not_running" in rs
|
|
assert "mic.disabled" in rs
|
|
reached = {m for m, _, t in r.milestones if t is not None}
|
|
assert {"rom_boot", "app_started", "sd_mounted", "env_loaded", "sensors_ready"} <= reached
|
|
assert "wifi_connected" not in reached
|
|
|
|
|
|
def test_live_bench_box_led_still_green_at_30s():
|
|
# Purple only starts when the first Influx flush fails (~60 s uptime).
|
|
assert replay("live_reset_A4CB8FC946F8_fw1.0_restored.txt").led == "GREEN"
|
|
|
|
|
|
def test_connected_box_is_healthy():
|
|
r = replay("serial_log_20260917_162922.txt")
|
|
assert r.verdict_severity is None
|
|
assert "time_synced" in {m for m, _, t in r.milestones if t is not None}
|
|
assert r.mid_run
|
|
|
|
|
|
@pytest.mark.parametrize("name", ["serial_log_20260917_162714.txt", "kota_serial_log_20260917_163439.txt"])
|
|
def test_offline_sample_boxes(name):
|
|
rs = roots(replay(name))
|
|
assert "net.dns_fail" in rs and "rtc.clock_not_running" in rs
|
|
|
|
|
|
# ---------------------------------------------------- synthetic scenarios
|
|
|
|
def _rtc_fail_boot(reset="RTC_SW_CPU_RST"):
|
|
return BOOT + [f"rst:0xc ({reset}),boot:0x8 (SPI_FAST_FLASH_BOOT)"] + APP + [
|
|
"I (2493) MAIN: ESP-Matter started successfully",
|
|
"E (3510) DS1307: Failed to initialize DS1307 control register: ESP_ERR_TIMEOUT",
|
|
"E (3511) MAIN: Failed to initialize RTC: ESP_ERR_TIMEOUT",
|
|
"ESP_ERROR_CHECK failed: esp_err_t 0x107 (ESP_ERR_TIMEOUT) at 0x4200f7b2",
|
|
'file: "./main/main.cpp" line 473',
|
|
"abort() was called at PC 0x40379b1f on core 0",
|
|
]
|
|
|
|
|
|
def test_rtc_i2c_reboot_loop():
|
|
_, r = run(_rtc_fail_boot("POWERON") + _rtc_fail_boot() + _rtc_fail_boot() + _rtc_fail_boot())
|
|
rs = roots(r)
|
|
root = rs["rtc.init_failed"]
|
|
assert root.finding.severity == Severity.CRITICAL
|
|
assert "I2C" in root.finding.meaning
|
|
grouped = {c.rule_id for c in root.consequences}
|
|
assert {"sys.crash", "reboot.loop"} <= grouped
|
|
assert r.led == "BLUE"
|
|
assert r.verdict.startswith("Problem")
|
|
|
|
|
|
def test_sd_card_missing():
|
|
_, r = run(BOOT + APP + ["E (1519) SD_CARD: Failed to initialize the card (ESP_ERR_TIMEOUT). "
|
|
"Make sure SD card lines have pull-up resistors in place."])
|
|
f = roots(r)["sd.card_fail"].finding
|
|
assert "ESP_ERR_TIMEOUT" in f.title and f.severity == Severity.CRITICAL
|
|
assert r.led == "WHITE"
|
|
|
|
|
|
def test_influx_token_rejected():
|
|
_, r = run(BOOT + APP + ["E (30000) INFLUX: Failed to perform HTTP request: 401, ESP_OK"])
|
|
f = roots(r)["influx.auth"].finding
|
|
assert "401" in f.title and f.severity == Severity.CRITICAL
|
|
|
|
|
|
def test_recovery_resolves_network_findings():
|
|
lines = BOOT + APP + [
|
|
"E (21837) esp-tls: couldn't get hostname for :influx.kota-app.com: getaddrinfo() returns 202, addrinfo=0x0",
|
|
"I (30222) SNTP: Time synchronized: Thu Sep 17 14:29:22 2026",
|
|
"I (31000) INFLUX: HTTP request performed, status code: 204",
|
|
]
|
|
_, r = run(lines)
|
|
assert "net.dns_fail" not in roots(r)
|
|
assert any(f.rule_id == "net.dns_fail" for f in r.resolved)
|
|
assert r.verdict.startswith("Healthy")
|
|
|
|
|
|
def test_led_conditional_green_after_purple():
|
|
e, r = run(BOOT + APP + ["E (60009) INFLUX_BUFFER: Failed to write chunk with 5 retries to Influx: ESP_FAIL"])
|
|
assert r.led == "PURPLE"
|
|
_, r = run(["I (61000) OTA_CLIENT: No update available"], e)
|
|
assert r.led == "GREEN"
|
|
_, r = run(["I (62000) OTA_CLIENT: New update available, starting OTA update..."], e)
|
|
assert r.led == "YELLOW"
|
|
|
|
|
|
def test_status_watchdog_restart_grouped_under_network_cause():
|
|
lines = BOOT + APP + [
|
|
"I (2542) MAIN: Commissioning window opened",
|
|
"E (60009) INFLUX_BUFFER: Failed to write chunk with 5 retries to Influx: ESP_FAIL",
|
|
"W (200000) BME688: x",
|
|
] + BOOT + ["rst:0x3 (RTC_SW_SYS_RST),boot:0x8 (SPI_FAST_FLASH_BOOT)"] + APP + [
|
|
"I (2542) MAIN: Commissioning window opened",
|
|
"E (60009) INFLUX_BUFFER: Failed to write chunk with 5 retries to Influx: ESP_FAIL",
|
|
]
|
|
_, r = run(lines)
|
|
grouped = {c.rule_id for c in roots(r)["matter.not_commissioned"].consequences}
|
|
assert "reboot.status_watchdog" in grouped
|
|
assert "reboot.loop" not in grouped # 200 s boots are not a loop
|
|
|
|
|
|
def test_tool_reset_is_not_a_finding():
|
|
lines = BOOT + APP + [LogLine("--- reset issued by tool ---", LineKind.HOST, 0.5)]
|
|
lines += BOOT + ["rst:0x15 (USB_UART_CHIP_RESET),boot:0x2b (SPI_FAST_FLASH_BOOT)"] + APP
|
|
_, r = run(lines)
|
|
assert not [rc for rc in r.roots if rc.finding.rule_id.startswith("reboot.")]
|
|
assert len(r.boots) == 3 # mid-run record + 2 boots
|
|
|
|
|
|
def test_first_seen_is_per_boot_and_own_severity_breaks_ties():
|
|
rtc = "W ({}) BME688: RTC is uninitialized, using current time as timestamp"
|
|
lines = [rtc.format(200_000 + i) for i in range(5)] # before the reset
|
|
lines += BOOT + APP + [rtc.format(15_000 + i) for i in range(5)]
|
|
lines += ["W (24756) MICROPHONE: Microphone is in shutdown state, skipping sample collection",
|
|
"E (24773) SENSOR_GATHERER: Failed to get line protocol from sensor"] * 2
|
|
_, r = run(lines)
|
|
order = [rc.finding.rule_id for rc in r.roots]
|
|
assert order.index("rtc.clock_not_running") < order.index("mic.disabled")
|
|
assert roots(r)["rtc.clock_not_running"].finding.first_uptime_ms == 15_000
|
|
|
|
|
|
def test_reboot_detected_without_banner():
|
|
e, r = run(["I (90000) A: running", "I (95000) A: running", "I (1200) B: fresh boot"])
|
|
assert len(r.boots) == 2 and not r.boots[1].explicit
|
|
|
|
|
|
def test_split_by_sensor():
|
|
_, r = run(BOOT + APP + [
|
|
"E (14000) MAIN: Failed to initialize BME688: ESP_FAIL",
|
|
"E (14001) MAIN: Failed to initialize SEN66: ESP_ERR_TIMEOUT",
|
|
])
|
|
titles = {rc.finding.title for rc in r.roots if rc.finding.rule_id == "sensor.init_failed"}
|
|
assert titles == {"BME688 failed to start", "SEN66 failed to start"}
|
|
assert r.led == "RED"
|
|
|
|
|
|
def test_rtc_loses_time_after_sync():
|
|
_, r = run(BOOT + APP + ["I (10222) SNTP: Time synchronized: Thu Sep 17 14:29:22 2026"]
|
|
+ [f"W ({30000 + i}) BME688: RTC is uninitialized, using current time as timestamp" for i in range(3)])
|
|
assert "rtc.lost_after_sync" in roots(r)
|
|
|
|
|
|
def test_unknown_firmware_reports_sd_wifi_from_evidence():
|
|
_, r = run(BOOT + ["I (1317) app_init: ELF file SHA256: 0123456789..."] + [
|
|
"I (1523) WIFI_CREDENTIALS: SSID/PASSWORD not set in .env; Matter will be used"])
|
|
assert r.firmware.known is None and r.firmware.supports_sd_wifi is True
|
|
assert "OTA" in r.firmware.describe()
|
|
|
|
|
|
def test_report_text_renders():
|
|
text = replay("live_boot_A4CB8FC946F8_fw1.0.txt").to_text()
|
|
assert text.startswith("VERDICT: Problem") and "consequence:" in text
|
|
|
|
|
|
# ------------------------------------------------------------- log files
|
|
|
|
def test_session_log_format(tmp_path):
|
|
p = tmp_path / "session.log"
|
|
p.write_text("[14:00:00.000] --- reset issued by tool ---\n"
|
|
"[14:00:00.400] ESP-ROM:esp32s3-20210327\n"
|
|
"[14:00:00.401] rst:0x15 (USB_UART_CHIP_RESET),boot:0x2b (SPI_FAST_FLASH_BOOT)\n"
|
|
"[14:00:01.500] I (1299) app_init: Project name: kota-sensorboard\n", encoding="utf-8")
|
|
lines = read_log_file(p)
|
|
assert [ln.kind for ln in lines] == [LineKind.HOST, LineKind.BOOT, LineKind.RESET, LineKind.LOG]
|
|
e = DiagnosticsEngine()
|
|
for ln in lines:
|
|
e.feed(ln)
|
|
assert e.boot.tool_reset
|
|
|
|
|
|
def test_live_brownout_loop_is_a_power_problem():
|
|
# Real capture (2026-09-28): "E BOD: Brownout detector was triggered" has no
|
|
# timestamp and the ROM reports RTC_SW_SYS_RST, not a brown-out reset.
|
|
r = replay("live_brownout_loop_A4CB8FC946F8_fw1.0.txt")
|
|
top = r.roots[0]
|
|
assert top.finding.rule_id == "reboot.brownout" and top.finding.count >= 5
|
|
assert "reboot.loop" in {c.rule_id for c in top.consequences}
|
|
assert r.verdict.startswith("Problem: Power supply")
|
|
|
|
|
|
@pytest.mark.parametrize("line", [
|
|
"E (2893) wifi:sta is connecting, return error",
|
|
"E (2892) chip[DL]: esp_wifi_connect() failed: ESP_ERR_WIFI_CONN",
|
|
"E (5120) chip[DL]: Failed to get configured network when updating network status: Error ESP32:0x0500300F",
|
|
])
|
|
def test_harmless_wifi_start_errors_are_not_reported(line):
|
|
_, r = run(BOOT + APP + [line])
|
|
assert r.roots == [] and "Unrecognised" not in r.to_text()
|