FabledGrimoire/tests/conftest.py

"""
Shared pytest fixtures.

Integration tests that need a real database should use a separate PostgreSQL
instance (e.g. a Docker service spun up by the CI job) and set DATABASE_URL
in the environment before importing the app.

For unit tests of pure functions no database is needed at all.

The fixtures below are the ONE definition of things a dozen test modules
each need. They are deliberately not
autouse: a module opts in with
``pytestmark = pytest.mark.usefixtures("<name>")`` (or a test names the
fixture as a parameter), so a unit test that never touches the engine or the
MCP context pays nothing for them.
"""
import os
from unittest.mock import AsyncMock, patch

import pytest
import pytest_asyncio


@pytest.fixture(autouse=True)
def _isolate_env(request, monkeypatch):
    """Prevent unit tests from accidentally reading production env vars.

    Integration tests (marked `integration`) are skipped here: they must use the
    real DATABASE_URL injected by the CI integration lane, not the fake one.
    """
    if request.node.get_closest_marker("integration"):
        return
    monkeypatch.setenv("DATABASE_URL", "postgresql+asyncpg://test:test@localhost/test")
    monkeypatch.setenv("SECRET_KEY", "test-secret-key")
    monkeypatch.setenv("OLLAMA_URL", "http://localhost:11434")


@pytest.fixture
def _bind_user():
    """Bind MCP caller #7 for the duration of a test.

    The MCP tool layer reads the caller from a ContextVar the HTTP transport
    sets per request; a unit test of a tool has no request, so it binds the
    caller itself. Every tool-layer test module opts in with
    ``pytestmark = pytest.mark.usefixtures("_bind_user")`` and builds its fakes
    with user_id=7 so ownership checks see the caller as the owner.
    """
    from scribe.mcp._context import _user_id_ctx

    token = _user_id_ctx.set(7)
    yield
    _user_id_ctx.reset(token)


@pytest_asyncio.fixture
async def _dispose_engine():
    """Dispose the app's module-level engine after a test that hit Postgres.

    The engine pools asyncpg connections per event loop, but pytest-asyncio
    runs each test on a fresh loop — so without this, test 2 gets handed
    test 1's connection bound to a now-dead loop ("Future attached to a
    different loop"). Disposing in the test's own loop teardown clears the
    pool cleanly. The import is deferred so merely collecting a module that
    mixes unit and integration tests never builds an engine.
    """
    from scribe.models import engine

    yield
    await engine.dispose()


@pytest.fixture
def _no_embedding():
    """Stub the fire-and-forget embedding refresh a note write detaches.

    For integration tests about ids, transactions and access rather than
    recall: `embed_note` spawns a task that loads the embedding model, which
    outlives the test's event loop and makes the lane slower for nothing. Opt
    in alongside `_dispose_engine`.
    """
    from unittest.mock import MagicMock

    # Milestones embed too; a plan created in a test would
    # otherwise detach the same model-loading task.
    with patch("scribe.services.notes.embed_note", MagicMock()), \
         patch("scribe.services.milestones.embed_milestone", MagicMock()):
        yield


@pytest.fixture
def _no_supersession():
    """Stub the auto-inject menu's "which lines are superseded?" lookup.

    That is a real database call on a path the plugin-context tests exercise
    without one. Stubbed to "nothing superseded" — the ordinary state — rather
    than hidden behind a try/except in the product, which would make the code
    lie about what it does. The label's own behaviour is covered in
    tests/test_supersession_ranking.py.
    """
    with patch("scribe.services.plugin_context.superseded_ids",
               AsyncMock(return_value=set())):
        yield


@pytest.fixture(autouse=True)
def _no_reply_shape_log():
    """Stub the reply-shape delivery row.

    Autouse for _no_system_labels' reason: every turn's retrieval, every
    moment request and every Scribe tool that reaches a shaped moment writes
    one AppLog row, and those paths are tested without a database.
    tests/test_reply_shapes.py binds the real function at import time,
    before this patch runs.
    """
    with patch("scribe.services.reply_shapes.record_delivery", AsyncMock()):
        yield


@pytest.fixture(autouse=True)
def _no_system_labels():
    """Stub the menu's "which System is each line about?" lookup.

    Autouse because every test that renders an injected menu reaches it, and
    it is a real database call on a path those tests run without one. Stubbed
    to "no labels", the state of any untagged record. Tests of the label
    itself patch it with a value.
    """
    with patch("scribe.services.plugin_context.system_names_for",
               AsyncMock(return_value={})):
        yield


