FabledScribe/tests/test_services_rulebooks.py

"""Tests for services/rulebooks.py — mocks async_session, no real DB.

Mirrors the pattern in tests/test_events_service.py.
"""
from datetime import datetime, timezone
from unittest.mock import AsyncMock, MagicMock, patch

import pytest
from tests.helpers import fake_rule, fake_rulebook, fake_topic, make_mock_session


@pytest.fixture(autouse=True)
def _no_exclusions():
    """get_applicable_rules asks for the project's always-on exclusions
    (milestone 297) through its own session; these mocked-session tests
    script the rule queries only, so the exclusions lookup is stubbed empty."""
    if True:
        yield


@pytest.mark.asyncio
async def test_create_rulebook_stores_to_db():
    mock_session = make_mock_session()
    with patch("scribe.services.rulebooks.async_session") as mock_cls:
        mock_cls.return_value = mock_session
        from scribe.services.rulebooks import create_rulebook
        await create_rulebook(
            user_id=7, title="FabledSword family", description="rules for the family",
        )
    assert mock_session.add.called
    assert mock_session.commit.called


@pytest.mark.asyncio
async def test_list_rulebooks_returns_owned_only():
    rb = fake_rulebook(id=1)
    mock_session = make_mock_session()
    mock_result = MagicMock()
    mock_result.scalars.return_value.all.return_value = [rb]
    mock_session.execute = AsyncMock(return_value=mock_result)
    with patch("scribe.services.rulebooks.async_session") as mock_cls:
        mock_cls.return_value = mock_session
        from scribe.services.rulebooks import list_rulebooks
        results = await list_rulebooks(user_id=7)
    assert len(results) == 1


@pytest.mark.asyncio
async def test_get_rulebook_returns_none_when_not_owner():
    """get_rulebook scopes by owner_user_id — wrong user gets None."""
    mock_session = make_mock_session()
    mock_result = MagicMock()
    mock_result.scalar_one_or_none.return_value = None
    mock_session.execute = AsyncMock(return_value=mock_result)
    with patch("scribe.services.rulebooks.async_session") as mock_cls:
        mock_cls.return_value = mock_session
        from scribe.services.rulebooks import get_rulebook
        result = await get_rulebook(rulebook_id=1, user_id=99)
    assert result is None


@pytest.mark.asyncio
async def test_update_rulebook_only_sets_provided_fields():
    rb = fake_rulebook(id=1, title="old")
    mock_session = make_mock_session()
    mock_result = MagicMock()
    mock_result.scalar_one_or_none.return_value = rb
    mock_session.execute = AsyncMock(return_value=mock_result)
    with patch("scribe.services.rulebooks.async_session") as mock_cls:
        mock_cls.return_value = mock_session
        from scribe.services.rulebooks import update_rulebook
        await update_rulebook(rulebook_id=1, user_id=7, title="new")
    assert rb.title == "new"


@pytest.mark.asyncio
async def test_delete_rulebook_calls_delete():
    rb = fake_rulebook(id=1)
    mock_session = make_mock_session()
    mock_result = MagicMock()
    mock_result.scalar_one_or_none.return_value = rb
    mock_session.execute = AsyncMock(return_value=mock_result)
    mock_session.delete = AsyncMock()
    with patch("scribe.services.rulebooks.async_session") as mock_cls:
        mock_cls.return_value = mock_session
        from scribe.services.rulebooks import delete_rulebook
        await delete_rulebook(rulebook_id=1, user_id=7)
    assert mock_session.delete.called


# ── Topic CRUD ───────────────────────────────────────────────────────────

@pytest.mark.asyncio
async def test_create_topic_requires_owned_rulebook():
    """create_topic raises ValueError if the rulebook isn't owned by user."""
    mock_session = make_mock_session()
    mock_result = MagicMock()
    mock_result.scalar_one_or_none.return_value = None
    mock_session.execute = AsyncMock(return_value=mock_result)
    with patch("scribe.services.rulebooks.async_session") as mock_cls:
        mock_cls.return_value = mock_session
        from scribe.services.rulebooks import create_topic
        with pytest.raises(ValueError, match="not found"):
            await create_topic(
                rulebook_id=999, user_id=7, title="git-workflow",
            )


@pytest.mark.asyncio
async def test_list_topics_returns_topics_for_owned_rulebook():
    rb = fake_rulebook(id=1)
    topic = fake_topic(id=10, rulebook_id=1, title="git-workflow")

    # Two execute calls: ownership check, then topic select.
    mock_session = make_mock_session()
    rb_result = MagicMock()
    rb_result.scalar_one_or_none.return_value = rb
    topic_result = MagicMock()
    topic_result.scalars.return_value.all.return_value = [topic]
    mock_session.execute = AsyncMock(side_effect=[rb_result, topic_result])

    with patch("scribe.services.rulebooks.async_session") as mock_cls:
        mock_cls.return_value = mock_session
        from scribe.services.rulebooks import list_topics
        results = await list_topics(rulebook_id=1, user_id=7)
    assert len(results) == 1
    assert results[0].title == "git-workflow"


