tests/test_recsys_graph.py

"""Sanity checks for NetworkX Personalized PageRank wrapper."""
from __future__ import annotations

from pathlib import Path

import networkx as nx
import pytest
from sqlmodel import Session, SQLModel, create_engine

from backend import Backend
from backend.recsys_graph import build_similarity_digraph, personalized_pagerank_scores, warm_ppr_graph
from models import Kink, SimilarityEdge, Source


@pytest.fixture()
def tiny_engine(tmp_path: Path):
    db = tmp_path / "t.db"
    engine = create_engine(f"sqlite:///{db}")
    SQLModel.metadata.create_all(engine)
    with Session(engine) as session:
        session.add(Source(id="test", name="t", source_type="t", base_url="", license_model="t"))
        for kid, name in [("seed", "Seed"), ("a", "A Play"), ("b", "B Play")]:
            session.add(
                Kink(
                    id=kid,
                    name=name,
                    cluster="fetlife_fetish",
                    short_definition="x",
                    notes="",
                    risk_level="low",
                    is_extreme=False,
                )
            )
        session.add(
            SimilarityEdge(
                id="sim_seed_a",
                left_kink_id="seed",
                right_kink_id="a",
                similarity_type="catalog",
                score=0.9,
                method="test",
            )
        )
        session.add(
            SimilarityEdge(
                id="sim_a_b",
                left_kink_id="a",
                right_kink_id="b",
                similarity_type="catalog",
                score=0.5,
                method="test",
            )
        )
        session.commit()
    return engine


def test_build_digraph_sums_parallel_weights():
    rows = [
        SimilarityEdge(
            id="1",
            left_kink_id="u",
            right_kink_id="v",
            similarity_type="catalog",
            score=0.5,
            method="m",
        ),
        SimilarityEdge(
            id="2",
            left_kink_id="u",
            right_kink_id="v",
            similarity_type="catalog",
            score=0.5,
            method="m",
        ),
    ]
    G = build_similarity_digraph(rows)
    assert G["u"]["v"]["weight"] == pytest.approx(0.6)  # 1.0 * 0.6 catalog weight


def test_build_digraph_scenario_bridge_weight():
    rows = [
        SimilarityEdge(
            id="sim_scenario_bridge_u_v",
            left_kink_id="u",
            right_kink_id="v",
            similarity_type="scenario_bridge",
            score=0.5,
            method="m",
        ),
    ]
    G = build_similarity_digraph(rows)
    assert G["u"]["v"]["weight"] == pytest.approx(0.5 * 0.85)


def test_ppr_prefers_neighbor_of_seed(tiny_engine):
    pr = personalized_pagerank_scores(tiny_engine, {"seed": 1.0}, alpha=0.85)
    assert "a" in pr and "b" in pr
    # Mass should flow seed -> a -> b; a typically >= b
    assert pr["a"] >= pr["b"]


def test_ppr_backend_cached_matches_engine_only(tiny_engine):
    """CSR-cached backend path must match one-shot NetworkX for the same DB."""
    db_path = Path(tiny_engine.url.database)
    b = Backend(db_path)
    warm_ppr_graph(b)
    w = {"seed": 1.0}
    alpha = 0.85
    ref = personalized_pagerank_scores(b.engine, w, alpha=alpha)
    fast = personalized_pagerank_scores(b.engine, w, alpha=alpha, backend=b)
    keys = set(ref) | set(fast)
    for k in keys:
        assert abs(ref.get(k, 0.0) - fast.get(k, 0.0)) < 1e-5, (k, ref.get(k), fast.get(k))


@pytest.fixture()
def bridge_engine(tmp_path: Path):
    """scenario --scenario_bridge--> canon --catalog--> leaf (seed = scenario)."""
    db = tmp_path / "bridge.db"
    engine = create_engine(f"sqlite:///{db}")
    SQLModel.metadata.create_all(engine)
    with Session(engine) as session:
        session.add(Source(id="test", name="t", source_type="t", base_url="", license_model="t"))
        for kid, name in [
            ("scenario", "Very long scenario-style kink name here"),
            ("canon", "Short"),
            ("leaf", "Leaf play"),
        ]:
            session.add(
                Kink(
                    id=kid,
                    name=name,
                    cluster="fetlife_fetish",
                    short_definition="x",
                    notes="",
                    risk_level="low",
                    is_extreme=False,
                )
            )
        session.add(
            SimilarityEdge(
                id="sim_scenario_bridge_scenario_canon",
                left_kink_id="scenario",
                right_kink_id="canon",
                similarity_type="scenario_bridge",
                score=1.0,
                method="test",
            )
        )
        session.add(
            SimilarityEdge(
                id="sim_canon_leaf",
                left_kink_id="canon",
                right_kink_id="leaf",
                similarity_type="catalog",
                score=0.9,
                method="test",
            )
        )
        session.commit()
    return engine


def test_ppr_reaches_canonical_via_scenario_bridge(bridge_engine):
    pr = personalized_pagerank_scores(bridge_engine, {"scenario": 1.0}, alpha=0.85)
    assert "canon" in pr and "leaf" in pr
    assert pr["canon"] >= pr["leaf"]


def test_networkx_pagerank_two_node_chain():
    """OTS reference: two-node chain personalization peaks at first hop."""
    G = nx.DiGraph()
    G.add_edge("s", "x", weight=1.0)
    G.add_edge("x", "y", weight=1.0)
    pers = {"s": 1.0}
    pr = nx.pagerank(G, alpha=0.85, personalization=pers, weight="weight")
    assert pr["x"] > pr["y"]