FabledGrimoire/tests/test_moment_actions.py

"""Which actions reach which moment — the pure half.

Matching and resolution take no database: the defaults are code and an
install's mappings arrive as rows, so every case below hands `resolve` the
rows it needs. The writes that store those rows are exercised against real
Postgres in `test_integration_moment_mappings.py`.
"""
from types import SimpleNamespace
from unittest.mock import AsyncMock, patch

import pytest

from grimoire.services import moment_actions as ma
from grimoire.services import moments


def _row(tool, match, moment, effect=ma.ADD):
    return SimpleNamespace(tool=tool, match=match, moment=moment, effect=effect)


def _reached(tool, tool_input=None, mappings=()):
    return [h["moment"] for h in ma.resolve(tool, tool_input, mappings)]


# ── the defaults ─────────────────────────────────────────────────────────

def test_every_default_reaches_a_real_moment():
    """A default naming a moment that is not in the catalog would never be
    mounted on — the typo require_moment refuses at the door."""
    bad = [a for a in ma.DEFAULT_ACTIONS if not moments.is_moment(a.moment)]
    assert not bad, bad


def test_no_default_is_listed_twice():
    keys = [(ma.tool_key(a.tool), a.match, a.moment) for a in ma.DEFAULT_ACTIONS]
    assert len(keys) == len(set(keys))


def test_every_default_is_a_mapping_the_door_would_accept():
    """The defaults pass the same validation an install's mapping does, so a
    default and an install's copy of it can never disagree about its shape."""
    for a in ma.DEFAULT_ACTIONS:
        assert ma._clean(a.tool, a.match, a.moment) == (a.tool, a.match, a.moment)


# ── matching ─────────────────────────────────────────────────────────────

@pytest.mark.parametrize("command", [
    "git push origin dev",
    "git push",
    "cd app && git push",
    "make lint; git push --tags",
    "GIT_TRACE=1 git push",
    "  git   push  ",
])
def test_a_command_prefix_matches_any_segment_of_the_line(command):
    assert "work.deliver" in _reached("Bash", {"command": command})


@pytest.mark.parametrize("command", ["git pushd", "echo git push", "git status"])
def test_a_command_prefix_needs_a_word_boundary_and_the_head_of_a_segment(command):
    assert "work.deliver" not in _reached("Bash", {"command": command})


def test_one_action_reaches_every_moment_it_is():
    """`kubectl apply` is a run, a deliver and a reach outside the workspace."""
    assert _reached("Bash", {"command": "kubectl apply -f deploy.yaml"}) == [
        "work.run", "work.deliver", "env.reach",
    ]


def test_the_most_specific_action_is_the_one_named():
    hits = ma.resolve("Bash", {"command": "git push"})
    deliver = next(h for h in hits if h["moment"] == "work.deliver")
    assert deliver["match"] == "git push"
    run = next(h for h in hits if h["moment"] == "work.run")
    assert run["match"] == ""


def test_an_ordinary_command_is_only_a_run():
    assert _reached("Bash", {"command": "ls -la"}) == ["work.run"]


def test_an_unknown_tool_reaches_nothing():
    assert _reached("SomeOtherTool", {"x": 1}) == []


@pytest.mark.parametrize("name", [
    "mcp__plugin_grimoire_grimoire__update_task", "mcp__grimoire__update_task",
    "update_task", "Update_Task",
])
def test_the_server_prefix_and_case_do_not_matter(name):
    assert _reached(name, {"task_id": 1, "status": "done"}) == ["work.finish"]


@pytest.mark.parametrize("status,expected", [
    ("done", ["work.finish"]),
    ("cancelled", ["work.finish"]),
    ("in_progress", ["work.start"]),
    ("todo", []),
    ("", []),
])
def test_arguments_select_the_moment(status, expected):
    assert _reached("update_task", {"task_id": 1, "status": status}) == expected


def test_a_whole_tool_mapping_ignores_its_arguments():
    assert _reached("Edit", {"file_path": "x", "old_string": "a"}) == ["work.change"]


def test_loading_a_procedure_reaches_its_own_moment():
    assert _reached("Skill", {"skill": "release-notes"}) == ["skill.release-notes"]


def test_a_bundled_skill_also_reaches_the_moment_it_is_for():
    """Loading the planning procedure IS planning: a rule mounted on work.plan
    arrives on the load, not only on the plan it goes on to open."""
    assert _reached("Skill", {"skill": "grimoire:writing-plans"}) == [
        "work.plan", "skill.grimoire:writing-plans",
    ]
    assert _reached("Skill", {"skill": "grimoire:reporting-back"})[0] == "reply.report"


def test_only_the_products_own_skill_reaches_a_bundled_default():
    """Another plugin's `verification` skill is not the product's procedure;
    the plugin's name is what makes the declaration ours."""
    assert _reached("Skill", {"skill": "other:verification"}) == ["skill.other:verification"]


def test_a_procedure_name_that_is_not_a_moment_reaches_nothing():
    assert _reached("Skill", {"skill": "two words"}) == []
    assert _reached("Skill", {}) == []


# ── a stored process's own moments ───────────────────────────────────────

def test_a_process_declares_moments_its_load_reaches():
    hits = ma.resolve("Skill", {"skill": "grimoire-proc-release"}, [], ["work.deliver"])
    assert [(h["moment"], h["via"]) for h in hits] == [
        ("work.deliver", ma.PROCESS), ("skill.grimoire-proc-release", ma.DEFAULT),
    ]
    assert hits[0]["match"] == "skill=grimoire-proc-release"


