FabledGrimoire/tests/test_services_supersession.py

"""The supersession claim — who may make it, and what it refuses.

Ranking behaviour is in test_supersession_ranking.py; this covers only
recording and reading the relation.

The cycle tests are the ones worth reading. Under FLAT demotion a ring of
records that supersede each other claims every member is obsolete, so all of
them get demoted equally and the whole set drops out of ranked retrieval
together — with nothing in the data saying why.
"""
from __future__ import annotations

from unittest.mock import AsyncMock, MagicMock, patch

import pytest

from grimoire.services import supersession
from tests.helpers import make_mock_session


def _session(scalars_sequence=None, get_returns=None):
    """A mocked async_session whose execute() yields successive scalar lists."""
    s = make_mock_session()

    results = []
    for scalars in scalars_sequence or []:
        r = MagicMock()
        r.scalars.return_value.all.return_value = scalars
        results.append(r)
    s.execute = AsyncMock(side_effect=results or None)
    s.get = AsyncMock(side_effect=get_returns) if get_returns else AsyncMock()
    s.commit = AsyncMock()
    s.add = MagicMock()
    return s


def _live_note(note_id=1):
    n = MagicMock()
    n.id, n.deleted_at = note_id, None
    return n


@pytest.mark.asyncio
async def test_a_caller_who_cannot_write_the_note_gets_none():
    with patch("grimoire.services.supersession.access.can_write_note",
               AsyncMock(return_value=False)):
        assert await supersession.set_supersedes(7, 1, [2]) is None


@pytest.mark.asyncio
async def test_superseding_a_note_you_can_only_READ_is_refused_not_dropped():
    """The one case that raises rather than silently skipping.

    Demoting someone else's record out of their retrieval is damage that is
    invisible from the outside — the caller would believe it worked, and the
    owner would have no symptom to trace.
    """
    # writable for the superseder (id 1), not for the target (id 2)
    writable = AsyncMock(side_effect=lambda uid, nid: nid == 1)
    session = _session(get_returns=[_live_note(2)])
    with patch("grimoire.services.supersession.access.can_write_note", writable), \
         patch("grimoire.services.supersession.async_session", return_value=session):
        with pytest.raises(PermissionError, match="not yours to supersede"):
            await supersession.set_supersedes(7, 1, [2])


@pytest.mark.asyncio
async def test_self_supersession_is_dropped_silently():
    """Meaningless rather than dangerous, and the DB CHECK refuses it anyway —
    so it is a drop, not an error the caller has to handle."""
    session = _session(scalars_sequence=[[]])
    with patch("grimoire.services.supersession.access.can_write_note",
               AsyncMock(return_value=True)), \
         patch("grimoire.services.supersession.async_session", return_value=session):
        assert await supersession.set_supersedes(7, 5, [5]) == []
    session.add.assert_not_called()


@pytest.mark.asyncio
async def test_a_trashed_target_is_dropped():
    """A claim needs a subject. `get` returns the row, `deleted_at` says it is
    in the trash, so there is nothing to demote."""
    trashed = MagicMock()
    trashed.deleted_at = "2026-08-08"
    session = _session(scalars_sequence=[[]], get_returns=[trashed])
    with patch("grimoire.services.supersession.access.can_write_note",
               AsyncMock(return_value=True)), \
         patch("grimoire.services.supersession.async_session", return_value=session):
        assert await supersession.set_supersedes(7, 1, [2]) == []


@pytest.mark.asyncio
async def test_a_direct_cycle_is_refused():
    """B already supersedes A; A may not now supersede B.

    Walk from the proposed target (B) and see whether it reaches the proposer
    (A). It does — B -> A — so the edge would close a ring.
    """
    session = _session(scalars_sequence=[[1]])   # B supersedes A(=1)
    with patch("grimoire.services.supersession.async_session", return_value=session):
        assert await supersession._closes_a_cycle(session, 1, 2) is True


@pytest.mark.asyncio
async def test_an_indirect_cycle_is_refused():
    """A -> B -> C exists; C may not supersede A.

    Walking from A follows A -> B, then B -> C... and the walk must reach the
    proposer. Here the proposer is C and the target is A, so: A -> B -> C.
    """
    session = _session(scalars_sequence=[[2], [3]])   # A->B, B->C
    with patch("grimoire.services.supersession.async_session", return_value=session):
        assert await supersession._closes_a_cycle(session, 3, 1) is True


@pytest.mark.asyncio
async def test_a_chain_that_does_not_loop_is_allowed():
    """A -> B exists; C may supersede A. Walking from A reaches only B."""
    session = _session(scalars_sequence=[[2], []])
    with patch("grimoire.services.supersession.async_session", return_value=session):
        assert await supersession._closes_a_cycle(session, 3, 1) is False


@pytest.mark.asyncio
async def test_the_cycle_walk_terminates_on_an_existing_ring():
    """Defensive: if a ring somehow exists (written directly to the DB), the
    walk must not spin. The visited set is what guarantees it, and this pins
    that guarantee rather than trusting it."""
    # 1 -> 2, 2 -> 1: a ring that does not contain the proposer (99).
    session = _session(scalars_sequence=[[2], [1], []])
    with patch("grimoire.services.supersession.async_session", return_value=session):
        assert await supersession._closes_a_cycle(session, 99, 1) is False


@pytest.mark.asyncio
async def test_superseded_ids_is_empty_for_an_empty_candidate_set():
    """Ranking calls this per query. An empty candidate set must not become a
    `WHERE id IN ()`, which Postgres accepts and every reader misreads."""
    assert await supersession.superseded_ids([]) == set()


@pytest.mark.asyncio
async def test_reads_are_empty_when_the_caller_cannot_read_the_note():
    with patch("grimoire.services.supersession.access.can_read_note",
               AsyncMock(return_value=False)):
        assert await supersession.get_relations(7, 1) == {
            "supersedes": [], "superseded_by": []
        }


@pytest.mark.asyncio
async def test_get_relations_partitions_both_directions_from_one_query():
    """ONE round trip for both directions, because this runs on every note read.

    Note 5 supersedes 2 and 3, and is itself superseded by 9. All four rows come
    back from a single OR query and are partitioned by which column holds 5.
    """
    session = make_mock_session()
    result = MagicMock()
    result.all.return_value = [(5, 3), (5, 2), (9, 5)]   # (superseder, superseded)
    session.execute = AsyncMock(return_value=result)

    with patch("grimoire.services.supersession.access.can_read_note",
               AsyncMock(return_value=True)), \
         patch("grimoire.services.supersession.async_session", return_value=session):
        rel = await supersession.get_relations(7, 5)

    assert rel == {"supersedes": [2, 3], "superseded_by": [9]}
    assert session.execute.await_count == 1, (
        "both directions must come from one query — asking separately doubles "
        "the round trips on the hottest path in the product"
    )