"""Firmware 1.0.16+ structured "KOTA:" lines and error IDs. The lines below are built from the exact format strings in the firmware source (boot_info.cpp, status.cpp, i2c.cpp, ds1307.cpp, sd_card.cpp, main.cpp, microphone.cpp, health.cpp); no hardware capture exists yet. """ from pathlib import Path from kota_tech.core.logfile import read_log_file from kota_tech.core.logline import LineKind, parse_line from kota_tech.diag.engine import DiagnosticsEngine, Severity from kota_tech.diag.firmware import same_build from kota_tech.matter.payload import payload_from_log from kota_tech.service.report import CONNECTIVITY, HARDWARE, service_status FIX = Path(__file__).parent / "fixtures" BANNER = ["ESP-ROM:esp32s3-20210327", "rst:0x15 (USB_UART_CHIP_RESET),boot:0x8 (SPI_FAST_FLASH_BOOT)"] I2C_OK = "I (1790) KOTA: i2c SDA=1 SCL=1 0x39=AS7341 0x68=DS1307 0x6B=SEN66 0x76=BME688" SD_OK = ("I (1990) KOTA: sd type=SDHC size=7.4GB fs=FAT32 free=7.3GB " "env_keys=INFLUX_URL,INFLUX_TOKEN buffer_lines=0") BOOT = ("I (2310) KOTA: boot fw=1.0.16 build=4f10989e hw=2 mac=A4CB8FC946F8 reset=USB_UART_CHIP_RESET " "mode=matter commissioned=1 last_restart={last}") def run(lines): e = DiagnosticsEngine() for i, text in enumerate(lines): e.feed(parse_line(text, host_time=i * 0.01)) return e, e.report() def roots(report): return {rc.finding.rule_id: rc for rc in report.roots} # ------------------------------------------------------------- parsing def test_kota_lines_are_parsed_into_fields(): ln = parse_line(BOOT.format(last="status_watchdog")) assert ln.kind == LineKind.KOTA and ln.record == "boot" and ln.level == "I" assert ln.fields["fw"] == "1.0.16" and ln.fields["last_restart"] == "status_watchdog" st = parse_line("I (5) KOTA: status OK -> CONNECTION_ERROR code=1 reason=influx_rejected") assert st.words == ("OK", "->", "CONNECTION_ERROR") and st.fields == {"code": "1", "reason": "influx_rejected"} mic = parse_line("I (6) KOTA: mic enabled=0 (GPIO3=1, expected 0)") assert mic.fields["enabled"] == "0" and mic.fields["GPIO3"] == "1" i2c = parse_line("I (7) KOTA: i2c SDA=1 SCL=1 0x39=AS7341 0x68=DS1307(missing) 0x50=?") assert i2c.fields["0x68"] == "DS1307(missing)" and i2c.fields["0x50"] == "?" verify = parse_line("I (8) KOTA: rtc verify running=1 delta=2s ok=1") assert verify.record == "rtc" and "verify" in verify.words and verify.fields["ok"] == "1" def test_error_id_is_split_off_but_kept_in_text(): ln = parse_line("E (1200) SD_CARD: [E-SD-03] Failed to initialize the card (ESP_ERR_TIMEOUT).") assert ln.kind == LineKind.LOG and ln.error_id == "E-SD-03" assert ln.message.startswith("Failed to initialize") and "[E-SD-03]" in ln.text assert ln.match_text == "E (1200) SD_CARD: Failed to initialize the card (ESP_ERR_TIMEOUT)." def test_sd_line_with_secret_key_names_is_not_redacted(): ln = parse_line("I (9) KOTA: sd type=SDHC size=7.4GB fs=FAT32 free=7.3GB " "env_keys=INFLUX_URL,INFLUX_TOKEN,SSID,PASSWORD buffer_lines=0") assert ln.fields["env_keys"] == "INFLUX_URL,INFLUX_TOKEN,SSID,PASSWORD" # -------------------------------------------------------------- engine def test_healthy_box_from_kota_lines(): e = DiagnosticsEngine() for ln in read_log_file(FIX / "synthetic_fw1.0.16_healthy.txt"): e.feed(ln) r = e.report() assert r.verdict.startswith("Healthy") assert (r.led, r.led_pattern, r.led_exact, r.led_meaning) == ("GREEN", "solid", True, "OK") assert "v1.0.16" in r.firmware.describe() and "4f10989e" in r.firmware.describe() assert e.mac == "A4:CB:8F:C9:46:F8" reached = {m for m, _, t in r.milestones if t is not None} assert {"i2c_ok", "sd_mounted", "env_loaded", "matter_commissioned", "wifi_connected", "time_synced", "clock_verified", "data_uploaded", "ota_registered"} <= reached topics = {t: st for t, _, st in r.box} assert topics["I2C"] == topics["SD card"] == topics["Health"] == "ok" # The purple state before the tool's reset is shown, but only as information now. prev = [f for f in r.earlier + [rc.finding for rc in r.roots] if f.rule_id == "boot.previous_status"] assert not prev or prev[0].severity == Severity.INFO assert "E-MAIN-04" not in r.to_text() # MAC line is a benign ESP_LOGE def test_i2c_bus_stuck_is_root_of_blue_led(): _, r = run(BANNER + [ "I (1480) KOTA: led color=blue pattern=pulses n=2", "I (1790) KOTA: i2c SDA=0 SCL=1 bus=stuck", "I (1791) KOTA: status OK -> RTC_ERROR code=2 reason=i2c_bus_stuck", ]) rs = roots(r) assert "kota.i2c_stuck" in rs and rs["kota.i2c_stuck"].finding.severity == Severity.CRITICAL assert "status.rtc.2" in {c.rule_id for c in rs["kota.i2c_stuck"].consequences} assert (r.led, r.led_pattern) == ("BLUE", "2 pulses") and "I2C bus stuck" in r.led_meaning assert service_status(r) == HARDWARE def test_missing_sensor_named_by_i2c_scan_and_red_pulses(): _, r = run(BANNER + [ "I (1790) KOTA: i2c SDA=1 SCL=1 0x39=AS7341 0x68=DS1307 0x6B=SEN66(missing) 0x76=BME688", SD_OK, "E (2100) MAIN: [E-MAIN-16] Failed to initialize SEN66: ESP_ERR_TIMEOUT", "I (2101) KOTA: status OK -> ERROR code=2 reason=sen66_init_failed", "I (2102) KOTA: led color=red pattern=pulses n=2", ]) rs = roots(r) root = rs["kota.sensor_missing"] assert root.finding.title == "SEN66 does not answer on the I2C bus" grouped = {c.rule_id for c in root.consequences} assert {"status.error.2", "sensor.init_failed"} <= grouped init = next(c for c in root.consequences if c.rule_id == "sensor.init_failed") assert "E-MAIN-16" in init.error_ids and "SEN66" in init.error_ids["E-MAIN-16"] assert r.led_meaning == "SEN66 failed" assert "missing: SEN66" in dict((t, x) for t, x, _ in r.box)["I2C"] def test_server_rejected_box_is_purple_blinking(): _, r = run(BANNER + [I2C_OK, SD_OK, BOOT.format(last="none"), "I (70000) KOTA: status OK -> CONNECTION_ERROR code=1 reason=influx_rejected", "I (70001) KOTA: led color=purple pattern=blink period=1000ms"]) assert r.roots[0].finding.rule_id == "status.conn.1" assert r.roots[0].finding.severity == Severity.CRITICAL assert (r.led, r.led_pattern) == ("PURPLE", "blinking (1000 ms)") assert "server rejected" in r.led_meaning assert "influx_rejected" in r.led_reason assert service_status(r) == CONNECTIVITY def test_status_back_to_ok_resolves_the_status_finding(): _, r = run(BANNER + [I2C_OK, "I (5000) KOTA: status OK -> CONNECTION_ERROR reason=influx_failed", "I (9000) KOTA: status CONNECTION_ERROR -> OK reason=influx_ok"]) assert "status.conn" not in roots(r) assert any(f.rule_id == "status.conn" and f.resolved_by == "status -> OK" for f in r.resolved) def test_restart_reason_from_the_next_boot_line(): _, r = run(BANNER + [I2C_OK, BOOT.format(last="none"), "I (5000) KOTA: status OK -> ERROR code=3 reason=as7341_failed", "W (155000) KOTA: restart reason=status_watchdog", "I (155001) KOTA: led restart_flash color=red", "ESP-ROM:esp32s3-20210327", "rst:0xc (RTC_SW_CPU_RESET),boot:0x8 (SPI_FAST_FLASH_BOOT)", "I (1480) KOTA: led previous_boot status=ERROR code=3", I2C_OK, BOOT.format(last="status_watchdog")]) ids = {f.rule_id: f for rc in r.roots for f in [rc.finding, *rc.consequences]} f = ids["reboot.status_watchdog"] assert "150 s" in f.title and "state ERROR code=3" in f.meaning def test_brownout_restart_recorded_by_firmware(): _, r = run(BANNER + [I2C_OK, BOOT.format(last="none"), "I (90000) KOTA: health up=90s status=OK " "wifi=connected rssi=-60 time_valid=1 sd=ok buffer=0 heap=150000 psram=7.6M " "last_upload=10s ota=registered", "ESP-ROM:esp32s3-20210327", "rst:0xf (RTCWDT_BROWN_OUT_RESET),boot:0x8", I2C_OK, BOOT.format(last="brownout")]) assert r.roots[0].finding.rule_id == "reboot.brownout" def test_uncaptured_error_id_uses_the_csv_action(): _, r = run(BANNER + [I2C_OK, "E (3000) INFLUX_BUFFER: [E-INF-13] Failed to open buffer file for appending"]) f = next(f for rc in r.roots for f in [rc.finding, *rc.consequences] if "E-INF-13" in f.error_ids) # A rule already knows this message, so the rule finding carries the code. assert f.rule_id == "sd.write_fail" _, r = run(BANNER + [I2C_OK, "E (3000) SPI: [E-SPI-01] Failed to take SPI mutex"]) f = roots(r)["eid.SPI"].finding assert f.title == "SPI bus: Failed to take SPI mutex" assert "firmware team" in f.actions[0] and f.error_ids == {"E-SPI-01": f.actions[0]} def test_legacy_rules_still_match_lines_with_error_ids(): _, r = run(BANNER + ["E (1200) MAIN: [E-MAIN-13] Failed to initialize RTC: ESP_ERR_TIMEOUT"]) assert "rtc.init_failed" in roots(r) def test_health_line_gives_status_when_connected_mid_run(): e, r = run(["I (600000) KOTA: health up=600s status=CONNECTION_ERROR wifi=disconnected rssi=0 " "time_valid=1 sd=ok buffer=120 heap=150000 psram=7.6M last_upload=never ota=registered", "I (660000) KOTA: health up=660s status=CONNECTION_ERROR wifi=disconnected rssi=0 " "time_valid=1 sd=ok buffer=130 heap=150000 psram=7.6M last_upload=never ota=registered"]) assert (r.led, r.led_pattern) == ("PURPLE", "solid") and not r.led_exact rs = roots(r) assert "kota.wifi_down" in rs assert "status.conn" in {c.rule_id for c in rs["kota.wifi_down"].consequences} assert r.mid_run def test_sd_and_rtc_self_report(): _, r = run(BANNER + [ "I (1795) KOTA: rtc running=0 time=2000-01-01T00:00:00", "I (1990) KOTA: sd type=SDHC size=7.4GB fs=FAT32 free=7.3GB env_keys=INFLUX_URL buffer_lines=0", "I (8100) KOTA: rtc verify running=0 delta=0s ok=0", ]) ids = {f.key for rc in r.roots for f in [rc.finding, *rc.consequences]} assert {"influx.env_missing:INFLUX_TOKEN", "kota.rtc_stopped", "kota.rtc_not_ticking"} <= ids _, r = run(BANNER + ["I (1990) KOTA: sd mounted=0 err=ESP_ERR_TIMEOUT", "I (1991) KOTA: status OK -> SD_ERROR reason=sd_mount_failed"]) rs = roots(r) assert "kota.sd_unmounted" in rs and "ESP_ERR_TIMEOUT" in rs["kota.sd_unmounted"].finding.title assert "status.sd" in {c.rule_id for c in rs["kota.sd_unmounted"].consequences} def test_matter_and_mic_lines(): e, r = run(BANNER + ["I (2315) KOTA: matter fabrics=0 window=open qr=MT:Y.K9042C00KA0648G00 manual=34970112332", "I (2400) KOTA: mic enabled=0 (GPIO3=1, expected 0)", "I (4481) KOTA: led color=cyan pattern=breathe period=3000ms"]) rs = roots(r) assert "matter.not_commissioned" in rs and "mic.disabled" in rs assert (r.led, r.led_meaning) == ("CYAN", "waiting to be paired in Matter") p = payload_from_log([], e.matter_qr, e.matter_manual) assert p.qr_text() == "MT:Y.K9042C00KA0648G00" and p.manual_code_pretty() == "3497-011-2332" _, r = run(BANNER + ["I (2315) KOTA: matter fabrics=0 window=open qr=MT:Y.K9042C00KA0648G00 manual=34970112332", "I (90000) KOTA: matter fabrics=1 window=closed qr=MT:Y.K9042C00KA0648G00 manual=34970112332"]) assert "matter.not_commissioned" not in roots(r) def test_build_prefixes_of_different_length_match(): full = "4f10989e2b40c1d29fb3e8f8faaa50e5555997909c6d6b1db9d4e1e4d04555dc" assert same_build("4f10989e", full) and same_build(full, "4f10989e2") assert not same_build("4f10989f", full) and not same_build(None, full) def test_live_fw1016_uncommissioned_box(): # Real capture (2026-09-28) after flashing 1.0.16 App-only: not paired in Matter. e = DiagnosticsEngine() for ln in read_log_file(FIX / "live_boot_A4CB8FC946F8_fw1.0.16_uncommissioned.txt"): e.feed(ln) r = e.report() assert r.firmware.version == "1.0.16" and same_build(r.firmware.elf_sha, "4f10989e") assert (r.led, r.led_pattern, r.led_exact) == ("CYAN", "breathing", True) rs = roots(r) assert r.roots[0].finding.rule_id == "matter.not_commissioned" grouped = {c.rule_id for c in rs["matter.not_commissioned"].consequences} assert {"kota.wifi_down", "net.dns_fail", "status.conn", "kota.buffer_backlog"} <= grouped assert "rtc.clock_not_running" in {c.rule_id for c in rs["kota.rtc_stopped"].consequences} assert e.matter_qr == "MT:Y.K9042C00KA0648G00" assert not any(f.rule_id.startswith("other.") for f in e.findings.values())