RAGcore-Public/tests/evaluation/metrics/test_calculator.py

"""compute_case_metrics / aggregate_metrics against a small evaluation run."""

from __future__ import annotations

from datetime import UTC, datetime
from uuid import uuid4

import pytest

from ragcore.evaluation import (
    AnswerabilityStatus,
    CaseResult,
    EvaluationCase,
    EvaluationDataset,
    EvaluationRun,
    RetrievalRunProfile,
)
from ragcore.evaluation.metrics import MetricsError, aggregate_metrics, compute_case_metrics


def make_profile() -> RetrievalRunProfile:
    return RetrievalRunProfile(
        embedding_profile_digest="a" * 64,
        sparse_profile_version="v1",
        index_alias="rag_dense_hybrid_active",
        index_collection_version="v1",
    )


def test_case_with_no_relevant_chunks_is_excluded_from_scoring() -> None:
    case = EvaluationCase(
        id="no-evidence",
        question="q",
        language="en",
        expected_answerability=AnswerabilityStatus.NOT_ANSWERABLE,
    )
    result = CaseResult(
        case_id="no-evidence",
        retrieved_chunk_ids=(),
        matched_expected_ids=(),
        missing_expected_ids=(),
        unexpected_ids=(),
    )
    assert compute_case_metrics(case, result) is None


def test_case_metrics_document_diversity_and_duplicate_ratio() -> None:
    chunk_a, chunk_b, chunk_c = uuid4(), uuid4(), uuid4()
    document_x, document_y = uuid4(), uuid4()
    case = EvaluationCase(
        id="case-1",
        question="q",
        language="en",
        expected_answerability=AnswerabilityStatus.ANSWERABLE,
        expected_chunk_ids=(chunk_a,),
        expected_document_ids=(document_x,),
        required_facts=("fact",),
    )
    result = CaseResult(
        case_id="case-1",
        retrieved_chunk_ids=(chunk_a, chunk_b, chunk_c),
        matched_expected_ids=(chunk_a,),
        missing_expected_ids=(),
        unexpected_ids=(chunk_b, chunk_c),
    )
    document_ids = {chunk_a: document_x, chunk_b: document_x, chunk_c: document_y}

    metrics = compute_case_metrics(case, result, document_ids=document_ids)

    assert metrics is not None
    assert metrics.document_diversity == 2  # document_x and document_y
    assert metrics.duplicate_ratio == pytest.approx(1 / 3)  # 1 duplicate out of 3
    assert metrics.recall_at_5 == 1.0
    assert metrics.reciprocal_rank == 1.0


def test_aggregate_metrics_overall_and_per_tag() -> None:
    matched_chunk = uuid4()
    document_id = uuid4()
    answerable_case = EvaluationCase(
        id="answerable-1",
        question="q1",
        language="en",
        expected_answerability=AnswerabilityStatus.ANSWERABLE,
        expected_chunk_ids=(matched_chunk,),
        expected_document_ids=(document_id,),
        required_facts=("fact",),
        tags=("exact-identifier",),
    )
    no_evidence_case = EvaluationCase(
        id="no-evidence-1",
        question="q2",
        language="en",
        expected_answerability=AnswerabilityStatus.NOT_ANSWERABLE,
        tags=("out-of-scope",),
    )
    dataset = EvaluationDataset(id="ds", version="v1", cases=(answerable_case, no_evidence_case))
    run = EvaluationRun(
        id=uuid4(),
        dataset_id="ds",
        dataset_version="v1",
        run_profile=make_profile(),
        started_at=datetime(2026, 8, 2, tzinfo=UTC),
        case_results=(
            CaseResult(
                case_id="answerable-1",
                retrieved_chunk_ids=(matched_chunk,),
                matched_expected_ids=(matched_chunk,),
                missing_expected_ids=(),
                unexpected_ids=(),
            ),
            CaseResult(
                case_id="no-evidence-1",
                retrieved_chunk_ids=(),
                matched_expected_ids=(),
                missing_expected_ids=(),
                unexpected_ids=(),
            ),
        ),
    )

    report = aggregate_metrics(dataset, run)

    assert report.overall.case_count == 1  # only the answerable case is scored
    assert report.overall.mean_recall_at_10 == 1.0
    assert report.overall.zero_result_rate == pytest.approx(0.5)  # 1 of 2 cases retrieved nothing
    assert "exact-identifier" in report.per_tag
    assert report.per_tag["exact-identifier"].mean_recall_at_10 == 1.0
    assert "out-of-scope" not in report.per_tag  # no scored case carries this tag


def test_aggregate_metrics_rejects_a_run_for_a_different_dataset() -> None:
    case = EvaluationCase(
        id="c",
        question="q",
        language="en",
        expected_answerability=AnswerabilityStatus.ANSWERABLE,
        expected_chunk_ids=(uuid4(),),
        expected_document_ids=(uuid4(),),
        required_facts=("fact",),
    )
    dataset = EvaluationDataset(id="ds", version="v1", cases=(case,))
    run = EvaluationRun(
        id=uuid4(),
        dataset_id="other",
        dataset_version="v1",
        run_profile=make_profile(),
        started_at=datetime(2026, 8, 2, tzinfo=UTC),
        case_results=(
            CaseResult(
                case_id="c",
                retrieved_chunk_ids=(),
                matched_expected_ids=(),
                missing_expected_ids=(),
                unexpected_ids=(),
            ),
        ),
    )
    with pytest.raises(MetricsError, match="does not belong"):
        aggregate_metrics(dataset, run)


def test_aggregate_metrics_requires_at_least_one_scoreable_case() -> None:
    case = EvaluationCase(
        id="c",
        question="q",
        language="en",
        expected_answerability=AnswerabilityStatus.NOT_ANSWERABLE,
    )
    dataset = EvaluationDataset(id="ds", version="v1", cases=(case,))
    run = EvaluationRun(
        id=uuid4(),
        dataset_id="ds",
        dataset_version="v1",
        run_profile=make_profile(),
        started_at=datetime(2026, 8, 2, tzinfo=UTC),
        case_results=(
            CaseResult(
                case_id="c",
                retrieved_chunk_ids=(),
                matched_expected_ids=(),
                missing_expected_ids=(),
                unexpected_ids=(),
            ),
        ),
    )
    with pytest.raises(MetricsError, match="relevant chunks to score"):
        aggregate_metrics(dataset, run)