232 lines
8.3 KiB
Python
232 lines
8.3 KiB
Python
"""Guided troubleshooting: runner + engine with simulated technician
|
|
answers, simulated board output and a fake clock."""
|
|
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from kota_tech.core.logfile import read_log_file
|
|
from kota_tech.core.logline import parse_line
|
|
from kota_tech.diag.engine import DiagnosticsEngine, Severity
|
|
from kota_tech.diag.guides import LED_ANSWERS, GuideRunner, load_guides, recommended
|
|
|
|
FIX = Path(__file__).parent / "fixtures"
|
|
BOOT = ["ESP-ROM:esp32s3-20210327", "rst:0x1 (POWERON),boot:0x8 (SPI_FAST_FLASH_BOOT)",
|
|
"I (1299) app_init: Project name: kota-sensorboard",
|
|
"I (1317) app_init: ELF file SHA256: 45fa9d5b9..."]
|
|
|
|
|
|
class Clock:
|
|
def __init__(self):
|
|
self.t = 1000.0
|
|
|
|
def __call__(self):
|
|
return self.t
|
|
|
|
|
|
@pytest.fixture
|
|
def rig():
|
|
engine, clock, actions, log = DiagnosticsEngine(), Clock(), [], []
|
|
runner = GuideRunner(engine, on_action=lambda a: actions.append(a) or True,
|
|
on_log=log.append, clock=clock)
|
|
|
|
def board(*lines):
|
|
for ln in lines:
|
|
parsed = parse_line(ln)
|
|
engine.feed(parsed)
|
|
runner.feed(parsed)
|
|
|
|
return engine, runner, clock, actions, board, log
|
|
|
|
|
|
def _bench(engine):
|
|
for ln in read_log_file(FIX / "live_boot_A4CB8FC946F8_fw1.0.txt"):
|
|
engine.feed(ln)
|
|
|
|
|
|
def test_all_guides_load_and_validate():
|
|
guides = load_guides()
|
|
assert "quick" in guides and len(guides) >= 8
|
|
led = guides["quick"].steps["led"]
|
|
assert set(led.route) == set(LED_ANSWERS) # every colour has a route
|
|
|
|
|
|
def test_recommendations_follow_findings(rig):
|
|
engine, *_ = rig
|
|
_bench(engine)
|
|
ids = [g.id for g in recommended(load_guides(), engine)]
|
|
assert ids[0] == "quick"
|
|
assert ids.index("network") < ids.index("clock") # root order preserved
|
|
assert "mic" in ids
|
|
|
|
|
|
def test_purple_uncommissioned_box_skips_to_pairing_and_follows_it_live(rig):
|
|
engine, runner, clock, _, board, log = rig
|
|
_bench(engine) # log: not commissioned, LED purple
|
|
runner.start("quick")
|
|
runner.answer("PURPLE")
|
|
# "Was it paired before?" is skipped: the log already shows it is not.
|
|
assert runner.step.id == "pair_intro"
|
|
runner.proceed()
|
|
assert runner.view().waiting and runner.step.id == "pair_start"
|
|
board("I (90000) MAIN: Commissioning session started")
|
|
assert runner.step.id == "pair_wifi"
|
|
board("I (95000) MAIN: Commissioning complete")
|
|
assert runner.step.id == "pair_time"
|
|
board("I (99000) SNTP: Time synchronized: Sat Sep 26 16:00:00 2026")
|
|
assert runner.step.id == "pair_upload"
|
|
board("I (120000) INFLUX: HTTP request performed, status code: 204")
|
|
assert runner.end.id == "online" and runner.end.severity == "ok"
|
|
assert any("Paired" in t or "paired" in t for t in engine.technician)
|
|
assert "matter.not_commissioned" not in {rc.finding.rule_id for rc in engine.report().roots}
|
|
assert any(l.startswith("technician:") for l in log)
|
|
|
|
|
|
def test_pairing_timeout_asks_what_the_app_shows(rig):
|
|
engine, runner, clock, _, board, _ = rig
|
|
_bench(engine)
|
|
runner.start("network")
|
|
runner.proceed() # pair_intro -> pair_start
|
|
clock.t += 301
|
|
runner.tick()
|
|
assert runner.step.id == "not_found"
|
|
runner.answer("1")
|
|
assert runner.end.id == "ble_not_found"
|
|
f = engine.findings["tech.network.ble_not_found"]
|
|
assert f.severity == Severity.ERROR and f.active
|
|
|
|
|
|
def test_led_mismatch_is_recorded(rig):
|
|
engine, runner, *_ = rig
|
|
_bench(engine) # log implies PURPLE
|
|
runner.start("quick")
|
|
runner.answer("GREEN")
|
|
f = engine.findings["tech.led_mismatch"]
|
|
assert "Green" in f.title and "Purple" in f.title
|
|
assert runner.step.id == "heartbeat"
|
|
|
|
|
|
def test_led_match_records_nothing(rig):
|
|
engine, runner, *_ = rig
|
|
_bench(engine)
|
|
runner.start("quick")
|
|
runner.answer("PURPLE")
|
|
assert "tech.led_mismatch" not in engine.findings
|
|
|
|
|
|
def test_blue_reseat_fixed_after_power_cycle(rig):
|
|
engine, runner, clock, _, board, _ = rig
|
|
board(*BOOT, "E (3511) MAIN: Failed to initialize RTC: ESP_ERR_TIMEOUT")
|
|
runner.start("quick")
|
|
runner.answer("BLUE")
|
|
runner.proceed()
|
|
runner.proceed() # -> verify (technician power-cycles)
|
|
assert runner.step.id == "verify"
|
|
board(*BOOT, "I (14800) temperature_sensor: Range [20°C ~ 100°C], error < 2°C")
|
|
assert runner.end.id == "i2c_fixed"
|
|
|
|
|
|
def test_blue_still_failing_then_module(rig):
|
|
engine, runner, clock, _, board, _ = rig
|
|
board(*BOOT, "E (3511) MAIN: Failed to initialize RTC: ESP_ERR_TIMEOUT")
|
|
runner.start("rtc_i2c")
|
|
runner.proceed()
|
|
runner.proceed()
|
|
board(*BOOT, "E (3511) MAIN: Failed to initialize RTC: ESP_ERR_TIMEOUT")
|
|
assert runner.step.id == "damaged"
|
|
runner.answer("1")
|
|
assert runner.end.id == "rtc_module"
|
|
|
|
|
|
def test_white_skips_portal_question_on_v1_firmware(rig):
|
|
engine, runner, clock, _, board, _ = rig
|
|
board(*BOOT) # known v1.0: no setup portal
|
|
runner.start("white")
|
|
runner.answer("1") # nobody held the button
|
|
assert runner.step.id == "sd_reseat"
|
|
|
|
|
|
def test_white_asks_portal_question_on_unknown_firmware(rig):
|
|
engine, runner, clock, _, board, _ = rig
|
|
board("ESP-ROM:esp32s3-20210327", "I (1317) app_init: ELF file SHA256: 0123456789...")
|
|
runner.start("white")
|
|
runner.answer("1")
|
|
assert runner.step.id == "portal"
|
|
|
|
|
|
def test_red_routes_to_failing_sensor_and_names_it(rig):
|
|
engine, runner, clock, _, board, _ = rig
|
|
board(*BOOT, "E (14000) MAIN: Failed to initialize SEN66: ESP_ERR_TIMEOUT")
|
|
runner.start("sensors")
|
|
assert runner.step.id == "reseat"
|
|
assert "SEN66" in runner.view().text
|
|
runner.proceed()
|
|
board(*BOOT, "E (14000) MAIN: Failed to initialize SEN66: ESP_ERR_TIMEOUT")
|
|
assert runner.end.id == "sensor_replace"
|
|
|
|
|
|
def test_reboot_loop_resets_board_then_routes_by_cause(rig):
|
|
engine, runner, clock, actions, board, _ = rig
|
|
runner.start("reboot_loop")
|
|
assert actions == ["reset"]
|
|
board(*BOOT, "E (1519) SD_CARD: Failed to initialize the card (ESP_ERR_TIMEOUT). x")
|
|
clock.t += 46
|
|
runner.tick()
|
|
assert runner.guide.id == "white"
|
|
|
|
|
|
def test_reset_action_failure_is_shown(rig):
|
|
engine = DiagnosticsEngine()
|
|
runner = GuideRunner(engine, on_action=lambda a: False, clock=Clock())
|
|
runner.start("reboot_loop")
|
|
assert "connect" in runner.view().note.lower()
|
|
|
|
|
|
def test_mic_disabled_switch_on_verified(rig):
|
|
engine, runner, clock, actions, board, _ = rig
|
|
_bench(engine)
|
|
runner.start("mic")
|
|
assert runner.step.id == "switch"
|
|
runner.answer("0") # set it to ON
|
|
assert actions == ["reset"]
|
|
board(*BOOT, "I (20000) MICROPHONE: Read 300000 bytes in 4936 ms")
|
|
assert runner.end.id == "mic_on"
|
|
|
|
|
|
def test_mic_clap_test_detects_dead_mic(rig):
|
|
engine, runner, clock, _, board, _ = rig
|
|
board(*BOOT, *["I (20036) MICROPHONE: Sum SQ: 0.000000, RMS pressure: 0.000000, "
|
|
"Normalized: 0.000000, dB peak: -inf"] * 2)
|
|
runner.start("mic")
|
|
assert runner.step.id == "clap"
|
|
board("I (30036) MICROPHONE: Sum SQ: 0.000000, RMS pressure: 0.000000, Normalized: 0.000000, dB peak: -inf")
|
|
clock.t += 46
|
|
runner.tick()
|
|
assert runner.end.id == "mic_faulty"
|
|
|
|
|
|
def test_mic_clap_test_passes_on_sound(rig):
|
|
engine, runner, clock, _, board, _ = rig
|
|
runner.start("mic")
|
|
board("I (30036) MICROPHONE: Sum SQ: 12.5, RMS pressure: 0.02, Normalized: 0.1, dB peak: 71.3")
|
|
assert runner.end.id == "mic_ok"
|
|
|
|
|
|
def test_back_never_reenters_a_wait(rig):
|
|
engine, runner, clock, actions, board, _ = rig
|
|
runner.start("sensors") # route -> capture (wait, resets board)
|
|
assert runner.step.id == "capture" and actions == ["reset"]
|
|
clock.t += 46
|
|
runner.tick() # -> which_again -> end:red_unclear
|
|
runner.back()
|
|
assert actions == ["reset"] # no second reset
|
|
|
|
|
|
def test_skip_wait_and_report_contains_technician_checks(rig):
|
|
engine, runner, clock, _, board, _ = rig
|
|
runner.start("update")
|
|
runner.proceed() # technician skips the wait
|
|
assert runner.end.id == "update_slow"
|
|
text = engine.report().to_text()
|
|
assert "Technician checks:" in text and "Update is taking very long" in text
|