"""Reciprocal Rank Fusion — the maths the whole retrieval story rests on.""" from __future__ import annotations import pytest from app.rag.retrieve import reciprocal_rank_fusion class TestReciprocalRankFusion: def test_single_list_is_monotonically_decreasing(self): fused = reciprocal_rank_fusion({"dense": ["a", "b", "c"]}, k=60) assert fused["a"] > fused["b"] > fused["c"] def test_exact_values_match_the_formula(self): """RRF(d) = sum over retrievers of 1 / (k + rank), rank 1-based.""" fused = reciprocal_rank_fusion({"dense": ["a", "b"]}, k=60) assert fused["a"] == pytest.approx(1 / 61) assert fused["b"] == pytest.approx(1 / 62) def test_agreement_beats_a_single_strong_vote(self): """A document both retrievers rank first must outrank one retriever's favourite. This is the property that makes fusion worth doing: it rewards agreement rather than any single retriever's confidence. """ fused = reciprocal_rank_fusion( {"dense": ["both", "dense_only"], "sparse": ["both", "sparse_only"]}, k=60 ) assert fused["both"] == pytest.approx(2 / 61) assert fused["both"] > fused["dense_only"] assert fused["both"] > fused["sparse_only"] def test_scores_are_never_compared_across_retrievers(self): """Only ranks matter — a retriever's score scale cannot influence the result.""" a = reciprocal_rank_fusion({"dense": ["x", "y"], "sparse": ["y", "x"]}, k=60) b = reciprocal_rank_fusion({"sparse": ["y", "x"], "dense": ["x", "y"]}, k=60) assert a == b def test_k_damps_the_head_of_each_list(self): """Larger k flattens the gap between rank 1 and rank 2.""" small = reciprocal_rank_fusion({"d": ["a", "b"]}, k=1) large = reciprocal_rank_fusion({"d": ["a", "b"]}, k=1000) assert (small["a"] - small["b"]) > (large["a"] - large["b"]) def test_empty_input_yields_empty_output(self): assert reciprocal_rank_fusion({}, k=60) == {} assert reciprocal_rank_fusion({"dense": []}, k=60) == {} @pytest.mark.parametrize("k", [0, -1, -60]) def test_rejects_k_below_one(self, k: int): """k=-1 divides by zero at rank 1 and small k inverts the ordering.""" with pytest.raises(ValueError, match="RRF k must be"): reciprocal_rank_fusion({"dense": ["a"]}, k=k) def test_missing_from_one_list_still_scores(self): fused = reciprocal_rank_fusion({"dense": ["a"], "sparse": ["b"]}, k=60) assert fused["a"] == pytest.approx(1 / 61) assert fused["b"] == pytest.approx(1 / 61)