FabledGrimoire/tests/test_rule_moment_judgments.py

"""Moment proposals and judgments — what a mock can see.

The behaviour against real tables is tests/test_integration_rule_moment_judgments.py.
These pin the pieces that need no database: that the migration's CHECKs and
the model agree, which moments count as evidence, and that the line
the signal hands a session names the tool that answers it.
"""
from __future__ import annotations

import importlib.util
from datetime import datetime, timedelta, timezone
from pathlib import Path
from types import SimpleNamespace

from grimoire.models.rule_moment_judgment import JUDGMENT_STATES, NO_MOMENT, SOURCES
from grimoire.services import moments as moments_svc
from grimoire.services import rule_moment_judgments as svc

ROOT = Path(__file__).resolve().parents[1]


def _migration():
    path = ROOT / "alembic" / "versions" / "0118_rule_moment_judgments.py"
    spec = importlib.util.spec_from_file_location("m0118", path)
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    return module


def test_the_migration_checks_and_the_model_agree():
    m = _migration()
    assert tuple(m._STATES) == tuple(JUDGMENT_STATES)
    assert tuple(m._SOURCES) == tuple(SOURCES)


def test_verdicts_name_real_states():
    assert set(svc.VERDICTS.values()) <= set(JUDGMENT_STATES)


def test_no_moment_is_not_a_mountable_name():
    """"" is the "none fits" answer; if it ever became mountable, a pass's
    "no moment" would read as a mount on nothing."""
    assert not moments_svc.is_moment(NO_MOMENT)


def test_the_ubiquitous_moments_are_not_evidence():
    assert svc.signal_moments(["work.run", "work.verify", "work.change", "work.verify",
                               "not.a.moment", "skill.brainstorming"]) == [
        "work.verify", "skill.brainstorming",
    ]
    # Every skipped name is a real moment — a typo here would skip nothing.
    assert all(moments_svc.is_moment(m) for m in svc.SIGNAL_SKIP)


def test_the_proposal_line_names_the_tool_that_answers_it():
    rule = SimpleNamespace(id=11, title="Definition of done", kind="rule")
    line = svc._proposal_line(rule, "work.verify", {"situations": ["a", "b", "c"]})
    assert line.startswith("> Rule #11")
    assert "`work.verify`" in line and "3 distinct sessions" in line
    assert '"rule_id": 11' in line and '"moment": "work.verify"' in line
    assert "judge_rule_moments" in line and '"reject"' in line


def test_the_proposal_line_is_a_question_to_check_not_advice_to_relay():
    rule = SimpleNamespace(id=11, title="Definition of done", kind="rule")
    evidence = {"situations": ["a", "b", "c"],
                "reached_by": {"git push": 3, "git push --tags": 1}}
    line = svc._proposal_line(rule, "work.deliver", evidence)
    assert "co-occurrence, not a recommendation" in line
    assert "`list_moments`" in line and "if it is about a subject, leave it" in line
    assert "(reached by `git push`, `git push --tags`)" in line
    note = svc._signal_note("work.deliver", evidence)
    assert note.startswith("Opened just after `work.deliver` (reached by `git push`")
    assert "3 distinct sessions" in note


def test_the_actions_a_signal_remembers_are_bounded():
    ev: dict = {}
    for i in range(svc._SIGNAL_ACTIONS_CAP + 3):
        ev = svc._add_reached(ev, f"cmd {i}")
    ev = svc._add_reached(ev, "cmd 0")
    assert len(ev["reached_by"]) == svc._SIGNAL_ACTIONS_CAP
    assert ev["reached_by"]["cmd 0"] == 2
    assert svc._add_reached({}, "")["reached_by"] == {}


# ── The misfire ──────────────────────────────────────────────────────────────


def test_a_day_is_one_situation_and_days_are_distinct():
    """The day arm keys verbatim: tokenised, "10" and "05" fall under the
    minimum token length and every day of a year would be one situation."""
    a = datetime(2026, 10, 5, 1, tzinfo=timezone.utc)
    assert svc.misfire_situation(a) == svc.misfire_situation(a + timedelta(hours=20))
    assert svc.misfire_situation(a) != svc.misfire_situation(a + timedelta(days=1))
    assert svc.misfire_situation(a) != svc.misfire_situation(a + timedelta(days=31))


def test_a_misfire_counts_reasons_and_actions():
    now = datetime(2026, 10, 5, tzinfo=timezone.utc)
    mf = None
    for i in range(svc.MISFIRE_REASONS + 2):
        mf = svc.add_misfire(mf, svc.misfire_situation(now + timedelta(days=i)), 7, now,
                             why=f"reason {i}", reached_by="git push" if i % 2 else "make ship")
    assert len(mf["situations"]) == svc.MISFIRE_REASONS + 2
    assert [r["why"] for r in mf["reasons"]][-1] == f"reason {svc.MISFIRE_REASONS + 1}"
    assert len(mf["reasons"]) == svc.MISFIRE_REASONS
    assert mf["reached_by"] == {"make ship": 4, "git push": 3}
    assert mf["projects"] == [7]


def test_the_unmount_line_offers_both_fixes():
    rule = SimpleNamespace(id=10, title="CI verifies", kind="rule")
    mf = {"situations": ["a", "b", "c"], "reached_by": {"actions_run_read": 3}}
    line = svc._unmount_line(rule, "work.verify", mf)
    assert line.startswith("> Rule #10") and "3 distinct days" in line
    assert "`actions_run_read` ×3" in line
    assert '"verdict": "reject"' in line and "unmap_action" in line and '"confirm"' in line


def test_a_held_reply_names_the_misfire_call_only_for_a_mount():
    from grimoire.services.moment_delivery import reply_hold_reason

    mounted = {"rule_id": 11, "title": "Definition of done", "moment": "reply.report", "trigger": ""}
    scored = {"rule_id": 12, "title": "Other", "moment": "", "score": 0.81, "trigger": ""}
    assert 'rule_misfired(11, "reply.report", why)' in reply_hold_reason([mounted])
    assert "rule_misfired" not in reply_hold_reason([scored])