@pytest.fixture(autouse=True)
def _no_rulings_arm():
    """Stub the rulings arm both PreToolUse builders run.

    Autouse for _no_system_labels' reason: every write-path and tool-rule
    test with a project in scope reaches it, and its first act is reading the
    project's Systems from Postgres. Stubbed to "no area's files touched".
    tests/test_system_rulings.py binds the real function at import time,
    before this patch runs.
    """
    with patch("scribe.services.system_rulings.rulings_for_paths",
               AsyncMock(return_value={"lines": [], "system_ids": []})):
        yield


@pytest.fixture(autouse=True)
def _no_system_usage_readout():
    """Stub the per-System usage lookup the Systems list and get_system carry.

    Autouse for _no_rulings_arm's reason: every route and tool test that
    reads a System now reaches a GROUP BY on system_usage_events. Stubbed to
    the zero shape for whatever ids were asked. tests/test_system_usage_readout.py
    binds the real function at import time, before this patch runs.
    """
    from scribe.services.system_usage import empty_system_usage

    async def _zeros(system_ids):
        return {int(s): empty_system_usage() for s in system_ids}

    with patch("scribe.services.system_usage.usage_for_systems",
               AsyncMock(side_effect=_zeros)):
        yield


@pytest.fixture(autouse=True)
def _no_task_log_arm():
    """Stub the task-log read arm that get_task / list_tasks / get_milestone
    carry.

    Autouse for the reason _no_rule_arm is: those three tools now read work
    logs, and the reads go through the access layer to Postgres. Every unit
    test that opens a task — and most of them do, because a task is what this
    codebase is mostly about — would otherwise try to reach the fake
    DATABASE_URL this file sets, to learn that a fake task has no logs.

    The arm's own behaviour is covered where it belongs: the payload shape and
    the tool wiring in tests/test_task_work_log_surface.py, which re-patches
    these explicitly, and the ACL scoping against real Postgres in
    tests/test_integration_task_work_log.py. A test that wants the arm live
    re-patches it, same as the rules arm.
    """
    with patch("scribe.services.task_logs.logs_for_task",
               AsyncMock(return_value=[])), \
         patch("scribe.services.task_logs.count_logs_for_task",
               AsyncMock(return_value=0)), \
         patch("scribe.services.task_logs.log_counts_for_tasks",
               AsyncMock(return_value={})):
        yield


@pytest.fixture(autouse=True)
def _no_lesson_rule_links(request):
    """Stub the lesson↔rule link decorations.

    Every door that returns a lesson or a rule now attaches its links, and the
    read is a real database call — so every unit test that opens either would
    otherwise reach for the fake DATABASE_URL to learn that nothing is linked.
    Both decorations are fail-open, so the cost would be a slow failed connect
    per test rather than a failure, which is worse: it would never be noticed.

    The candidate search a new lesson runs is stubbed to "nothing
    resembles it" for the same reason: it embeds, and the model load would
    otherwise ride along with every unit test that creates a lesson. The
    co-surfacing recorder is stubbed to "no proposal": every hook
    builder test that renders a lesson beside a rule would otherwise reach
    for the database to count the pair. And the via-lesson rule step
    is stubbed to "no lesson brought a rule": all three rule arms run it, and
    its first act is a database read. tests/test_rule_via_lesson.py binds the
    real function at import time, before this patch runs. The convergence
    check a "no rule fits" answer runs is stubbed to "no group" for
    the same reason; tests/test_lesson_convergence.py binds the real one.

    Skipped for integration tests, which exercise the real links against
    Postgres (tests/test_integration_lesson_rule_links.py).
    """
    if request.node.get_closest_marker("integration"):
        yield
        return
    with patch("scribe.services.lesson_rules.attach_lesson_rules", AsyncMock()), \
         patch("scribe.services.lesson_rules.attach_rule_lessons", AsyncMock()), \
         patch("scribe.services.lesson_rules.rule_candidates", AsyncMock(return_value=[])), \
         patch("scribe.services.lesson_rules.co_surfaced", AsyncMock(return_value="")), \
         patch("scribe.services.plugin_context._rules_via_lessons",
               AsyncMock(return_value=([], []))), \
         patch("scribe.services.lesson_rules.convergence_for", AsyncMock(return_value=None)):
        yield