def test_a_declared_moment_another_action_reaches_is_named_once():
    rows = [_row("Skill", "skill=grimoire-proc-release", "work.deliver")]
    hits = ma.resolve("Skill", {"skill": "grimoire-proc-release"}, rows, ["work.deliver"])
    assert [h["moment"] for h in hits].count("work.deliver") == 1


def test_declared_moments_only_reach_a_skill_load():
    hits = ma.resolve("Bash", {"command": "ls"}, [], ["work.deliver"])
    assert [h["moment"] for h in hits] == ["work.run"]


@pytest.mark.parametrize("data, expected", [
    (None, []),
    ({}, []),
    ({"moments": ["work.deliver", "Work.Deliver", "reply.report"]}, ["work.deliver", "reply.report"]),
    ({"moments": "work.verify"}, ["work.verify"]),
    ({"moments": ["work.finished", "skill.two words", "work.debug"]}, ["work.debug"]),
])
def test_a_process_field_is_read_defensively(data, expected):
    """A moment the catalog no longer has stops firing; it does not break the load."""
    assert ma.declared_moments(data) == expected


async def test_only_a_process_skill_asks_for_its_process():
    asked = AsyncMock(return_value=["work.deliver"])
    with patch.object(ma, "list_mappings", AsyncMock(return_value=[])), \
         patch.object(ma, "process_moments", asked):
        reached = await ma.moments_for(1, "Skill", {"skill": "grimoire-proc-release"})
        await ma.moments_for(1, "Skill", {"skill": "grimoire:verification"})
        await ma.moments_for(1, "Bash", {"command": "git push"})
    assert "work.deliver" in [h["moment"] for h in reached]
    asked.assert_awaited_once_with(1, "grimoire-proc-release")


async def _process_moments(manifest, note=None, *, raises=False):
    from grimoire.services import notes as notes_svc
    from grimoire.services import plugin_context

    built = AsyncMock(side_effect=RuntimeError("down")) if raises else AsyncMock(return_value=manifest)
    loader = AsyncMock(return_value=(note, None) if note is not None else None)
    with patch.object(plugin_context, "build_process_manifest", built), \
         patch.object(notes_svc, "get_note_for_user", loader):
        return await ma.process_moments(1, "grimoire-proc-release"), loader


async def test_a_process_slug_resolves_through_the_sync_manifest():
    """The same manifest the sync script wrote the skill from, so the slug
    names the same process here as it did on disk."""
    note = SimpleNamespace(data={"moments": ["work.deliver"]})
    found, loader = await _process_moments(
        {"processes": [{"id": 7, "slug": "other"}, {"id": 9, "slug": "release"}]}, note)
    assert found == ["work.deliver"]
    loader.assert_awaited_once_with(1, 9)


async def test_an_unknown_or_unreadable_process_reaches_nothing_more():
    found, loader = await _process_moments({"processes": [{"id": 7, "slug": "other"}]})
    assert found == []
    loader.assert_not_awaited()
    found, _ = await _process_moments({}, raises=True)
    assert found == []


# ── an install's own ─────────────────────────────────────────────────────

def test_an_install_mapping_adds_a_moment():
    rows = [_row("Bash", "make ship", "work.deliver")]
    assert "work.deliver" in _reached("Bash", {"command": "make ship"}, rows)
    assert "work.deliver" not in _reached("Bash", {"command": "make ship"})


def test_removing_a_default_removes_only_that_default():
    """`curl` stops reaching env.reach here; it is still a run, and `ssh`
    still reaches out — a removal overrides nothing it did not name."""
    rows = [_row("Bash", "curl", "env.reach", ma.REMOVE)]
    assert _reached("Bash", {"command": "curl localhost:8000"}, rows) == ["work.run"]
    assert "env.reach" in _reached("Bash", {"command": "ssh box"}, rows)


def test_a_removal_matches_the_default_whatever_the_tool_is_called():
    rows = [_row("mcp__x__update_task", "status=done", "work.finish", ma.REMOVE)]
    assert _reached("update_task", {"status": "done"}, rows) == []


def test_an_install_mapping_is_named_as_the_installs():
    rows = [_row("Bash", "make ship", "work.deliver")]
    hit = next(h for h in ma.resolve("Bash", {"command": "make ship"}, rows)
               if h["moment"] == "work.deliver")
    assert (hit["via"], hit["match"]) == (ma.INSTALL, "make ship")


# ── validation ───────────────────────────────────────────────────────────

def test_a_prefix_on_a_tool_that_runs_no_command_is_refused():
    """It would be stored and never match — a mapping that looks made and is
    not, which is the misfire this table exists to fix."""
    with pytest.raises(ValueError, match="field=value"):
        ma._clean("update_task", "done", "work.finish")


def test_an_unknown_moment_is_refused():
    with pytest.raises(ValueError, match="unknown moment"):
        ma._clean("Bash", "make ship", "work.ship")


def test_a_mapping_is_normalised_before_it_is_stored():
    assert ma._clean("mcp__x__update_task", " status=done ", " Work.Finish") == (
        "update_task", "status=done", "work.finish",
    )
    assert ma._clean("Bash", "make   ship", "work.deliver")[1] == "make ship"


def test_a_tool_is_required():
    with pytest.raises(ValueError, match="tool is required"):
        ma._clean("  ", "", "work.run")


# ── failing open ─────────────────────────────────────────────────────────

async def test_an_unreadable_mapping_table_costs_the_corrections_not_the_moments():
    with patch.object(ma, "list_mappings", AsyncMock(side_effect=RuntimeError("db down"))):
        hits = await ma.moments_for(7, "Bash", {"command": "git push"})
    assert [h["moment"] for h in hits] == ["work.run", "work.deliver"]