# ── Rule CRUD ───────────────────────────────────────────────────────────

@pytest.mark.asyncio
async def test_create_rule_requires_owned_topic():
    mock_session = make_mock_session()
    mock_result = MagicMock()
    mock_result.scalar_one_or_none.return_value = None  # topic not found
    mock_session.execute = AsyncMock(return_value=mock_result)
    with patch("scribe.services.rulebooks.async_session") as mock_cls:
        mock_cls.return_value = mock_session
        from scribe.services.rulebooks import create_rule
        with pytest.raises(ValueError, match="topic .* not found"):
            await create_rule(
                topic_id=999, user_id=7, title="x", statement="y",
            )


@pytest.mark.asyncio
async def test_list_rules_filters_by_topic_id():
    """list_rules(topic_id=X) returns rules in that topic, ownership-scoped."""
    rule = fake_rule(id=1, topic_id=10)
    mock_session = make_mock_session()
    mock_result = MagicMock()
    mock_result.scalars.return_value.all.return_value = [rule]
    mock_session.execute = AsyncMock(return_value=mock_result)
    with patch("scribe.services.rulebooks.async_session") as mock_cls:
        mock_cls.return_value = mock_session
        from scribe.services.rulebooks import list_rules
        results = await list_rules(user_id=7, topic_id=10)
    assert len(results) == 1


@pytest.mark.asyncio
async def test_get_rule_returns_none_when_not_owner():
    mock_session = make_mock_session()
    mock_result = MagicMock()
    mock_result.scalar_one_or_none.return_value = None
    mock_session.execute = AsyncMock(return_value=mock_result)
    with patch("scribe.services.rulebooks.async_session") as mock_cls:
        mock_cls.return_value = mock_session
        from scribe.services.rulebooks import get_rule
        result = await get_rule(rule_id=1, user_id=99)
    assert result is None


# ── applicable_rules ────────────────────────────────────────────────────

def _empty():
    """A MagicMock result whose .all() returns [] (or .scalars().all() returns [])."""
    r = MagicMock()
    r.all.return_value = []
    r.scalars.return_value.all.return_value = []
    return r


def _no_edges():
    """Silence the three post-query lookups get_applicable_rules now makes.

    They are separate service functions with their own coverage (and the real
    wiring is proven against Postgres in test_integration_rule_surfacing), so
    stubbing them here keeps each of these tests about the one projection it
    was written to check — rather than about the order a mocked session's
    execute() calls happen to arrive in.
    """
    return (
        patch("scribe.services.rulebooks.co_surfaced_partners",
              AsyncMock(return_value=[])),
        patch("scribe.services.rulebooks.list_rule_relations",
              AsyncMock(return_value={})),
        patch("scribe.services.rulebooks.list_rule_systems",
              AsyncMock(return_value={})),
    )


def _areas(*canonical_ids):
    """The project-areas query's result: `.scalars().all()` of canonical ids."""
    r = MagicMock()
    r.scalars.return_value.all.return_value = list(canonical_ids)
    return r


@pytest.mark.asyncio
async def test_get_applicable_rules_returns_shape():
    """The listing is a project's own rules plus the global rules tagged to its
    areas (milestone 414) — no subscriptions or suppressions in the shape."""
    mock_session = make_mock_session()
    rules_result = MagicMock()
    rules_result.all.return_value = [
        # The query selects the ENTITY plus three labels, so rule_brief stays
        # the one place deciding what a surfaced rule carries (note 3026).
        (fake_rule(id=i, title=f"Rule {i}", statement=f"Statement {i}", topic_id=2),
         "git-workflow", 1, "FabledSword family")
        for i in range(50)
    ]
    # Execute order: project-areas_q, rules_q (only with areas), proj_rules_q
    mock_session.execute = AsyncMock(side_effect=[
        _areas(4), rules_result, _empty(),
    ])

    _p1, _p2, _p3 = _no_edges()
    with patch("scribe.services.rulebooks.async_session") as mock_cls, _p1, _p2, _p3:
        mock_cls.return_value = mock_session
        from scribe.services.rulebooks import get_applicable_rules
        result = await get_applicable_rules(project_id=3, user_id=7, limit=50)

    assert set(result) == {"rules", "project_rules", "truncated"}
    assert len(result["rules"]) == 50
    assert result["rules"][0]["topic_id"] == 2
    assert result["rules"][0]["rulebook_id"] == 1
    assert result["project_rules"] == []
    assert result["truncated"] is False


