test_fault_simulation.py

"""Smoke tests: fault profiles change live PIDs and DTCs on the emulator."""

from __future__ import annotations

import logging
import time

import obd

logging.basicConfig(level=logging.WARNING)

from elm_server import start_elm_emulator, stop_elm_emulator
from fault_simulation import apply_fault, clear_faults
from obd_connection import init_obd_session, obd_session_lock


def _read_sample() -> dict:
    time.sleep(0.2)
    with obd_session_lock:
        conn = init_obd_session()
        rpm = conn.query(obd.commands.RPM)
        stft = conn.query(obd.commands.SHORT_FUEL_TRIM_1)
        coolant = conn.query(obd.commands.COOLANT_TEMP)
        dtcs = conn.query(obd.commands.GET_DTC)
        dtc_list: list[str] = []
        if not dtcs.is_null() and dtcs.value:
            for entry in dtcs.value:
                code = entry[0] if isinstance(entry, tuple) else entry
                dtc_list.append(str(code))
        return {
            "rpm": None if rpm.is_null() else rpm.value.magnitude,
            "stft": None if stft.is_null() else float(stft.value.magnitude),
            "coolant": None if coolant.is_null() else float(coolant.value.magnitude),
            "dtcs": dtc_list,
        }


def test_vacuum_leak_changes_pids() -> None:
    clear_faults()
    healthy = _read_sample()
    apply_fault("vacuum_leak")
    faulted = _read_sample()
    assert faulted["rpm"] is not None and faulted["rpm"] < 1000, faulted
    assert faulted["stft"] is not None and faulted["stft"] > 15, faulted
    assert healthy != faulted, (healthy, faulted)


def test_vacuum_leak_sets_dtc() -> None:
    apply_fault("vacuum_leak")
    faulted = _read_sample()
    assert any("P0171" in d for d in faulted["dtcs"]), faulted


def test_thermostat_stuck_closed() -> None:
    apply_fault("thermostat_stuck_closed")
    faulted = _read_sample()
    assert faulted["coolant"] is not None and faulted["coolant"] > 100, faulted


def test_clear_restores_healthy() -> None:
    apply_fault("vacuum_leak")
    clear_faults()
    cleared = _read_sample()
    assert cleared["dtcs"] == [], cleared


if __name__ == "__main__":
    try:
        start_elm_emulator()
        time.sleep(0.5)
        test_vacuum_leak_changes_pids()
        print("OK vacuum_leak PIDs")
        test_vacuum_leak_sets_dtc()
        print("OK vacuum_leak DTC")
        test_thermostat_stuck_closed()
        print("OK thermostat_stuck_closed")
        test_clear_restores_healthy()
        print("OK clear_faults")
    finally:
        stop_elm_emulator()