607 lines
26 KiB
Python
607 lines
26 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 _build_diagnosis_prompt
|
||
from api.dtc import _load_dtc_dict
|
||
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"] == [])
|
||
|
||
# ── 13. Идемпотентность ─────────────────────────────────────────
|
||
print("\n═══ 13. Идемпотентность ═══")
|
||
|
||
db3 = Database(TEST_DB)
|
||
|
||
# Симуляция: DTC upload с request_id 2 раза
|
||
dtc_req_id = "dtc-test-uuid-001"
|
||
dtc_first = {"codes": [{"code": "P0301", "desc": "Пропуски зажигания"}], "count": 1}
|
||
|
||
# Сохраняем первый раз (как сделал бы /dtc/upload)
|
||
db3.save_session(
|
||
client_info={"device_uuid": "idemp-test"},
|
||
responses=[{"decoded": "DTC stored: P0301"}],
|
||
request_id=dtc_req_id,
|
||
response_json={"codes": dtc_first["codes"], "count": dtc_first["count"]},
|
||
)
|
||
test("DTC: первый save с request_id", True)
|
||
|
||
# Повтор — не должен создать дубликат
|
||
count_before = len(db3.get_recent_sessions(100))
|
||
db3.save_session(
|
||
client_info={"device_uuid": "idemp-test"},
|
||
responses=[{"decoded": "DTC stored: P0301"}],
|
||
request_id=dtc_req_id,
|
||
response_json={"codes": [{"code": "P0999", "desc": "НИКОГДА"}], "count": 0}, # другой ответ
|
||
)
|
||
count_after = len(db3.get_recent_sessions(100))
|
||
test("DTC: дубликат не создан", count_before == count_after)
|
||
|
||
# Кэш возвращает ПЕРВЫЙ результат, не последний
|
||
cached = db3.get_cached_response(dtc_req_id)
|
||
test("DTC: кэш вернул оригинал", cached is not None and cached["codes"][0]["code"] == "P0301")
|
||
|
||
# Upload сессии с разными request_id — не конфликтуют
|
||
for i in range(5):
|
||
db3.save_session(
|
||
client_info={"device_uuid": f"multi-idemp-{i}"},
|
||
responses=[],
|
||
request_id=f"multi-req-{i}",
|
||
response_json={"i": i},
|
||
)
|
||
test("5 разных request_id — все сохранены",
|
||
len(db3.get_recent_sessions(10)) == count_after + 5)
|
||
|
||
db3.close()
|
||
if TEST_DB.exists():
|
||
TEST_DB.unlink()
|
||
|
||
# ── 14. DTC модуль (api/dtc.py) ─────────────────────────────────
|
||
print("\n═══ 14. DTC модуль ═══")
|
||
|
||
# Загрузка словаря из dtc.py
|
||
dtc = _load_dtc_dict()
|
||
test("dtc загружен из dtc.py", len(dtc) > 100)
|
||
|
||
# Известные коды
|
||
test("P0301 = пропуски", "Пропуски" in dtc.get("P0301", ""))
|
||
test("P0420 = катализатор", "катализатор" in dtc.get("P0420", "").lower())
|
||
|
||
# Протокольные коды
|
||
test("ABS C0000", "ABS" in dtc.get("C0000", ""))
|
||
test("шина U0000", "шина" in dtc.get("U0000", "").lower())
|
||
test("кузов B0000", "кузов" in dtc.get("B0000", "").lower())
|
||
|
||
# ── 15. Ping модуль (api/ping.py) ────────────────────────────────
|
||
print("\n═══ 15. Ping модуль ═══")
|
||
|
||
from api.ping import _ping_llm_cache
|
||
|
||
# Изначально кэш пуст
|
||
test("ping кэш пуст изначально", _ping_llm_cache == {})
|
||
|
||
# ── 16. State модуль (obd/state.py) ─────────────────────────────
|
||
print("\n═══ 16. State ELM ═══")
|
||
|
||
from obd.state import State, Rsp
|
||
|
||
# State enum
|
||
test("State.UNDEFINED", State.UNDEFINED is not None)
|
||
test("State.READY", State.READY is not None)
|
||
test("State.ERROR", State.ERROR is not None)
|
||
|
||
# Rsp.identify
|
||
test("Rsp.identify OK", Rsp.identify("OK") == Rsp.OK)
|
||
test("Rsp.identify BUS ERROR", Rsp.identify("BUS ERROR") == Rsp.BUS_ERROR)
|
||
test("Rsp.identify SEARCHING", Rsp.identify("SEARCHING...") == Rsp.SEARCHING)
|
||
test("Rsp.identify NO DATA", Rsp.identify("NO DATA") == Rsp.NODATA)
|
||
test("Rsp.identify CAN ERROR", Rsp.identify("CAN ERROR") == Rsp.CAN_ERROR)
|
||
test("Rsp.identify BUFFER FULL", Rsp.identify("BUFFER FULL") == Rsp.BUFFER_FULL)
|
||
test("Rsp.identify STOPPED", Rsp.identify("STOPPED") == Rsp.STOPPED)
|
||
test("Rsp.identify UNABLE", Rsp.identify("UNABLE TO CONNECT") == Rsp.UNABLE)
|
||
test("Rsp.identify RX ERROR", Rsp.identify("RX ERROR") == Rsp.RX_ERROR)
|
||
test("Rsp.identify DATA ERROR", Rsp.identify("DATA ERROR") == Rsp.DATA_ERROR)
|
||
|
||
# PROMPT
|
||
test("Rsp.identify PROMPT >", Rsp.identify(">") == Rsp.PROMPT)
|
||
|
||
# UNKNOWN — данные
|
||
test("Rsp.identify UNKNOWN — данные", Rsp.identify("41 0C 0C A8") == Rsp.UNKNOWN)
|
||
test("Rsp.identify UNKNOWN — VIN", Rsp.identify("49 02 01 57 56") == Rsp.UNKNOWN)
|
||
|
||
# Нижний регистр, пробелы
|
||
test("Rsp.identify lower case", Rsp.identify("bus error") == Rsp.BUS_ERROR)
|
||
test("Rsp.identify пробелы", Rsp.identify(" OK ") == Rsp.OK)
|
||
|
||
# ── 17. Timing модуль (obd/timing.py) ────────────────────────────
|
||
print("\n═══ 17. Timing ELM ═══")
|
||
|
||
from obd.timing import AdaptiveTiming
|
||
|
||
t = AdaptiveTiming()
|
||
test("timing DEFAULT", t.ms == 500)
|
||
test("atst = ms/4", t.atst == 125)
|
||
|
||
t.increase()
|
||
test("timing увеличился", t.ms > 500)
|
||
|
||
t.decrease()
|
||
test("timing уменьшился", t.ms >= 500)
|
||
|
||
t.reset()
|
||
test("timing сброс", t.ms == 500)
|
||
|
||
# Многократное увеличение
|
||
for _ in range(100):
|
||
t.increase()
|
||
test("timing MAX не превышен", t.ms <= 2000)
|
||
|
||
# Многократное уменьшение
|
||
for _ in range(100):
|
||
t.decrease()
|
||
test("timing MIN не превышен", t.ms >= 50)
|
||
|
||
t2 = AdaptiveTiming()
|
||
t2.increase()
|
||
t2.increase()
|
||
v = t2.ms
|
||
t2.reset()
|
||
test("reset после increase", t2.ms == 500 and v > 500)
|
||
|
||
atst = t2.atst
|
||
test("atst >= 1", atst >= 1)
|
||
|
||
# ── 18. DB — только sessions (мёртвые таблицы удалены) ──────────
|
||
print("\n═══ 18. DB чистота ═══")
|
||
|
||
import sqlite3
|
||
db4 = Database(TEST_DB)
|
||
tables = [r[0] for r in db4.conn.execute("SELECT name FROM sqlite_master WHERE type='table'").fetchall()]
|
||
test("нет мёртвых таблиц (cars,tokens)", not any(t in tables for t in ["cars", "diagnostic_tokens", "dtc_codes"]))
|
||
|
||
# Убедиться что мёртвые методы вызывают AttributeError
|
||
test("get_or_create_car удалён", not hasattr(db4, "get_or_create_car"))
|
||
test("create_token удалён", not hasattr(db4, "create_token"))
|
||
test("add_llm_message удалён", not hasattr(db4, "add_llm_message"))
|
||
test("add_parameter удалён", not hasattr(db4, "add_parameter"))
|
||
test("add_dtc удалён", not hasattr(db4, "add_dtc"))
|
||
|
||
db4.close()
|
||
if TEST_DB.exists():
|
||
TEST_DB.unlink()
|