tests/observability/test_candidate_stats.py

"""Focused tests for `duplicate_candidate_ratio` (Task 8, required gap-close).

Task 7's diagnostic record used `document_diversity` as its only diversity
proxy for the dedup decision -- this is the canonical retrieval-behavior
measure (`duplicate_candidate_ratio`, named in the plan preamble) it was
missing. These tests exercise the measurement in isolation, as pure
data-in/data-out, with no Chroma/DB/observability boundary involved -- the
mechanism (over-fetch + dedup + truncate) is tested separately in
`tests/rag/test_dedup_mechanism.py`, and this file is the "supporting,
behavior-neutral measurement addition" the task brief calls out as
verified on its own.

Fixture note on the Jaccard math: for two texts that differ in exactly one
token out of N shared tokens, intersection = N-1 and union = N+1, so
jaccard = (N-1)/(N+1). To clear the 0.8 threshold with margin, fixtures below
use base phrases of >= 12 tokens (jaccard ~0.86-0.93 for a single differing
trailing token) rather than short phrases that would land under 0.8.
"""

from __future__ import annotations

from app.observability.retrieval import (
    candidate_stats,
    duplicate_candidate_ratio,
    find_duplicate_indices,
    is_near_duplicate,
)

# 12-token base phrases with essentially no shared vocabulary between groups,
# so a single trailing "variant N" token difference clears 0.8 within a
# group while cross-group pairs land far below it.
_ALPHA = "alpha membrane transport channel proteins regulate ion flow across the lipid bilayer"
_BETA = "beta ribosome translation machinery assembles polypeptide chains from messenger rna transcripts today"


def _variant(base: str, n: str) -> str:
    return f"{base} variant {n}"


def _c(text: str, **meta) -> dict:
    return {"text": text, **meta}


# --- is_near_duplicate --------------------------------------------------


def test_identical_text_is_near_duplicate():
    assert is_near_duplicate("the quick brown fox", "the quick brown fox") is True


def test_completely_different_text_is_not_near_duplicate():
    assert (
        is_near_duplicate(
            "mitochondria produce cellular energy through respiration",
            "ribosomes translate messenger rna into polypeptide chains",
        )
        is False
    )


def test_case_and_whitespace_differences_still_count_as_near_duplicate():
    assert is_near_duplicate("Hello   World", "hello world") is True


def test_one_word_difference_in_long_overlapping_chunk_is_near_duplicate():
    # Models the chunker's 64-char-overlap adjacency: two chunks sharing all
    # but one trailing token (e.g. a section-marker suffix differs).
    a = _variant(_ALPHA, "one")
    b = _variant(_ALPHA, "two")
    assert is_near_duplicate(a, b) is True


def test_different_base_phrase_with_shared_suffix_is_not_near_duplicate():
    # Same trailing "variant one" suffix, but otherwise unrelated core text --
    # must NOT be flagged just because a few filler words match.
    a = _variant(_ALPHA, "one")
    b = _variant(_BETA, "one")
    assert is_near_duplicate(a, b) is False


def test_empty_strings_are_near_duplicate_of_each_other():
    assert is_near_duplicate("", "") is True


# --- find_duplicate_indices ---------------------------------------------


def test_find_duplicate_indices_keeps_first_occurrence_per_group():
    candidates = [
        _c(_variant(_ALPHA, "one")),
        _c(_variant(_ALPHA, "two")),  # dup of [0]
        _c(_variant(_BETA, "one")),
        _c(_variant(_ALPHA, "three")),  # dup of [0]
        _c(_variant(_BETA, "two")),  # dup of [2]
    ]
    dups = find_duplicate_indices(candidates)
    assert dups == {1, 3, 4}


def test_find_duplicate_indices_empty_list():
    assert find_duplicate_indices([]) == set()


def test_find_duplicate_indices_no_duplicates():
    candidates = [
        _c(_variant(_ALPHA, "one")),
        _c(_variant(_BETA, "one")),
        _c("gamma mitochondrial electron transport chain produces atp"),
    ]
    assert find_duplicate_indices(candidates) == set()


# --- duplicate_candidate_ratio -------------------------------------------


def test_duplicate_candidate_ratio_empty_list_is_zero():
    assert duplicate_candidate_ratio([]) == 0.0


def test_duplicate_candidate_ratio_no_duplicates_is_zero():
    candidates = [
        _c(_variant(_ALPHA, "one")),
        _c(_variant(_BETA, "one")),
    ]
    assert duplicate_candidate_ratio(candidates) == 0.0


def test_duplicate_candidate_ratio_all_but_first_are_duplicates():
    candidates = [
        _c(_variant(_ALPHA, "one")),
        _c(_variant(_ALPHA, "two")),
        _c(_variant(_ALPHA, "three")),
    ]
    # First occurrence is kept (not counted as a duplicate); the other two
    # are near-duplicates of it -> 2 of 3 flagged.
    assert duplicate_candidate_ratio(candidates) == 2 / 3


def test_duplicate_candidate_ratio_partial():
    candidates = [
        _c(_variant(_ALPHA, "one")),
        _c(_variant(_ALPHA, "two")),  # dup of [0]
        _c(_variant(_BETA, "one")),
        _c("gamma mitochondrial electron transport chain produces atp via oxidative phosphorylation"),
    ]
    assert duplicate_candidate_ratio(candidates) == 0.25


# --- candidate_stats (integration of all three measures) -----------------


def test_candidate_stats_includes_duplicate_candidate_ratio():
    candidates = [
        _c(_variant(_ALPHA, "one"), document_id="doc-a", source="paper-a"),
        _c(_variant(_ALPHA, "two"), document_id="doc-a", source="paper-a"),
        _c(_variant(_BETA, "one"), document_id="doc-b", source="paper-b"),
    ]
    stats = candidate_stats(candidates)
    assert stats["raw_candidate_count"] == 3
    assert stats["document_diversity"] == 2
    assert stats["section_diversity"] == 2
    assert stats["duplicate_candidate_ratio"] == 1 / 3


def test_candidate_stats_is_a_pure_function_of_the_list_its_given():
    """Same yardstick regardless of which pipeline produced the list -- the
    function has no notion of pipeline_version at all, it only looks at the
    candidates handed to it."""
    naive_style = [
        _c(_variant(_ALPHA, "one"), document_id="doc-a"),
        _c(_variant(_ALPHA, "two"), document_id="doc-a"),
    ]
    deduped_style = [
        _c(_variant(_ALPHA, "one"), document_id="doc-a"),
        _c(_variant(_BETA, "one"), document_id="doc-b"),
    ]
    assert candidate_stats(naive_style)["duplicate_candidate_ratio"] == 0.5
    assert candidate_stats(deduped_style)["duplicate_candidate_ratio"] == 0.0