@pytest.mark.asyncio
async def test_get_applicable_rules_truncates_when_over_limit():
    """When limit+1 rows are returned, truncated=True and only `limit` returned."""
    mock_session = make_mock_session()
    rules_result = MagicMock()
    rules_result.all.return_value = [
        (fake_rule(id=i, title=f"r{i}"), "topic", 1, "rb") for i in range(51)
    ]
    mock_session.execute = AsyncMock(side_effect=[
        _areas(4), rules_result, _empty(),
    ])

    _p1, _p2, _p3 = _no_edges()
    with patch("scribe.services.rulebooks.async_session") as mock_cls, _p1, _p2, _p3:
        mock_cls.return_value = mock_session
        from scribe.services.rulebooks import get_applicable_rules
        result = await get_applicable_rules(project_id=3, user_id=7, limit=50)

    assert result["truncated"] is True
    assert len(result["rules"]) == 50


@pytest.mark.asyncio
async def test_a_project_with_no_areas_lists_no_global_rules():
    """Untagged global rules apply everywhere and arrive by retrieval; the
    listing only names the ones bound by area, so no areas means none — and
    the global-rules query is not run at all."""
    mock_session = make_mock_session()
    mock_session.execute = AsyncMock(side_effect=[_areas(), _empty()])

    _p1, _p2, _p3 = _no_edges()
    with patch("scribe.services.rulebooks.async_session") as mock_cls, _p1, _p2, _p3:
        mock_cls.return_value = mock_session
        from scribe.services.rulebooks import get_applicable_rules
        result = await get_applicable_rules(project_id=3, user_id=7)

    assert result["rules"] == []
    assert mock_session.execute.await_count == 2


@pytest.mark.asyncio
async def test_get_applicable_rules_includes_project_scoped_rules():
    """Project-scoped rules surface in the project_rules field."""
    mock_session = make_mock_session()
    proj_rules_result = MagicMock()
    proj_rules_result.all.return_value = [
        (fake_rule(id=100, topic_id=None, project_id=3, title="Use alembic",
                   statement="Always run migrations via alembic, never raw SQL."),),
        (fake_rule(id=101, topic_id=None, project_id=3, title="PR-bound",
                   statement="Land schema changes in their own PR."),),
    ]
    mock_session.execute = AsyncMock(side_effect=[_areas(), proj_rules_result])

    _p1, _p2, _p3 = _no_edges()
    with patch("scribe.services.rulebooks.async_session") as mock_cls, _p1, _p2, _p3:
        mock_cls.return_value = mock_session
        from scribe.services.rulebooks import get_applicable_rules
        result = await get_applicable_rules(project_id=3, user_id=7)

    assert len(result["project_rules"]) == 2
    assert result["project_rules"][0]["title"] == "Use alembic"
    assert result["project_rules"][1]["id"] == 101


@pytest.mark.asyncio
async def test_a_co_surfaces_partner_on_another_project_is_not_dragged_in():
    """An edge is not a way into a project: a partner that is global or on this
    project arrives; one on a different project does not."""
    mock_session = make_mock_session()
    proj_rules_result = MagicMock()
    proj_rules_result.all.return_value = [(fake_rule(id=100, topic_id=None, project_id=3),)]
    mock_session.execute = AsyncMock(side_effect=[_areas(), proj_rules_result])
    partners = [
        fake_rule(id=200, topic_id=9, project_id=None, title="global partner"),
        fake_rule(id=201, topic_id=None, project_id=3, title="same-project partner"),
        fake_rule(id=202, topic_id=None, project_id=8, title="other-project partner"),
    ]

    with patch("scribe.services.rulebooks.async_session") as mock_cls, \
         patch("scribe.services.rulebooks.co_surfaced_partners", AsyncMock(return_value=partners)), \
         patch("scribe.services.rulebooks.list_rule_relations", AsyncMock(return_value={})), \
         patch("scribe.services.rulebooks.list_rule_systems", AsyncMock(return_value={})):
        mock_cls.return_value = mock_session
        from scribe.services.rulebooks import get_applicable_rules
        result = await get_applicable_rules(project_id=3, user_id=7)

    assert [r["id"] for r in result["rules"]] == [200, 201]


# ── rule_brief (milestone 307) ──────────────────────────────────────────

