FabledScribe/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 three things that used to be
copied into a dozen test modules each (#2825). 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 since milestone 415; 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 (#278).

    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_system_labels():
    """Stub the menu's "which System is each line about?" lookup (#4364).

    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_task_log_arm():
    """Stub the task-log read arm that get_task / list_tasks / get_milestone
    grew in #4241.

    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 (milestone 440).

    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 (#4631) 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 (#4637) 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 (#4633)
    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 (#4634) 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_rule_arm():
    """Stub the write-path hint's standing-RULES arm (milestone 307).

    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 (#4134).

    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 (#4385).

    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