diff --git a/api/parser.py b/api/parser.py index 3179e5c..12e4b73 100644 --- a/api/parser.py +++ b/api/parser.py @@ -52,9 +52,13 @@ def _parse_vin(result: dict, raw: str, decoded: str): # Fallback: парсим VIN из raw HEX if "49" in raw and ("02" in raw or "4902" in raw.replace(" ", "")): - clean = raw.replace(":", "").replace(" ", "").upper() + # Чистим: убираем CAN адресацию (N:), ISO-TP (10..), двоеточия + import re + clean = re.sub(r'^\d\s*:\s*', '', raw, flags=re.MULTILINE) # "0:" "1:" + clean = re.sub(r'\b(?:10|14|21|22)\s+', '', clean) # ISO-TP PCI + clean = clean.replace(":", "").replace(" ", "").upper() if "490201" in clean: - hex_str = clean.split("490201")[-1][:34] + hex_str = clean.split("490201")[-1].replace("\n", "").replace("\r", "")[:34] vin = "" for i in range(0, len(hex_str) - 1, 2): try: diff --git a/api/routes.py b/api/routes.py index 1a468af..fdb0ff2 100644 --- a/api/routes.py +++ b/api/routes.py @@ -30,24 +30,27 @@ def _build_diagnosis_prompt(data: dict, car_info: str = "") -> str: if car_info: parts.append(f"**Водитель сообщил:** {car_info}") - if data["vin"]: + if data.get("vin"): parts.append(f"**VIN:** {data['vin']}") - if data["dtc_stored"]: + if data.get("dtc_stored"): parts.append(f"\n**Сохранённые ошибки (mode 03):** {', '.join(data['dtc_stored'])}") - if data["dtc_pending"]: + if data.get("dtc_pending"): parts.append(f"**Ожидающие ошибки (mode 07):** {', '.join(data['dtc_pending'])}") - if data["parameters"]: + if data.get("parameters"): parts.append("\n**Параметры в реальном времени:**") for p in data["parameters"]: parts.append(f"- {p['name']}: {p['value']}") - if not data["vin"] and not data["dtc_stored"] and not data["parameters"]: + if not data.get("vin") and not data.get("dtc_stored") and not data.get("parameters"): parts.append("\n(данные не распознаны)") + raw_log = data.get("raw_log") or [] parts.append("\n**Сырые ответы ЭБУ:**") - parts.extend(data["raw_log"]) + parts.extend(raw_log) + + parts.append("\n---") parts.append("\n---") parts.append("## Запрос на анализ") diff --git a/tests/test_all.py b/tests/test_all.py index 7fe65e9..6cfa33f 100644 --- a/tests/test_all.py +++ b/tests/test_all.py @@ -189,10 +189,255 @@ 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("ВСЁ ОК!") +# ── 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"] == [])