199 lines
8.8 KiB
Python
199 lines
8.8 KiB
Python
"""
|
|
Сквозные тесты elmAI — без LLM.
|
|
Проверяет: эндпоинты, БД, идемпотентность, DTC, chat, ping, script.
|
|
"""
|
|
import json
|
|
import os
|
|
import sqlite3
|
|
import sys
|
|
import tempfile
|
|
from pathlib import Path
|
|
|
|
# Добавляем корень проекта в путь
|
|
sys.path.insert(0, str(Path(__file__).parent))
|
|
|
|
# Временная БД для тестов
|
|
TEST_DB = Path(tempfile.gettempdir()) / f"elmer_test_{os.getpid()}.db"
|
|
os.environ["ELMER_CONFIG"] = str(Path(__file__).parent / "config.yaml")
|
|
|
|
from api.db import Database
|
|
from api.parser import parse_batch, format_no_llm
|
|
from api.routes import _load_dtc_dict, _build_diagnosis_prompt
|
|
from api.scripts import build_default_script, build_full_script
|
|
|
|
passed = 0
|
|
failed = 0
|
|
|
|
def test(name: str, ok: bool, detail: str = ""):
|
|
global passed, failed
|
|
if ok:
|
|
passed += 1
|
|
print(f" ✅ {name}")
|
|
else:
|
|
failed += 1
|
|
print(f" ❌ {name}: {detail}")
|
|
|
|
# ── 1. Скрипты ──────────────────────────────────────────────────
|
|
print("\n═══ 1. Сборка скриптов ═══")
|
|
s = build_default_script()
|
|
test("default_script — есть steps", "steps" in s)
|
|
test("default_script — 1+ шагов", len(s.get("steps", [])) >= 1)
|
|
|
|
s = build_full_script()
|
|
test("full_script — есть steps", "steps" in s)
|
|
test("full_script — 2+ шагов", len(s.get("steps", [])) >= 2)
|
|
|
|
# ── 2. Парсер ELM-ответов ──────────────────────────────────────
|
|
print("\n═══ 2. Парсер ответов ═══")
|
|
|
|
# VIN из decoded
|
|
r = parse_batch([{"cmd": "0902", "raw": "", "decoded": "VIN: WVWZZZ1KZAW123456"}])
|
|
test("VIN из decoded", r["vin"] == "WVWZZZ1KZAW123456")
|
|
|
|
# VIN из raw HEX (полный 17-символьный)
|
|
r = parse_batch([{"cmd": "0902", "raw": "490201 57 56 57 5A 5A 5A 31 4B 5A 41 57 31 32 33 34 35 36", "decoded": ""}])
|
|
test("VIN из raw HEX", r["vin"] == "WVWZZZ1KZAW123456")
|
|
|
|
# DTC stored
|
|
r = parse_batch([{"cmd": "03", "raw": "", "decoded": "DTC stored: P0301 P0302"}])
|
|
test("DTC stored", r["dtc_stored"] == ["P0301", "P0302"])
|
|
|
|
# DTC pending
|
|
r = parse_batch([{"cmd": "07", "raw": "", "decoded": "DTC pending: P0302"}])
|
|
test("DTC pending", r["dtc_pending"] == ["P0302"])
|
|
|
|
# PID
|
|
r = parse_batch([{"cmd": "0105", "raw": "", "decoded": "ОЖ: 83 °C"}])
|
|
test("PID разобран", len(r["parameters"]) == 1 and r["parameters"][0]["name"] == "ОЖ")
|
|
|
|
# Пустой батч — нет данных
|
|
r = parse_batch([])
|
|
test("пустой батч", r["vin"] is None and r["dtc_stored"] == [] and r["dtc_pending"] == [])
|
|
|
|
# DTC из raw HEX (mode 03, ответ начинается с 43)
|
|
r = parse_batch([{"cmd": "03", "raw": "43 02 01 00 02 00", "decoded": ""}])
|
|
test("DTC из raw HEX", len(r["dtc_stored"]) > 0)
|
|
|
|
# DTC из raw HEX mode 07 (byte count 01, DTC P0100)
|
|
r = parse_batch([{"cmd": "07", "raw": "47 01 01 00", "decoded": ""}])
|
|
test("DTC pending из raw HEX", len(r["dtc_pending"]) > 0)
|
|
|
|
# Mixed: данные + команда без decoded
|
|
r = parse_batch([
|
|
{"cmd": "010C", "raw": "410C 0C A8", "decoded": "RPM: 948.0 RPM"},
|
|
{"cmd": "0902", "raw": "", "decoded": "VIN: WVWZZZ1KZAW123456"},
|
|
])
|
|
test("mixed: VIN + PID", r["vin"] == "WVWZZZ1KZAW123456" and len(r["parameters"]) >= 1)
|
|
|
|
# ── 3. Промпт билдер ────────────────────────────────────────────
|
|
print("\n═══ 3. Промпт билдер ═══")
|
|
data = {
|
|
"vin": "WLL0333272859",
|
|
"dtc_stored": ["P0301"],
|
|
"dtc_pending": [],
|
|
"parameters": [{"name": "RPM", "value": "948"}],
|
|
"raw_log": ["→ 010C\n← RPM: 948"],
|
|
}
|
|
p = _build_diagnosis_prompt(data)
|
|
test("промпт содержит VIN", "WLL0333272859" in p)
|
|
test("промпт содержит ошибку", "P0301" in p)
|
|
test("промпт содержит параметр", "RPM" in p)
|
|
test("промпт содержит запрос на анализ", "краткий" in p or "диагноз" in p)
|
|
|
|
p2 = _build_diagnosis_prompt(data, car_info="Volkswagen Passat 1.8T 2005")
|
|
test("car_info в промпте", "Volkswagen" in p2)
|
|
|
|
# Без данных
|
|
p3 = _build_diagnosis_prompt({"vin": None, "dtc_stored": [], "dtc_pending": [], "parameters": [], "raw_log": []})
|
|
test("промпт без данных содержит 'не распознаны'", "не распознаны" in p3)
|
|
|
|
# ── 4. DTC справочник ───────────────────────────────────────────
|
|
print("\n═══ 4. DTC декодер ═══")
|
|
dtc = _load_dtc_dict()
|
|
test("DTC словарь загружен", len(dtc) > 50)
|
|
test("P0301 есть", "P0301" in dtc)
|
|
test("P0301 описание", "цилиндр" in dtc.get("P0301", "").lower())
|
|
test("P0420 есть", "P0420" in dtc)
|
|
|
|
# ── 5. База данных ─────────────────────────────────────────────
|
|
print("\n═══ 5. База данных ═══")
|
|
db_path = TEST_DB
|
|
db = Database(db_path)
|
|
test("БД создана", db_path.exists())
|
|
|
|
# Сохраняем сессию
|
|
db.save_session(
|
|
client_info={"phone_model": "Pixel", "android_id": "test123", "device_uuid": "uuid-1", "elm_mac": "00:11:22:33:44:55"},
|
|
responses=[{"cmd": "0105", "raw": "", "decoded": "ОЖ: 83"}],
|
|
diagnosis="Тестовый диагноз",
|
|
llm_model="deepseek-v4-flash",
|
|
llm_success=True,
|
|
request_id="req-001",
|
|
response_json={"diagnosis": "Тестовый диагноз"},
|
|
)
|
|
sessions = db.get_recent_sessions(1)
|
|
test("сессия сохранена", len(sessions) == 1)
|
|
test("diagnosis_text сохранён", sessions[0]["diagnosis_text"] == "Тестовый диагноз")
|
|
test("llm_success", sessions[0]["llm_success"] == 1)
|
|
test("request_id сохранён", sessions[0]["request_id"] == "req-001")
|
|
test("device_uuid сохранён", sessions[0]["device_uuid"] == "uuid-1")
|
|
test("elm_mac сохранён", sessions[0]["elm_mac"] == "00:11:22:33:44:55")
|
|
|
|
# Идемпотентность — повторный request_id
|
|
cached = db.get_cached_response("req-001")
|
|
test("кэш по request_id работает", cached is not None and cached.get("diagnosis") == "Тестовый диагноз")
|
|
|
|
# Несуществующий request_id
|
|
cached = db.get_cached_response("nonexistent")
|
|
test("несуществующий request_id → None", cached is None)
|
|
|
|
# Сессия без request_id
|
|
db.save_session(
|
|
client_info={"phone_model": "Pixel", "android_id": "test456", "device_uuid": "uuid-2"},
|
|
responses=[{"cmd": "010C", "raw": "", "decoded": "RPM: 948"}],
|
|
)
|
|
sessions = db.get_recent_sessions(5)
|
|
test("сессия без request_id", len(sessions) >= 2)
|
|
|
|
# DTC scan
|
|
db.save_dtc_scan(
|
|
client_info={"phone_model": "Pixel", "device_uuid": "uuid-3", "elm_mac": "00:11:22:33:44:66"},
|
|
dtc_codes=["P0301", "P0302"],
|
|
)
|
|
test("DTC scan сохранён в sessions", db.get_recent_sessions(10)[0]["response_count"] == 0)
|
|
|
|
# Валидация колонок
|
|
col = [r[1] for r in db.conn.execute("PRAGMA table_info(sessions)").fetchall()]
|
|
for c in ["device_uuid", "phone_lang", "phone_tz", "phone_display", "request_id", "response_json"]:
|
|
test(f"колонка {c} существует", c in col)
|
|
|
|
db.close()
|
|
# Чистим
|
|
if db_path.exists():
|
|
db_path.unlink()
|
|
|
|
# ── 6. Format no LLM ────────────────────────────────────────────
|
|
print("\n═══ 6. Format no LLM ═══")
|
|
f = format_no_llm({"vin": "VIN123", "dtc_stored": ["P0301"], "dtc_pending": [], "parameters": [], "raw_log": []})
|
|
test("format_no_llm содержит VIN", "VIN123" in f)
|
|
|
|
f = format_no_llm({"vin": None, "dtc_stored": [], "dtc_pending": [], "parameters": [], "raw_log": []})
|
|
test("format_no_llm без данных", len(f) > 0)
|
|
|
|
# ── 7. Скрипты через API (без HTTP) ─────────────────────────────
|
|
print("\n═══ 7. Скрипты ═══")
|
|
s = build_default_script()
|
|
test("default в JSON", isinstance(s, dict))
|
|
test("version", s.get("version") == 1)
|
|
|
|
s2 = build_full_script()
|
|
test("full", s2.get("version") == 1)
|
|
|
|
# ── Итог ─────────────────────────────────────────────────────────
|
|
print(f"\n═══ РЕЗУЛЬТАТ: {passed} пройдено, {failed} упало ═══")
|
|
if failed:
|
|
print("ЕСТЬ ОШИБКИ!")
|
|
sys.exit(1)
|
|
else:
|
|
print("ВСЁ ОК!")
|