lct-hack/backend/tests/test_scoring.py

"""Детерминированная оценка по ГОСТ.

Проверяются пункты приёмки lct-12: воспроизводимость, ни одной отметки без
обоснования, время опроса, полнота опроса, неверная ДДС.
"""

import json
from pathlib import Path

import pytest

from app.domain.events import CallEndReason
from app.domain.kio import KIO, derive_incident
from app.domain.taxonomy import Competency, ErrorCode
from app.scenarios.loader import load_file
from app.scoring.competency import radar
from app.scoring.gost import evaluate
from app.session.timers import SessionTimers

LIBRARY = Path(__file__).resolve().parents[2] / "scenarios"
ALL_FACTS = ["f_address", "f_what_burns", "f_people", "f_smoke", "f_gas"]


@pytest.fixture(scope="module")
def scenario():
    return load_file(LIBRARY / "fire-apartment-l2.yaml", LIBRARY)


def timeline(answer_s: float = 5, interview_s: float = 60) -> SessionTimers:
    """Таймеры с явными моментами событий — без часов, чтобы результат не плыл."""
    timers = SessionTimers()
    timers.on_event("call.incoming", now=0.0)
    timers.on_event("call.answer", now=answer_s)
    timers.on_event("dds.dispatch", now=answer_s + interview_s)
    return timers


def good_card() -> KIO:
    """Карточка, заполненная верно. Признаки — то, что курсант проставляет сам;
    код ЕКП и список оповещения из них выводит домен, а не тест."""
    card = KIO(
        address="ул. Ленина, д. 14, кв. 47",
        floor="5",
        incident_type="fire",
        dds="01",
        victims_count=2,
        description="горит балкон",
        signs=["жилой дом", "балкон", "открытое пламя"],
    )
    return derive_incident(card)


def test_perfect_call_passes_everything(scenario):
    result = evaluate(scenario=scenario, kio=good_card(), timers=timeline(), revealed_facts=ALL_FACTS)
    failed = [metric.key for metric in result.metrics if not metric.passed]
    assert failed == [], f"провалено: {failed}"
    assert result.findings == []
    assert result.score == 100.0


def test_same_call_gives_identical_json(scenario):
    """Оценку можно предъявить и проверить только если она повторяется."""
    def run():
        result = evaluate(
            scenario=scenario, kio=KIO(dds="03"), timers=timeline(interview_s=94),
            revealed_facts=["f_address"],
        )
        return json.dumps(
            {
                "metrics": [m.model_dump() for m in result.metrics],
                "findings": [f.model_dump(mode="json") for f in result.findings],
                "score": result.score,
            },
            ensure_ascii=False,
            sort_keys=True,
        )

    assert run() == run()


def test_interview_over_75_seconds_is_e3_with_the_numbers(scenario):
    result = evaluate(scenario=scenario, kio=good_card(), timers=timeline(interview_s=94), revealed_facts=ALL_FACTS)
    metric = next(m for m in result.metrics if m.key == "interview_time")
    assert not metric.passed
    assert (metric.fact, metric.norm) == ("94 с", "≤ 75 с")
    assert metric.ref == "ГОСТ Р 22.7.03-2021"

    finding = next(f for f in result.findings if f.code is ErrorCode.E3)
    assert "94 с" in finding.summary and "75 с" in finding.summary


def test_interview_never_stopped_is_a_failure_not_a_pass(scenario):
    """Таймер не остановлен событием — время недоказуемо. Молча засчитать
    такую метрику значит поставить зачёт за непереданную карточку."""
    timers = SessionTimers()
    timers.on_event("call.incoming", now=0.0)
    timers.on_event("call.answer", now=3.0)
    result = evaluate(scenario=scenario, kio=good_card(), timers=timers, revealed_facts=ALL_FACTS)
    metric = next(m for m in result.metrics if m.key == "interview_time")
    assert not metric.passed and "не зафиксировано" in metric.fact


def test_each_missing_fact_is_its_own_e1_with_the_reference_question(scenario):
    """В разборе нужен конкретный пропущенный вопрос, а не процент."""
    result = evaluate(
        scenario=scenario, kio=good_card(), timers=timeline(),
        revealed_facts=["f_address", "f_what_burns", "f_people"],
    )
    e1 = [f for f in result.findings if f.code is ErrorCode.E1]
    assert len(e1) == 2
    assert all("эталонный вопрос" in f.norm for f in e1)

    completeness = next(m for m in result.metrics if m.key == "checklist_completeness")
    assert completeness.fact == "добыто 3 из 5 обязательных фактов"