@pytest.fixture(autouse=True)
def _no_family_triggers(request):
    """Stub family canon's write-time triggers.

    create_note, create_task, create_snippet, the updates that change a body,
    and update_milestone(status="done") now look for a pattern carried between
    projects, and each look is a database read (the repeat check also embeds).
    All are fail-open, so unstubbed they would cost a slow failed connect per
    unit test rather than a failure. Patched at the trigger functions, beneath
    `attach_family_hint`, so the attach itself — its order and its fail-open —
    still runs. tests/test_family.py binds the real pure parts; the triggers
    themselves are exercised against Postgres in
    tests/test_integration_family_promotion.py, which this skips.
    """
    if request.node.get_closest_marker("integration"):
        yield
        return
    with patch("scribe.services.family.citation_trigger", AsyncMock(return_value=None)), \
         patch("scribe.services.family.repeat_trigger", AsyncMock(return_value=None)), \
         patch("scribe.services.family.milestone_trigger", AsyncMock(return_value=None)), \
         patch("scribe.services.family.milestone_is_open", AsyncMock(return_value=False)):
        yield


@pytest.fixture(autouse=True)
def _no_moment_delivery(request):
    """Stub the moment lookup the mapped MCP tools attach.

    update_task, the create_* record tools and the milestone tools now hand
    back the rules mounted on the moment they mark, and finding those is two
    database reads. Patched BENEATH the attach, at `deliver_for_act`, so the
    attach itself — its fail-open and its non-dict pass-through — still runs.
    Skipped for integration tests; tests/test_moment_delivery.py binds the
    real one.
    """
    if request.node.get_closest_marker("integration"):
        yield
        return
    from scribe.services.retrieval_pipeline import RuleResult

    with patch("scribe.services.moment_delivery.deliver_for_act",
               AsyncMock(return_value=([], RuleResult()))):
        yield


@pytest.fixture(autouse=True)
def _no_rule_arm():
    """Stub the write-path hint's standing-RULES arm.

    Autouse, and deliberately so. The arm calls semantic_search_rules, which
    loads the embedding model — so every unrelated plugin-context test that
    already stubs the NOTES search would otherwise pull a real model into a
    unit test through the one arm it forgot to stub. The forty-odd existing
    call sites should not each have to learn about a new arm.

    The arm's own behaviour is covered where it belongs: the document shape in
    tests/test_services_rule_embeddings.py, the surfacing rules against real
    Postgres in tests/test_integration_rule_surfacing.py, and the hook's dedup
    channel in tests/test_write_path_trigger.py. A test that wants the arm
    live can re-patch it.
    """
    with patch("scribe.services.plugin_context.semantic_search_rules",
               AsyncMock(return_value=[])):
        yield


@pytest.fixture(autouse=True)
def _no_rule_overlap():
    """Stub the rule/preference create path's overlap check.

    The same reason as _no_rule_arm, one door over: every create_rule /
    create_project_rule / create_preference now asks semantic_search_rules
    whether an existing record answers the same moment, so each existing
    rule-tool unit test would load the embedding model through a call it never
    meant to make. The check's own behaviour is tested in
    tests/test_rule_overlap_gate.py, which binds the real function at import
    time — before this patch runs — and stubs the search beneath it instead.
    """
    with patch("scribe.services.dedup.find_overlapping_rules",
               AsyncMock(return_value=[])):
        yield


@pytest.fixture(autouse=True)
def _default_gate_bars():
    """Read the create gate's bars as their defaults, not from settings.

    Every create through the note gate now asks the user's settings for its
    similarity bars, which is a database read on a path the gate's unit tests
    run without one. Stubbed one level down — `_gate_setting`, not `gate_bars`
    — so the copy-band logic above it (which kinds read an overlap floor, the
    floor never sitting above the bar) still runs in every test. The setting
    read itself is tested in tests/test_gate_settings.py, which binds the real
    function at import time, before this patch runs.
    """
    from scribe.services.dedup import GATE_DEFAULT_THRESHOLDS

    async def _default(user_id, key, lo):
        return GATE_DEFAULT_THRESHOLDS[key]

    with patch("scribe.services.dedup._gate_setting", AsyncMock(side_effect=_default)):
        yield