Files
Orchestrator/bahn/teamlandkarte-mcp/tests/test_rrf.py
T
ankn a5f8fb49ab Migrate all repos into monorepo context folders
Bahn: aisupport, Analyse-O2C-C2S, awesome-bahn-mcp-servers, beam-mcp,
      Confluence_Bot, db-planet-mcp-server, O2C-Harness, project-audit,
      Projekt-KIQ-HP, teamlandkarte-mcp
Dhive: Jury-Voting
Privat: CV, NoteGraph (NOTE: NoteGraph needs complete redo after consolidation)
Shared: AI-Orchestrator, OrgMyLife, power_skills_and_more
Shared/references: symphony (read-only)

Bahn repos remain available as independent remotes - this monorepo
pulls them in via subtree, the originals are untouched.
2026-06-30 20:39:52 +02:00

143 lines
4.6 KiB
Python

# filepath: tests/test_rrf.py
"""Unit tests for :mod:`teamlandkarte_mcp.matching.rrf`.
Covers edge cases of :func:`reciprocal_rank_fusion`: empty input, all-zero
input, single list, multiple lists, normalization, and the ``k`` parameter.
"""
from __future__ import annotations
import math
import pytest
from teamlandkarte_mcp.matching.rrf import reciprocal_rank_fusion
# ---------------------------------------------------------------------------
# Empty / all-zero inputs
# ---------------------------------------------------------------------------
def test_rrf_empty_input_returns_empty() -> None:
"""No ranked lists → empty result."""
assert reciprocal_rank_fusion([]) == {}
def test_rrf_empty_list_returns_empty() -> None:
"""Single empty ranked list → empty result."""
assert reciprocal_rank_fusion([[]]) == {}
def test_rrf_all_zero_scores_returns_empty() -> None:
"""All-zero BM25 scores → no token overlap signal → empty result."""
result = reciprocal_rank_fusion([[("A", 0.0), ("B", 0.0)]])
assert result == {}
def test_rrf_mixed_zero_and_nonzero_list() -> None:
"""Only the non-zero list contributes; all-zero list is skipped."""
result = reciprocal_rank_fusion(
[
[("A", 3.5), ("B", 1.0)],
[("A", 0.0), ("B", 0.0)],
]
)
# Both A and B should appear (from first list only)
assert "A" in result
assert "B" in result
# ---------------------------------------------------------------------------
# Single list — normalization to [0, 1]
# ---------------------------------------------------------------------------
def test_rrf_single_list_single_candidate_score_one() -> None:
"""Single candidate in single list → score 1.0."""
result = reciprocal_rank_fusion([[("Python", 3.5)]])
assert result == {"Python": 1.0}
def test_rrf_single_list_top_candidate_score_one() -> None:
"""Top-ranked candidate always maps to 1.0 after normalization."""
result = reciprocal_rank_fusion(
[[("A", 5.0), ("B", 2.0), ("C", 0.5)]]
)
assert math.isclose(result["A"], 1.0, rel_tol=1e-9)
def test_rrf_single_list_scores_in_descending_order() -> None:
"""Higher-ranked candidates receive higher normalized scores."""
result = reciprocal_rank_fusion(
[[("A", 5.0), ("B", 2.0), ("C", 0.5)]]
)
assert result["A"] > result["B"] > result["C"]
def test_rrf_single_list_all_scores_in_unit_interval() -> None:
"""All scores are in [0.0, 1.0]."""
result = reciprocal_rank_fusion(
[[("A", 5.0), ("B", 2.0), ("C", 0.5)]]
)
for score in result.values():
assert 0.0 <= score <= 1.0
def test_rrf_zero_score_entry_not_in_result() -> None:
"""Candidate with BM25 score 0.0 receives no RRF credit (zero-out rule)."""
result = reciprocal_rank_fusion(
[[("A", 3.5), ("B", 0.0)]]
)
assert "A" in result
assert "B" not in result
# ---------------------------------------------------------------------------
# k parameter
# ---------------------------------------------------------------------------
def test_rrf_default_k_rank2_score() -> None:
"""With default k=60, rank-2 score normalized = 61/62 ≈ 0.9839."""
result = reciprocal_rank_fusion(
[[("A", 2.0), ("B", 1.0)]]
)
# raw(A) = 1/61, raw(B) = 1/62
# normalized(B) = (1/62) / (1/61) = 61/62
expected_b = 61.0 / 62.0
assert math.isclose(result["B"], expected_b, rel_tol=1e-9)
def test_rrf_higher_k_flattens_differences() -> None:
"""Higher k → smaller gap between rank-1 and rank-2 scores."""
result_low_k = reciprocal_rank_fusion(
[[("A", 2.0), ("B", 1.0)]], k=1
)
result_high_k = reciprocal_rank_fusion(
[[("A", 2.0), ("B", 1.0)]], k=1000
)
gap_low = result_low_k["A"] - result_low_k["B"]
gap_high = result_high_k["A"] - result_high_k["B"]
assert gap_low > gap_high
# ---------------------------------------------------------------------------
# Multiple lists (fusion)
# ---------------------------------------------------------------------------
def test_rrf_two_lists_fuse_correctly() -> None:
"""Candidate appearing in both lists accumulates RRF scores."""
result = reciprocal_rank_fusion(
[
[("A", 3.0), ("B", 1.0)],
[("B", 2.5), ("A", 0.5)],
]
)
# A gets credit from list-1 (rank 1) and list-2 (rank 2).
# B gets credit from list-1 (rank 2) and list-2 (rank 1).
assert "A" in result
assert "B" in result
# Both should be < 1.0 unless one dominates; top candidate maps to 1.0.
assert max(result.values()) == pytest.approx(1.0)