444 lines
20 KiB
Python
444 lines
20 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)
|
||
|
||
# ── 8. Экстремальные тесты ──────────────────────────────────────
|
||
print("\n═══ 8. Экстремальные тесты ═══")
|
||
|
||
# SQL-инъекция через decoded
|
||
try:
|
||
r = parse_batch([{"cmd": "0105", "raw": "", "decoded": "ОЖ: 83'; DROP TABLE sessions; --"}])
|
||
test("SQL-инъекция в decoded — не падает", not r.get("error"))
|
||
except Exception:
|
||
test("SQL-инъекция в decoded — не падает", False)
|
||
|
||
# Бинарный мусор в raw
|
||
r = parse_batch([{"cmd": "0902", "raw": "\x00\x01\x02\xFF\xFE\xFD", "decoded": ""}])
|
||
test("бинарный мусор в raw", r["vin"] is None and not r.get("error"))
|
||
|
||
# VIN — слишком короткий
|
||
r = parse_batch([{"cmd": "0902", "raw": "", "decoded": "VIN: SHORT"}])
|
||
test("VIN короткий → null", r["vin"] is None)
|
||
|
||
# VIN — слишком длинный (18 символов)
|
||
r = parse_batch([{"cmd": "0902", "raw": "", "decoded": "VIN: WVWZZZ1KZAW1234567"}])
|
||
test("VIN 18 символов → null", r["vin"] is None)
|
||
|
||
# VIN с русскими буквами
|
||
r = parse_batch([{"cmd": "0902", "raw": "", "decoded": "VIN: ПРИВЕТЭТОТЕСТ"}])
|
||
test("VIN кириллица → null", r["vin"] is None)
|
||
|
||
# Пустой decoded, пустой raw
|
||
r = parse_batch([{"cmd": "0105", "raw": "", "decoded": ""}])
|
||
test("пустой ответ на PID", len(r["parameters"]) == 0 and not r.get("error"))
|
||
|
||
# Ответ начинается с SEARCHING
|
||
r = parse_batch([{"cmd": "0105", "raw": "SEARCHING...", "decoded": ""}])
|
||
test("SEARCHING → без параметров", len(r["parameters"]) == 0)
|
||
|
||
# BUS ERROR
|
||
r = parse_batch([{"cmd": "0105", "raw": "BUS ERROR", "decoded": ""}])
|
||
test("BUS ERROR → без паники", not r.get("error"))
|
||
|
||
# DTC: none (нет ошибок)
|
||
r = parse_batch([{"cmd": "03", "raw": "", "decoded": "DTC stored: none"}])
|
||
test("DTC none → пустой список", len(r["dtc_stored"]) == 0)
|
||
|
||
# DTC: 20 штук
|
||
r = parse_batch([{"cmd": "03", "raw": "", "decoded": "DTC stored: " + " ".join([f"P{str(i).zfill(4)}" for i in range(1, 21)])}])
|
||
test("DTC 20 штук", len(r["dtc_stored"]) == 20)
|
||
|
||
# Номер протокола как ответ
|
||
r = parse_batch([{"cmd": "0105", "raw": "OK", "decoded": ""}])
|
||
test("OK → без параметров", len(r["parameters"]) == 0)
|
||
|
||
# Много пробелов и разных разделителей
|
||
r = parse_batch([{"cmd": "03", "raw": " 43 02 01 00 02 00 ", "decoded": ""}])
|
||
test("DTC с лишними пробелами", len(r["dtc_stored"]) > 0)
|
||
|
||
# Ответ от ELM с кавычками
|
||
try:
|
||
r = parse_batch([{"cmd": "0105", "raw": "", "decoded": 'ОЖ: 83"С\'тест'}])
|
||
test("кавычки в decoded — не падает", True)
|
||
except Exception:
|
||
test("кавычки в decoded — не падает", False)
|
||
|
||
# В ответе только цифры
|
||
r = parse_batch([{"cmd": "0105", "raw": "1234567890", "decoded": ""}])
|
||
test("только цифры — не распознано", not r.get("error"))
|
||
|
||
# Длинная строка (10000 символов)
|
||
r = parse_batch([{"cmd": "03", "raw": "SEARCHING" + "A" * 9990, "decoded": ""}])
|
||
test("длинный ответ SEARCHING", not r.get("error"))
|
||
|
||
# ── 9. DB экстремальные ────────────────────────────────────────
|
||
print("\n═══ 9. DB экстремальные ═══")
|
||
|
||
db2 = Database(TEST_DB)
|
||
|
||
# Ультра-длинный диагноз
|
||
long_diag = "тест " * 10_000
|
||
db2.save_session(
|
||
client_info={"phone_model": "x", "device_uuid": "long-test"},
|
||
responses=[{"cmd": "0105", "raw": "", "decoded": "ОЖ: 83"}],
|
||
diagnosis=long_diag,
|
||
)
|
||
test("диагноз 50k символов", True)
|
||
|
||
# Пустой diagnosis
|
||
db2.save_session(
|
||
client_info={"phone_model": "x", "device_uuid": "empty-diag"},
|
||
responses=[],
|
||
diagnosis="",
|
||
)
|
||
test("пустой диагноз", True)
|
||
|
||
# Client_info с неожиданными типами
|
||
db2.save_session(
|
||
client_info={
|
||
"phone_model": None, "phone_maker": 12345, "android_version": "",
|
||
"android_sdk": None, "app_version": "0", "android_id": None,
|
||
"device_uuid": "null-test", "elm_mac": None,
|
||
},
|
||
responses=[{"cmd": "0105", "raw": "", "decoded": "тест"}],
|
||
)
|
||
test("мусор в client_info — None/числа", True)
|
||
|
||
# 10 сессий с разными request_id для проверки идемпотентности
|
||
for i in range(10):
|
||
db2.save_session(
|
||
client_info={"phone_model": f"device_{i}", "device_uuid": f"uuid_{i}"},
|
||
responses=[{"cmd": "0105", "raw": "", "decoded": "ОЖ: 83"}],
|
||
request_id=f"req-{i}",
|
||
response_json={"ok": i},
|
||
)
|
||
test("10 сессий с разными request_id", True)
|
||
|
||
# Повторный request_id — не должен создать дубликат
|
||
count_before = len(db2.get_recent_sessions(100))
|
||
db2.save_session(
|
||
client_info={"phone_model": "duplicate_test", "device_uuid": "dup"},
|
||
responses=[],
|
||
request_id="req-0",
|
||
response_json={"ok": "duplicate"},
|
||
)
|
||
count_after = len(db2.get_recent_sessions(100))
|
||
test("идемпотентность — дубликат не создан", count_before == count_after)
|
||
|
||
# Кэш по request_id возвращает старый результат
|
||
cached = db2.get_cached_response("req-0")
|
||
test("идемпотентность — кэш вернул старые данные", cached is not None and cached.get("ok") == 0)
|
||
|
||
# Конкурентный доступ (симуляция)
|
||
import threading
|
||
errors = []
|
||
def concurrent_write(idx: int):
|
||
try:
|
||
db2.save_session(
|
||
client_info={"phone_model": f"concurrent_{idx}", "device_uuid": f"cuuid_{idx}"},
|
||
responses=[],
|
||
request_id=f"creq-{idx}",
|
||
)
|
||
except Exception as e:
|
||
errors.append(str(e))
|
||
|
||
threads = [threading.Thread(target=concurrent_write, args=(i,)) for i in range(20)]
|
||
[t.start() for t in threads]
|
||
[t.join() for t in threads]
|
||
test(f"20 конкурентных записей — максимум 3 ошибки", len(errors) <= 3)
|
||
|
||
db2.close()
|
||
if TEST_DB.exists():
|
||
TEST_DB.unlink()
|
||
|
||
# ── 10. DTC словарь — граничные случаи ──────────────────────────
|
||
print("\n═══ 10. DTC словарь — граничные ═══")
|
||
dtc = _load_dtc_dict()
|
||
|
||
# Код, которого нет в словаре
|
||
test("неизвестный код → сам код", dtc.get("P9999", "P9999") == "P9999")
|
||
test("код в нижнем регистре → нет", dtc.get("p0301", "") == "")
|
||
|
||
# Пустой код
|
||
test("пустой код → пусто", dtc.get("", None) is None)
|
||
|
||
# Не-P код (C-код)
|
||
test("C-код C0000 есть", "C0000" in dtc)
|
||
|
||
# Не-P код (B-код, U-код)
|
||
test("B-код B0000 есть", "B0000" in dtc)
|
||
test("U-код U0000 есть", "U0000" in dtc)
|
||
|
||
# ── 11. Промпт-билдер — граничные ───────────────────────────────
|
||
print("\n═══ 11. Промпт-билдер граничные ═══")
|
||
|
||
# Пустой data
|
||
try:
|
||
_build_diagnosis_prompt({"vin": None, "dtc_stored": [], "dtc_pending": [], "parameters": [], "raw_log": []})
|
||
test("пустой словарь в промпт", True)
|
||
except Exception:
|
||
test("пустой словарь в промпт", False)
|
||
|
||
# data без ключей
|
||
try:
|
||
_build_diagnosis_prompt({})
|
||
test("пустой {} в промпт", True)
|
||
except Exception:
|
||
test("пустой {} в промпт", False)
|
||
|
||
# None вместо списков
|
||
try:
|
||
_build_diagnosis_prompt({"vin": None, "dtc_stored": None, "dtc_pending": None, "parameters": None, "raw_log": None})
|
||
test("None вместо списков", True)
|
||
except Exception:
|
||
test("None вместо списков", False)
|
||
|
||
# Очень длинный car_info
|
||
p = _build_diagnosis_prompt(
|
||
{"vin": None, "dtc_stored": [], "dtc_pending": [], "parameters": [], "raw_log": []},
|
||
car_info="A" * 5000,
|
||
)
|
||
test("car_info 5000 символов", "A" in p)
|
||
|
||
# ── 12. ELM — мусор, разрывы, протоколы ────────────────────────
|
||
print("\n═══ 12. ELM — мусор, разрывы, протоколы ═══")
|
||
|
||
# HEX с символами кадра CAN (0: 1: и т.д.)
|
||
r = parse_batch([{"cmd": "0902", "raw": "0: 49 02 01 57 56\n1: 57 5A 5A 5A 31 4B 5A\n2: 41 57 31 32 33 34 35 36", "decoded": ""}])
|
||
test("CAN multi-frame с 0: 1:", r["vin"] == "WVWZZZ1KZAW123456")
|
||
|
||
# ELM ответил SEARCHING, потом данные, потом PROMPT
|
||
r = parse_batch([{"cmd": "010C", "raw": "SEARCHING\nSEARCHING\n41 0C 0C A8", "decoded": ""}])
|
||
test("SEARCHING перед данными — PID", "RPM" not in str(r.get("decoded", "")))
|
||
|
||
# Заголовок ISO-TP (длина)
|
||
r = parse_batch([{"cmd": "0902", "raw": "10 14 49 02 01 57 56\n21 57 5A 5A 5A 31 4B\n22 5A 41 57 31 32 33 34 35 36", "decoded": ""}])
|
||
test("ISO-TP заголовки 10/21/22", r["vin"] == "WVWZZZ1KZAW123456")
|
||
|
||
# ELM вернул UDP-подобный формат
|
||
r = parse_batch([{"cmd": "03", "raw": "43 02 01 00 02 00 03 00 04 00", "decoded": ""}])
|
||
test("много DTC в одном ответе", len(r["dtc_stored"]) >= 3)
|
||
|
||
# Дефолтный ответ ELM (пробелы, переводы)
|
||
r = parse_batch([{"cmd": "0105", "raw": "\n\n\n 41 05 47 \n\n", "decoded": ""}])
|
||
test("ELM с лишними переводами", not r.get("error"))
|
||
|
||
# CAN bus error
|
||
r = parse_batch([{"cmd": "0105", "raw": "CAN ERROR", "decoded": ""}])
|
||
test("CAN ERROR", not r.get("error"))
|
||
|
||
# Нет ответа — пустая строка
|
||
r = parse_batch([{"cmd": "0105", "raw": "", "decoded": ""}])
|
||
test("нет ответа — пусто", not r.get("error") and r["parameters"] == [])
|
||
|
||
# Garbage в raw (не HEX, не ELM)
|
||
r = parse_batch([{"cmd": "0105", "raw": "Lorem ipsum dolor sit amet, consectetur adipiscing elit. Sed do eiusmod tempor incididunt ut labore et dolore magna aliqua.", "decoded": ""}])
|
||
test("Lorem Ipsum вместо ответа", len(r["parameters"]) == 0)
|
||
|
||
# VIN с пробелами в HEX
|
||
r = parse_batch([{"cmd": "0902", "raw": "4902015756 575A 5A5A 314B5A 4157 3132 3334 3536", "decoded": ""}])
|
||
test("VIN HEX с группами по 2-4 байта", r["vin"] == "WVWZZZ1KZAW123456")
|
||
|
||
# PID с отрицательным значением
|
||
r = parse_batch([{"cmd": "0111", "raw": "", "decoded": "Дроссель: 0.0 %"}])
|
||
test("дроссель 0%", len(r["parameters"]) == 1)
|
||
|
||
# PID с большим значением
|
||
r = parse_batch([{"cmd": "0105", "raw": "", "decoded": "ОЖ: 127 °C"}])
|
||
test("ОЖ 127°C", len(r["parameters"]) == 1)
|
||
|
||
# Ответ содержит именованный PID с ':'
|
||
r = parse_batch([{"cmd": "010B", "raw": "", "decoded": "MAP: 101 кПа"}])
|
||
test("MAP parsed", len(r["parameters"]) == 1)
|
||
|
||
# ACK ответ
|
||
r = parse_batch([{"cmd": "ATSP0", "raw": "OK", "decoded": "OK"}])
|
||
test("AT OK — не данные", r["parameters"] == [])
|