"""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")