tests/test_adversarial_api.py

from __future__ import annotations

import importlib
import os
import sys
from pathlib import Path
from types import SimpleNamespace

import pytest
from fastapi.testclient import TestClient
from sqlmodel import Session

from backend.core import Backend
from backend.models import RecsysSettings
from models import Kink, SimilarityEdge


def _headers(user) -> dict[str, str]:
    return {"x-private-token": user.private_token}


def _all_match_lists_empty(payload: dict) -> bool:
    keys = ["direct_matches", "related_matches", "scenario_matches", "scenario_related_matches", "similar_matches"]
    return all(payload.get(key) == [] for key in keys)


@pytest.fixture()
def adversarial_api(tmp_path: Path, monkeypatch: pytest.MonkeyPatch):
    os.environ["KINK_SKIP_HEAVY_WARM"] = "1"
    db = tmp_path / "adversarial.db"
    b = Backend(db, recsys_settings=RecsysSettings())
    with Session(b.engine) as session:
        for kid, name in [
            ("oral", "Oral Sex"),
            ("impact", "Spanking"),
            ("bondage", "Rope Bondage"),
        ]:
            session.add(
                Kink(
                    id=kid,
                    name=name,
                    cluster="fetlife_fetish",
                    short_definition=f"{name} fixture",
                    notes="Kinksters: 1000",
                ),
            )
        session.add(
            SimilarityEdge(
                id="sim_oral_bondage",
                left_kink_id="oral",
                right_kink_id="bondage",
                similarity_type="catalog",
                score=0.8,
                method="test",
            ),
        )
        session.commit()
    b._invalidate_catalog_cache(force=True)

    owner = b.create_user()
    member = b.create_user()
    other_partner = b.create_user()
    stranger = b.create_user()
    assert b.request_partner_link(owner.id, member.id) == (True, "sent")
    assert b.accept_partner_request(member.id, owner.id)
    assert b.request_partner_link(owner.id, other_partner.id) == (True, "sent")
    assert b.accept_partner_request(other_partner.id, owner.id)
    group = b.create_partner_group(owner.id, "Private pair", [member.id])
    assert group
    assert b.toggle_share_full_list(member.id, group["id"], True)
    b.save_play_preference(owner.id, "oral", "like", ["together"])
    b.save_play_preference(member.id, "oral", "love", ["together"])
    b.save_play_preference(owner.id, "impact", "like", ["together"])
    b.save_play_preference(other_partner.id, "impact", "love", ["together"])

    monkeypatch.setenv("KINK_STORE_PATH", str(db))
    monkeypatch.setenv("KINK_SKIP_HEAVY_WARM", "1")
    monkeypatch.setenv("KINK_HF_REQUIRE_FULL_CATALOG", "0")
    monkeypatch.delenv("KINK_RECSYS_SOURCE", raising=False)
    monkeypatch.delenv("KINK_RECSYS_CANDIDATES_PATH", raising=False)
    sys.modules.pop("api", None)
    api_mod = importlib.import_module("api")
    api_mod._backend_impl = None
    api_mod._get_backend()
    client = TestClient(api_mod.app)
    return SimpleNamespace(
        backend=b,
        client=client,
        owner=owner,
        member=member,
        other_partner=other_partner,
        stranger=stranger,
        group=group,
    )


def test_cross_user_tokens_and_stranger_group_access_are_rejected(adversarial_api) -> None:
    ctx = adversarial_api

    wrong_user = ctx.client.get(f"/users/{ctx.owner.id}", headers=_headers(ctx.member))
    assert wrong_user.status_code == 401

    wrong_path_token = ctx.client.get(
        f"/users/{ctx.owner.id}/partner-groups/{ctx.group['id']}/overlap",
        headers=_headers(ctx.member),
    )
    assert wrong_path_token.status_code == 401

    stranger_board = ctx.client.get(
        f"/users/{ctx.stranger.id}/partner-groups/{ctx.group['id']}/partner-board/{ctx.member.id}",
        headers=_headers(ctx.stranger),
    )
    assert stranger_board.status_code == 403

    owner_board = ctx.client.get(
        f"/users/{ctx.owner.id}/partner-groups/{ctx.group['id']}/partner-board/{ctx.member.id}",
        headers=_headers(ctx.owner),
    )
    assert owner_board.status_code == 200


def test_group_overlap_does_not_leak_non_group_partner_data(adversarial_api) -> None:
    ctx = adversarial_api

    overlap = ctx.client.get(
        f"/users/{ctx.owner.id}/partner-groups/{ctx.group['id']}/overlap",
        headers=_headers(ctx.owner),
    )
    assert overlap.status_code == 200
    payload = overlap.json()
    direct_ids = {item["id"] for item in payload["direct_matches"]}
    assert "oral" in direct_ids
    assert "impact" not in direct_ids

    stranger_overlap = ctx.client.get(
        f"/users/{ctx.stranger.id}/partner-groups/{ctx.group['id']}/overlap",
        headers=_headers(ctx.stranger),
    )
    assert stranger_overlap.status_code == 403


def test_group_scoped_recs_and_prompts_reject_inaccessible_group(adversarial_api) -> None:
    ctx = adversarial_api

    recs = ctx.client.get(
        f"/users/{ctx.stranger.id}/recommendations?group_id={ctx.group['id']}",
        headers=_headers(ctx.stranger),
    )
    assert recs.status_code == 403

    prompts = ctx.client.get(
        f"/users/{ctx.stranger.id}/prompts?group_id={ctx.group['id']}",
        headers=_headers(ctx.stranger),
    )
    assert prompts.status_code == 403

    group_prompts = ctx.client.get(
        f"/users/{ctx.stranger.id}/partner-groups/{ctx.group['id']}/prompts",
        headers=_headers(ctx.stranger),
    )
    assert group_prompts.status_code == 403


def test_direct_overlap_requires_existing_partner_link(adversarial_api) -> None:
    ctx = adversarial_api

    save = ctx.client.post(
        f"/users/{ctx.stranger.id}/plays",
        headers=_headers(ctx.stranger),
        json={"kink_id": "oral", "interest_state": "love", "directions": ["together"]},
    )
    assert save.status_code == 200

    stranger_probe = ctx.client.get(
        f"/users/{ctx.stranger.id}/overlap/{ctx.owner.id}",
        headers=_headers(ctx.stranger),
    )
    assert stranger_probe.status_code == 403

    linked_overlap = ctx.client.get(
        f"/users/{ctx.owner.id}/overlap/{ctx.member.id}",
        headers=_headers(ctx.owner),
    )
    assert linked_overlap.status_code == 200
    assert {item["id"] for item in linked_overlap.json()["direct_matches"]} == {"oral"}