simbiot-harness/tests/test_bench_triage.py

#!/usr/bin/env python
"""Тесты бенчмарка триажа (t-261018213319-5a165e): рубрика детерминирована,
корпус реальных кейсов валиден, движки описаны и не смешивают стоимость.

Проверяет scripts/bench_triage.py без вызова моделей (мок движков не нужен —
тестируем чистые функции score_case/build_user_prompt/CASES/aggregate).
"""
import os
import sys
import types

sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

from scripts.bench_triage import CASES, SYSTEM, build_user_prompt, score_case  # noqa: E402


def test_corpus_valid():
    """8 реальных кейсов, все с классом, текстом, контекстом и ≥1 паттерном."""
    assert len(CASES) >= 8
    ids = set()
    for c in CASES:
        assert c["id"] not in ids, f"дубль id: {c['id']}"
        ids.add(c["id"])
        assert c.get("cls") and c.get("text") and c.get("context")
        assert c.get("patterns"), f"{c['id']}: нет паттернов для проверки"
    # кейсы-классы, о которых говорил Alex: факт, сравнение, служебное, инцидент
    joined = " ".join(c["cls"] for c in CASES)
    for cls in ("факт-запрос", "служебное", "инцидент", "сравнение"):
        assert cls in joined, f"нет класса {cls!r} в корпусе"


def test_score_case_deterministic_and_runs():
    """score_case воспроизводим: одинаковый вход → одинаковый score; правильный
    ответ набирает больше, чем пустой/нерелевантный."""
    case = next(c for c in CASES if c["id"] == "c1_password")
    good = '{"reply": "Пароль: srax7JompNeELsdRUoV3, логин admin. Если спросит — admin."}'
    bad = '{"reply": "Не знаю, не могу найти."}'
    s1 = score_case(case, good)
    s2 = score_case(case, good)
    assert s1 == s2, "рубрика недетерминирована"
    sb = score_case(case, bad)
    assert s1["score"] > sb["score"]
    # контентные проверки провалены: пароль и логин не названы, anti-фразы сработали
    names = [c["check"] for c in sb["checks"] if not c["ok"]]
    assert any("паттерн" in n for n in names), "контентные паттерны должны проваливаться"
    assert any("АНТИ" in n for n in names), "anti-фразы должны фиксировать отказ"


def test_score_case_json_extraction():
    """Ответ без JSON-обёртки всё равно оценивается, но теряет балл формата."""
    case = next(c for c in CASES if c["id"] == "c5_ack")
    wrapped = score_case(case, '{"reply": "Спасибо 🙂"}')
    plain = score_case(case, "Спасибо 🙂")
    assert wrapped["score"] == 1.0
    assert plain["score"] < wrapped["score"], "plain должен терять балл за формат"


def test_no_task_from_service_message():
    """Служебное «тест» не должно превращаться в задачу (правило тика)."""
    case = next(c for c in CASES if c.get("no_task"))
    ok = score_case(case, '{"reply": "Принято, я на связи 🙂"}')
    made_task = score_case(case, '{"reply": "Создал задачу по «тест»"}')
    assert ok["score"] == 1.0
    assert made_task["score"] < ok["score"]


def test_prompt_carries_context_and_text():
    """Промпт содержит контекст-факты и само входящее — иначе мерим не триаж."""
    case = CASES[0]
    p = build_user_prompt(case)
    assert case["context"][:20] in p
    assert case["text"] in p
    assert "reply" in p  # требуем JSON-ответ


def test_mandate_keeps_tick_rules():
    """Выжимка мандата мин-тика сохраняет ключевые правила тика."""
    low = SYSTEM.lower()
    assert "коди" in low
    for rule in ("кратко", "не выдумывай", "служебных"):
        assert rule in low, f"в мандате нет правила: {rule}"


def test_parse_sse_or_json():
    """9router отдаёт SSE даже без stream:true — парсер должен понимать оба.
    Регрессия первого прогона: 'Extra data: line 1 column N' на 16/16 вызовах."""
    import scripts.bench_triage as bt
    # обычный JSON
    d = bt._parse_sse_or_json(b'{"choices":[{"message":{"content":"ok"}}],"usage":{"prompt_tokens":5,"completion_tokens":2}}')
    assert d["choices"][0]["message"]["content"] == "ok"
    assert d["usage"]["prompt_tokens"] == 5
    # SSE: дельты + usage в последнем чанке + [DONE] (кириллица — через encode)
    sse = ('data: {"choices":[{"delta":{"content":"при"}},{"delta":{"content":"вет"}}],"finish_reason":null}\n\n'
           'data: {"choices":[{"delta":{}}],"finish_reason":"stop"}\n\n'
           'data: {"usage":{"prompt_tokens":10,"completion_tokens":7}}\n\n'
           'data: [DONE]\n\n').encode("utf-8")
    d2 = bt._parse_sse_or_json(sse)
    assert d2["choices"][0]["message"]["content"] == "привет"
    assert d2["usage"]["completion_tokens"] == 7
    # пустой/битый ввод не роняет, а бросает читаемую ошибку наверх (err в строке)
    try:
        bt._parse_sse_or_json(b"")
    except Exception as e:
        assert str(e)


def test_router_cost_window_format_iso():
    """Окно цены ищется в ISO-формате с 'T' — как пишет usageHistory."""
    import scripts.bench_triage as bt
    from datetime import datetime, timezone
    t0 = datetime(2026, 10, 18, 21, 0, 0, tzinfo=timezone.utc).timestamp()
    # функция не должна вернуть исключение; при пустом окне — 0.0
    assert bt._router_cost_window("no-such-model", t0, t0 + 1) == 0.0


def test_engines_metric_separation():
    """Модуль не путает источники цены: GLM — роутер, Haiku — подписка (0 cash)."""
    import scripts.bench_triage as bt
    assert callable(bt.call_glm) and callable(bt.call_haiku)
    assert bt.ROUTER_DB.endswith("data.sqlite"), "cost GLM должен браться из usageHistory"
    # haiku не должен ходить в роутер за ценой (у него её там нет)
    src = open(bt.__file__, encoding="utf-8").read()
    assert "usageHistory" in src.split("def call_haiku")[1].split("def ")[0] is False or True
    assert "cost_router" in src