def test_wrong_signs_are_e2(scenario):
    """Оператор ошибся признаком — и происшествие уехало не в ту службу.
    Службу он не выбирает, поэтому и ошибка именно в признаках."""
    card = wrong_routing_card()
    result = evaluate(scenario=scenario, kio=card, timers=timeline(), revealed_facts=ALL_FACTS)
    finding = next(f for f in result.findings if f.code is ErrorCode.E2)
    assert "квартира" in finding.summary, finding.summary
    assert "балкон" in finding.summary, "в отметке не видно верного признака"


def test_notify_list_is_derived_not_chosen(scenario):
    """Список оповещения считается по ЕКП: 01 в нём есть, а выбирать его
    оператор не может."""
    card = good_card()
    assert card.incident_code == scenario.ground_truth.incident_code
    assert "Служба 101" in card.notify
    assert len(card.notify) > 1, "происшествие почти всегда поднимает больше одной службы"


def test_address_abbreviations_are_not_punished(scenario):
    for written in ("улица Ленина, 14", "ул. Ленина д. 14 кв. 47", "Ленина 14"):
        card = good_card()
        card.address = written
        result = evaluate(scenario=scenario, kio=card, timers=timeline(), revealed_facts=ALL_FACTS)
        assert next(m for m in result.metrics if m.key == "address").passed, written


def test_wrong_building_is_caught(scenario):
    card = good_card()
    card.address = "улица Ленина, 41"
    result = evaluate(scenario=scenario, kio=card, timers=timeline(), revealed_facts=ALL_FACTS)
    assert not next(m for m in result.metrics if m.key == "address").passed


def test_no_finding_without_justification(scenario):
    """Ни одной отметки без кода и обоснования — ни в одном интерфейсе."""
    result = evaluate(
        scenario=scenario, kio=KIO(), timers=SessionTimers(), revealed_facts=[],
        end_reason=CallEndReason.DROPPED,
    )
    assert result.findings, "пустая карточка должна дать отметки"
    for finding in result.findings:
        assert finding.code and finding.summary and finding.fact and finding.norm, finding


def test_without_slot_machine_completeness_is_not_silently_passed(scenario):
    result = evaluate(scenario=scenario, kio=good_card(), timers=timeline(), revealed_facts=None)
    assert "checklist_completeness" not in [m.key for m in result.metrics]
    assert any("checklist_completeness" in note for note in result.unavailable)


def test_dropped_call_without_callback_fails(scenario):
    result = evaluate(
        scenario=scenario, kio=good_card(), timers=timeline(), revealed_facts=ALL_FACTS,
        end_reason=CallEndReason.DROPPED,
    )
    callback = next(m for m in result.metrics if m.key == "callback")
    assert not callback.passed and callback.fact == "обратного дозвона не было"


def wrong_routing_card() -> KIO:
    """Тот же вызов, но признак второго уровня выбран неверно: «квартира»
    вместо «балкона» — другой код ЕКП и другой список оповещения."""
    card = good_card()
    card.signs = ["жилой дом", "квартира", "открытое пламя"]
    return derive_incident(card)


def test_radar_is_a_projection_of_the_same_metrics(scenario):
    card = wrong_routing_card()
    result = evaluate(scenario=scenario, kio=card, timers=timeline(interview_s=94), revealed_facts=ALL_FACTS)
    values = {score.competency: score.value for score in radar(result.metrics)}

    # Маршрутизацию меряет одна метрика — признаки. Выбор службы из неё убран:
    # список оповещения производен, и штрафовать за него отдельно значит
    # наказать дважды за одну ошибку (docs/spec/DATASET.md).
    assert values["routing"] == 0.0, "единственная метрика маршрутизации провалена"
    assert values["card"] == 1.0
    assert Competency.COMMUNICATION.value not in values, "без судьи коммуникация не оценивалась, а не провалена"


def test_legal_interview_is_not_an_e3(scenario):
    """70 с опроса укладываются в норматив 75 с. Раньше таймер оповещения ДДС
    стартовал на ответе, мерил тот же отрезок с лимитом 60 с и выставлял E3
    за законный опрос. Пока нет события конца опроса, он не считается."""
    result = evaluate(scenario=scenario, kio=good_card(), timers=timeline(interview_s=70), revealed_facts=ALL_FACTS)
    assert [f for f in result.findings if f.code is ErrorCode.E3] == []
    assert any(note.startswith("dds_notify_time") for note in result.unavailable)