tests/test_benchmark_pair_sources.py

"""Unit tests for ``backend.benchmark_pair_sources`` (pair export strategies)."""
from __future__ import annotations

import sqlite3

import pytest

from backend.benchmark_pair_sources import (
    collect_all_pair_records,
    collect_fingerprint_collision_pairs,
    collect_merge_signature_pairs,
    collect_similarity_edge_pairs,
    dedupe_pair_records,
    fingerprint_play_title,
    sample_pair_records,
)
from backend.kink_merge import merge_signature


def test_fingerprint_collision_blowjob_vs_blow_job() -> None:
    rows = [
        ("a", "blowjob", "fetlife_fetish", 1.0),
        ("b", "Blow job", "fetlife_fetish", 2.0),
    ]
    assert fingerprint_play_title(rows[0][1]) == fingerprint_play_title(rows[1][1])
    assert merge_signature(rows[0][1]) != merge_signature(rows[1][1])
    pairs = collect_fingerprint_collision_pairs(rows, max_pairs_per_fingerprint=10)
    assert len(pairs) == 1
    assert pairs[0]["strategy"] == "fingerprint_collision"
    assert pairs[0]["pair_key"] == "a|b"


def test_merge_signature_pairs_same_bucket() -> None:
    rows = [
        ("dup", "threesome fmf", "fetlife_fetish", 10.0),
        ("canon", "FMF threesomes", "fetlife_fetish", 500.0),
    ]
    pairs = collect_merge_signature_pairs(rows, max_pairs_per_bucket=10)
    assert len(pairs) == 1
    assert pairs[0]["strategy"] == "merge_signature_same_bucket"
    assert pairs[0]["pair_key"] == "canon|dup"


def test_dedupe_pair_records_keeps_first() -> None:
    a = {"pair_key": "x|y", "strategy": "merge_signature_same_bucket"}
    b = {"pair_key": "x|y", "strategy": "similarity_edge"}
    out = dedupe_pair_records([a, b])
    assert len(out) == 1
    assert out[0]["strategy"] == "merge_signature_same_bucket"


def test_collect_all_pair_records_similarity_requires_conn() -> None:
    rows: list[tuple[str, str, str, float]] = []
    with pytest.raises(ValueError, match="similarity_edge"):
        collect_all_pair_records(rows, None, strategies=("similarity_edge",))


def test_collect_all_pair_records_merge_only_no_conn() -> None:
    rows = [
        ("dup", "threesome fmf", "fetlife_fetish", 10.0),
        ("canon", "FMF threesomes", "fetlife_fetish", 500.0),
    ]
    got = collect_all_pair_records(rows, None, strategies=("merge_signature",))
    assert len(got) == 1
    assert got[0]["strategy"] == "merge_signature_same_bucket"


def test_sample_pair_records_respects_cap() -> None:
    items = [{"pair_key": f"a|b{i}", "n": i} for i in range(10)]
    got = sample_pair_records(items, 3, seed=99)
    assert len(got) == 3


def test_collect_similarity_edge_pairs_minimal_db(tmp_path) -> None:
    db = tmp_path / "t.db"
    conn = sqlite3.connect(db)
    conn.row_factory = sqlite3.Row
    conn.executescript(
        """
        CREATE TABLE kink (id TEXT PRIMARY KEY, name TEXT NOT NULL, cluster TEXT NOT NULL);
        CREATE TABLE fetlifekinkmeta (kink_id TEXT PRIMARY KEY REFERENCES kink(id), popularity REAL);
        CREATE TABLE similarityedge (
            id TEXT PRIMARY KEY,
            left_kink_id TEXT NOT NULL REFERENCES kink(id),
            right_kink_id TEXT NOT NULL REFERENCES kink(id),
            similarity_type TEXT NOT NULL,
            score REAL NOT NULL,
            method TEXT NOT NULL,
            version TEXT NOT NULL DEFAULT 'v1'
        );
        INSERT INTO kink VALUES ('k1', 'Alpha play', 'fetlife_fetish');
        INSERT INTO kink VALUES ('k2', 'Beta rope', 'fetlife_fetish');
        INSERT INTO similarityedge VALUES ('s1', 'k1', 'k2', 'catalog', 0.9, 'test', 'v1');
        """,
    )
    pairs = collect_similarity_edge_pairs(conn, min_score=0.82)
    conn.close()
    assert len(pairs) == 1
    assert pairs[0]["strategy"] == "similarity_edge"
    assert pairs[0]["pair_key"] == "k1|k2"
    assert pairs[0]["score"] == pytest.approx(0.9)