FabledGrimoire/tests/test_family.py

"""The promotion engine without a database.

The parts that decide — which citations claim lineage, which criteria veto —
are pure and pinned here, beside the doors' wiring: the triggers ride the
write tools, fail open, and the criteria the service enforces are the ones
the agent is told. The state machine against Postgres is in
tests/test_integration_family_promotion.py.
"""
from __future__ import annotations

from types import SimpleNamespace
from unittest.mock import AsyncMock, patch

import pytest

from grimoire.mcp.server import _READ_ONLY_TOOLS, _WRITE_TOOLS
from grimoire.mcp.tools import family as family_tools
from grimoire.mcp.tools.milestones import update_milestone
from grimoire.services import family as family_svc
from grimoire.services.family import CRITERIA_KEYS, lineage_citations, vetoes
from tests.helpers import fake_milestone

pytestmark = pytest.mark.usefixtures("_bind_user")


# --- which citations claim lineage ---------------------------------------------

@pytest.mark.parametrize("text", [
    "Signed release lane, matching roundtable-android (task #1615).",
    "Ported from #1615, with the keystore step moved first.",
    "Same shape as #1615: the APK is baked into the image.",
    "This mirrors #1615 for the desktop client.",
    "Modelled on #1615.",
])
def test_a_citation_with_a_lineage_word_names_its_source(text):
    assert lineage_citations(text) == [1615]


@pytest.mark.parametrize("text", [
    "See #1615 for context.",
    "#1615 is related.",
    "Closes #1615.",
    # The lineage word is in the PREVIOUS sentence — a different claim.
    "This is matching the old flow. Unrelated: #1615.",
    "Matching the old flow\nsee #1615",
    "",
])
def test_a_citation_without_lineage_is_silent(text):
    assert lineage_citations(text) == []


def test_lineage_citations_are_in_order_and_unique():
    text = "Based on #20; ported from #10. Same pattern as #20."
    assert lineage_citations(text) == [20, 10]


# --- each criterion vetoes on its own -----------------------------------------------

GOOD = dict(
    applies_when="any app that installs its own updates",
    platforms=["android-app"],
    criteria={
        "platform_terms": "stated as an Android distribution concern",
        "platform_problem": "Play Protect and signature checks are the platform's",
        "proven": "shipped in two apps",
    },
    evidence=["CI run 8274 green", "#4775"],
)


def test_all_three_criteria_held_means_no_veto():
    assert vetoes(**GOOD) == []


@pytest.mark.parametrize("key", CRITERIA_KEYS)
def test_each_criterion_left_unsupported_vetoes_on_its_own(key):
    case = {**GOOD, "criteria": {**GOOD["criteria"], key: "   "}}
    assert vetoes(**case) == [key]


def test_platform_terms_needs_an_applicability_test_and_a_scope():
    assert vetoes(**{**GOOD, "applies_when": ""}) == ["platform_terms"]
    assert vetoes(**{**GOOD, "platforms": []}) == ["platform_terms"]


def test_proven_needs_named_evidence():
    assert vetoes(**{**GOOD, "evidence": []}) == ["proven"]
    assert vetoes(**{**GOOD, "evidence": ["  "]}) == ["proven"]


def test_the_criteria_the_agent_is_told_are_the_ones_enforced():
    """The criteria are product text. The promote tool's docstring
    must name every criterion the service checks, by its parameter name."""
    doc = family_tools.promote_family_idea.__doc__
    for key in CRITERIA_KEYS:
        assert key in doc, f"promote_family_idea's docstring never names {key}"
    assert len(family_svc.CRITERIA) == 3


# --- the doors ----------------------------------------------------------------------

def test_every_family_tool_is_classified():
    reads = {"list_family_ideas", "get_family_idea", "list_family_decisions"}
    writes = {"propose_family_idea", "promote_family_idea", "retire_family_idea",
              "undo_family_decision"}
    assert reads <= _READ_ONLY_TOOLS
    assert writes <= _WRITE_TOOLS


def _note(**kw):
    base = dict(id=7, project_id=2, title="t", body="b", is_task=False)
    return SimpleNamespace(**{**base, **kw})


async def test_a_citation_hint_wins_and_repeat_is_not_asked():
    data: dict = {}
    repeat = AsyncMock(return_value="repeat")
    with patch.object(family_svc, "citation_trigger", AsyncMock(return_value="cited")), \
         patch.object(family_svc, "repeat_trigger", repeat):
        await family_svc.attach_family_hint(1, data, _note(), created=True)
    assert data == {"family_hint": "cited"}
    repeat.assert_not_awaited()


async def test_the_repeat_check_runs_only_on_a_create():
    repeat = AsyncMock(return_value="repeat")
    with patch.object(family_svc, "citation_trigger", AsyncMock(return_value=None)), \
         patch.object(family_svc, "repeat_trigger", repeat):
        edited: dict = {}
        await family_svc.attach_family_hint(1, edited, _note(), created=False)
        created: dict = {}
        await family_svc.attach_family_hint(1, created, _note(), created=True)
    assert edited == {}
    assert created == {"family_hint": "repeat"}


async def test_a_failing_trigger_never_breaks_the_write():
    data: dict = {"id": 7}
    with patch.object(family_svc, "citation_trigger", AsyncMock(side_effect=RuntimeError("db down"))):
        await family_svc.attach_family_hint(1, data, _note(), created=True)
    assert data == {"id": 7}


async def test_closing_a_milestone_on_a_platform_carries_the_hint():
    closed = fake_milestone(id=5, project_id=3, status="done")
    with patch("grimoire.mcp.tools.milestones.milestones_svc.update_milestone",
               AsyncMock(return_value=closed)), \
         patch.object(family_svc, "milestone_is_open", AsyncMock(return_value=True)), \
         patch.object(family_svc, "milestone_trigger", AsyncMock(return_value="evaluate")):
        out = await update_milestone(project_id=3, milestone_id=5, status="done")
    assert out["family_hint"] == "evaluate"


async def test_re_saving_a_closed_milestone_asks_nothing():
    closed = fake_milestone(id=5, project_id=3, status="done")
    trigger = AsyncMock(return_value="evaluate")
    with patch("grimoire.mcp.tools.milestones.milestones_svc.update_milestone",
               AsyncMock(return_value=closed)), \
         patch.object(family_svc, "milestone_is_open", AsyncMock(return_value=False)), \
         patch.object(family_svc, "milestone_trigger", trigger):
        out = await update_milestone(project_id=3, milestone_id=5, status="done")
    assert "family_hint" not in out
    trigger.assert_not_awaited()