def test_rule_brief_carries_age_but_not_the_deep_fields():
    """The shape a SURFACED rule takes, and the reason it exists.

    There were three hand-written copies of this dict and they had already
    diverged — none carried the timestamps the model has always held, which is
    why a rule written before the capability it duplicates was
    indistinguishable at read time from one still doing work (note 3026).
    """
    from scribe.services.rulebooks import rule_brief

    out = rule_brief(fake_rule(
        when_to_apply="before any git push",
        updated_at=datetime(2026, 6, 1, 14, 30, tzinfo=timezone.utc),
    ))
    assert out["when_to_apply"] == "before any git push"
    # A DATE, not a stamp: the question is "how old is this", and a full ISO
    # string across the always-on set is ~2k characters of payload.
    assert out["updated_at"] == "2026-06-01"
    # The depth stays with get_rule — putting it in every listing is the bloat
    # this milestone is about.
    assert "why" not in out and "how_to_apply" not in out


def test_rule_brief_omits_keys_a_rule_has_no_value_for():
    """#2483: a null key reads as a capability the record has and isn't using,
    which is a different claim from not having one."""
    from scribe.services.rulebooks import rule_brief

    out = rule_brief(fake_rule())
    assert "when_to_apply" not in out
    assert "arose_from_id" not in out




# ── verify_with / expires_when (milestone 312) ──────────────────────────

def test_a_rule_with_no_check_says_nothing_about_verification():
    """The empty case is the COMMON case, and it must stay silent.

    Most rules are decisions: they have no truth value and there is nothing to
    go and check. If a brief carried `last_verified` for those too, the signal
    would be worthless — every rule would look like something someone ought to
    be verifying, and the handful that genuinely rot would stop standing out.
    """
    from scribe.services.rulebooks import last_verified_label, rule_brief

    rule = fake_rule()
    assert last_verified_label(rule) is None
    assert "last_verified" not in rule_brief(rule)


def test_an_unverified_constraint_reads_never_rather_than_null():
    """#2483 again: a null key reads as a capability going unused. "never" is
    a different and much stronger claim — this rule asserts a fact about
    someone else's software and nobody has ever confirmed it."""
    from scribe.services.rulebooks import last_verified_label, rule_brief

    rule = fake_rule(verify_with="cat CI-runner/renovate/config.js")
    assert last_verified_label(rule) == "never"
    assert rule_brief(rule)["last_verified"] == "never"


def test_a_verified_constraint_reports_the_date_it_was_checked():
    """A date, not a stamp — the question is "how old is this", the same call
    rule_brief makes for updated_at."""
    from scribe.services.rulebooks import last_verified_label

    rule = fake_rule(
        verify_with="cat CI-runner/renovate/config.js",
        verified_at=datetime(2026, 8, 27, 11, 46, tzinfo=timezone.utc),
    )
    assert last_verified_label(rule) == "2026-08-27"


def test_the_check_text_itself_never_enters_a_listing():
    """A listing says WHICH rules can rot, not how to test them. The check can
    be a long command; multiplied across an always-on set it is the same bloat
    `why` and `how_to_apply` are kept out of a brief to avoid."""
    from scribe.services.rulebooks import rule_brief

    out = rule_brief(fake_rule(
        verify_with="a very long command " * 20,
        expires_when="the runner learns a new shell",
    ))
    assert "verify_with" not in out
    assert "expires_when" not in out


# ── the sweep's row shape (milestone 312 step 3) ────────────────────────

def test_a_sweep_row_carries_the_check_in_full():
    """The OPPOSITE call from rule_brief, and deliberately so.

    A listing omits the depth because nobody reading it wants to act on one
    rule. A sweep row exists to be acted on — the reader is about to go and
    run the check — so the text is the payload's point, not its bloat.
    """
    from scribe.services.rulebooks import verification_row

    row = verification_row(fake_rule(
        verify_with="cat CI-runner/renovate/config.js",
        expires_when="dependencyDashboardApproval is turned off",
        when_to_apply="when a dependency bump is in play",
    ))
    assert row["verify_with"] == "cat CI-runner/renovate/config.js"
    assert row["expires_when"] == "dependencyDashboardApproval is turned off"
    assert row["when_to_apply"] == "when a dependency bump is in play"


def test_never_verified_reports_no_day_count_rather_than_zero():
    """"Never" is not "0 days ago" — the second reads as freshly checked.

    Getting this wrong would invert the row's meaning for exactly the rules
    that most need attention.
    """
    from scribe.services.rulebooks import verification_row

    row = verification_row(fake_rule(verify_with="read the workflow"))
    assert row["last_verified"] == "never"
    assert row["days_since_verified"] is None


def test_a_verified_row_counts_the_days():
    from datetime import timedelta

    from scribe.services.rulebooks import verification_row

    row = verification_row(fake_rule(
        verify_with="read the workflow",
        verified_at=datetime.now(timezone.utc) - timedelta(days=74, hours=1),
    ))
    assert row["days_since_verified"] == 74