FabledScribe/tests/helpers.py

"""Shared test helpers — the plain functions tests call, as opposed to the
fixtures in conftest.py.

Each of these was copied into several test modules before #2825 consolidated
them; a module imports what it needs with ``from tests.helpers import ...``.
"""
from __future__ import annotations

from contextlib import contextmanager
from datetime import datetime, timezone
from types import SimpleNamespace
from unittest.mock import AsyncMock, MagicMock, patch


async def drive_update_note(note, **kwargs):
    """Run `services/notes.update_note` against a stand-in row.

    The patch stack is the point: update_note reaches for a version snapshot,
    an embedding refresh and a project reactivation on its way out, none of
    which a unit test has. Written twice — once for the snippet mirror
    (#3128) and once for the verification fields (#3182/317) — before being
    consolidated here.

    Returns whatever update_note returned; assert on the `note` you passed in.
    """
    from unittest.mock import AsyncMock as _AsyncMock

    session = make_mock_session()
    result = MagicMock()
    result.scalars.return_value.first.return_value = note
    session.execute = _AsyncMock(return_value=result)
    with patch("scribe.services.notes.async_session") as cls, \
         patch("scribe.services.notes.embed_note", MagicMock()), \
         patch("scribe.services.notes._maybe_reactivate_project", _AsyncMock()), \
         patch("scribe.services.note_versions.create_version", _AsyncMock()):
        cls.return_value = session
        from scribe.services.notes import update_note
        return await update_note(user_id=7, note_id=note.id, **kwargs)


def tool_doc(module: str, name: str) -> str:
    """An MCP tool's docstring, whitespace-flattened.

    Flattened because these are hard-wrapped at ~76 characters, so any phrase
    worth asserting on is liable to straddle a line break — a property of the
    formatter, not of the guidance. The disambiguator guard (#3123) learned
    that on its own first run, matching raw text and reporting a phrase absent
    that was plainly there.

    Used by every test that pins the docstring CONTRACT rather than its
    wording. The tool docstring is the agent-facing contract (rule 119), so
    these guards exist to catch it being tidied down to a parameter list.
    """
    import importlib
    import re as _re

    fn = getattr(importlib.import_module(module), name)
    assert fn.__doc__, f"{name} has no docstring at all"
    return _re.sub(r"\s+", " ", fn.__doc__)


def compiled_sql(element, dialect=None) -> str:
    """A SQLAlchemy clause or statement rendered as literal SQL text.

    For asserting on the shape of a predicate without a database — which is how
    the visibility clauses and the knowledge facets are both tested. Was a
    private copy in each of those modules before #3128 needed a third.

    Pass `dialect` when the assertion is about something only one backend
    renders — a Postgres row-lock mode, say. The generic dialect is enough for
    a predicate's shape and would quietly drop the rest.
    """
    return str(element.compile(dialect=dialect, compile_kwargs={"literal_binds": True}))


def make_mock_session() -> AsyncMock:
    """A stand-in for ``async_session()`` — usable as ``async with``, with the
    commit/refresh/add surface a service touches.

    ``add`` is a MagicMock because the real ``Session.add`` is synchronous;
    an AsyncMock there would hand the service an un-awaited coroutine.
    """
    s = AsyncMock()
    s.__aenter__ = AsyncMock(return_value=s)
    s.__aexit__ = AsyncMock(return_value=False)
    s.add = MagicMock()
    s.commit = AsyncMock()
    s.refresh = AsyncMock()
    return s


async def ensure_user(session, username: str, role: str = "user"):
    """Get-or-create a User by username inside an open session (flushed, not
    committed).

    Integration tests share one database for the whole lane run, so a second
    test re-creating the same username dies on the unique constraint —
    every integration seed goes through this instead of ``User(...)`` + add.
    """
    from sqlalchemy import select

    from scribe.models.user import User

    existing = (
        await session.execute(select(User).where(User.username == username))
    ).scalar_one_or_none()
    if existing is not None:
        return existing
    user = User(username=username, role=role)
    session.add(user)
    await session.flush()
    return user


def fake_record(**attrs) -> MagicMock:
    """A MagicMock record with REAL values on the attributes named, and a
    ``to_dict()`` that mirrors them.

    The hazard this exists for (note 2109): an auto-created MagicMock attribute
    is truthy and has a repr — so a bare MagicMock handed to the product reads
    as trashed, shared, a task, and owned by a MagicMock. Name every attribute
    the code under test will read; the per-model ``fake_*`` builders below
    carry the ordinary defaults so a call site states only what the test is
    about. ``created_at`` / ``updated_at`` are set as attributes but kept out
    of ``to_dict()`` (no test serialises them, and the real models isoformat
    them).
    """
    n = MagicMock()
    for key, value in attrs.items():
        setattr(n, key, value)
    n.to_dict.return_value = {
        k: v for k, v in attrs.items() if k not in ("created_at", "updated_at")
    }
    return n


def _with_defaults(defaults: dict, attrs: dict) -> MagicMock:
    values = dict(defaults)
    values.update(attrs)
    return fake_record(**values)


def _now():
    return datetime.now(timezone.utc)


def fake_note(**attrs) -> MagicMock:
    """A stand-in Note: own (user_id=7, the caller `_bind_user` binds), live,
    not a task, no structured data. The injected menu reads is_task /
    task_kind / note_type / status for its kind marker, user_id for the
    "shared by …" attribution, data for a snippet's language, deleted_at for
    trash.

    `status` follows `is_task`, because on the real model it DEFINES it —
    `Note.is_task` is `status is not None`. A stand-in task with no status is
    a row the database cannot hold, and code that reads both would be tested
    against a shape it will never meet.
    """
    is_task = attrs.get("is_task", False)
    return _with_defaults({
        "id": 1, "title": "t", "body": "", "tags": [], "user_id": 7,
        "note_type": "note", "is_task": False, "task_kind": "work",
        "status": "todo" if is_task else None,
        "data": None, "deleted_at": None,
        # Milestone 317: a truthy mock here reads as "this note carries a
        # check", which trips the guard on records that may not have one.
        "verify_with": None, "expires_when": None, "verified_at": None,
    }, attrs)


def fake_task(**attrs) -> MagicMock:
    """A stand-in task note — get_task reads parent_id, deleted_at, user_id."""
    return _with_defaults({
        "id": 1, "title": "t", "body": "", "status": "todo", "priority": "none",
        "tags": [], "parent_id": None, "project_id": None, "is_task": True,
        "task_kind": "work", "user_id": 7, "deleted_at": None,
        "verify_with": None, "expires_when": None, "verified_at": None,
    }, attrs)


def fake_snippet(**attrs) -> MagicMock:
    """A stand-in snippet note. ``data`` is explicitly None: snippet_fields
    prefers `data` when truthy, and a MagicMock is truthy."""
    return _with_defaults({
        "id": 1, "title": "debounce — rate-limit a callback",
        "body": "```js\nreturn 1\n```\n", "tags": ["js", "snippet"],
        "note_type": "snippet", "is_task": False, "task_kind": "work",
        "user_id": 7, "data": None, "deleted_at": None,
        "status": None,
        "verify_with": None, "expires_when": None, "verified_at": None,
    }, attrs)


def fake_lesson(**attrs) -> MagicMock:
    """A stand-in lesson: a note whose `note_type` is what makes it one.

    The title carries the trigger because `compose_title` builds it that way —
    `{what} — {when it applies}` — so a menu line rendering only the title is
    already showing the reader when this lesson applies. Tests that used a bare
    title here would be testing a record the product cannot create.

    The check fields and `arose_from_id` are explicitly None for the reason
    `fake_snippet`'s `data` is: `update_note` reads `verify_with` and
    `expires_when` to decide whether to run the check-field guard, and an
    auto-created MagicMock attribute is truthy — so a default lesson driven
    through the update path would take a branch no real record takes.
    """
    attrs.setdefault(
        "title",
        "Give absolutely-positioned siblings an explicit stacking order — "
        "placing two absolutely-positioned elements in the same area",
    )
    attrs.setdefault("data", {"when_to_apply": "two absolute siblings overlap"})
    attrs.setdefault("status", None)
    attrs.setdefault("arose_from_id", None)
    attrs.setdefault("verify_with", None)
    attrs.setdefault("expires_when", None)
    attrs.setdefault("verified_at", None)
    return fake_note(note_type="lesson", **attrs)


def fake_project(**attrs) -> MagicMock:
    """design_system_id is explicit: a truthy auto-attribute would route every
    project through the design-system branch and out to a real database."""
    return _with_defaults({
        "id": 1, "title": "P", "description": "", "goal": "", "status": "active",
        "color": None, "design_system_id": None, "user_id": 7,
    }, attrs)


def fake_milestone(**attrs) -> MagicMock:
    return _with_defaults({
        "id": 1, "project_id": 1, "title": "MS", "description": None,
        "status": "active", "order_index": 0,
    }, attrs)


def fake_system(**attrs) -> MagicMock:
    return _with_defaults(
        {"id": 1, "name": "Reader", "project_id": 5, "canonical_id": None}, attrs,
    )


def fake_rulebook(**attrs) -> MagicMock:
    return _with_defaults({
        "id": 1, "owner_user_id": 7, "title": "FabledSword family",
        "description": "", "created_at": _now(), "updated_at": _now(),
    }, attrs)


def fake_topic(**attrs) -> MagicMock:
    return _with_defaults({
        "id": 10, "rulebook_id": 1, "title": "git-workflow", "description": "",
        "order_index": 0, "created_at": _now(), "updated_at": _now(),
    }, attrs)


def fake_rule(**attrs) -> MagicMock:
    return _with_defaults({
        "id": 1, "topic_id": 10, "project_id": None, "title": "dev is home",
        "statement": "Work directly on dev", "why": "", "how_to_apply": "",
        # Named for the note-2109 reason the whole helper exists: unnamed,
        # `when_to_apply` and `arose_from_id` would be truthy MagicMocks and
        # rule_brief would attach both keys on every stand-in.
        "when_to_apply": None, "arose_from_id": None,
        # Named for the same reason one line up, and it bites harder here.
        # `rule_brief` and `to_dict` both emit `kind or "rule"`, and a
        # MagicMock is truthy — so an unnamed `kind` would put a MagicMock
        # where every payload promises a force, and every stand-in rule would
        # read as neither a rule nor a preference.
        "kind": "rule",
        # Same reason, and the same trap one field further on: an unnamed
        # `verify_with` is a truthy MagicMock, so every stand-in rule would
        # claim to carry a check and rule_brief would stamp a MagicMock date
        # onto all of them. Most rules have none — that is the default here.
        "verify_with": None, "expires_when": None, "verified_at": None,
        "order_index": 0, "created_at": _now(), "updated_at": _now(),
    }, attrs)


def plain_rule_detail():
    """Stub `rulebooks_svc.rule_detail` down to the record's own dict.

    Every rule-tool unit test needs it and none of them wants it: the real
    `rule_detail` reads the rule's Systems and its typed edges from the
    database, which a unit test has none of. What these tests assert is that
    the TOOL forwarded the right arguments, so the seam is stubbed the same
    way the create/update calls themselves already are.

    Consolidated here on its second copy, per this module's own reason for
    existing (#2825) — two stubs for one seam drift apart quietly, and a test
    stubbing the seam slightly differently is a test asserting something
    slightly different than it appears to.
    """
    async def _detail(_uid, rule, _system_ids=None):
        return rule.to_dict()
    return patch("scribe.mcp.tools.rulebooks.rulebooks_svc.rule_detail", _detail)


class FakeMCP:
    """Stand-in for the FastMCP server a tool module's ``register(mcp)`` is
    handed: records the ``name=`` of every ``@mcp.tool(...)`` registration in
    ``names`` and leaves the function untouched, so a test can assert which
    tools a module exposes."""

    def __init__(self) -> None:
        self.names: list[str] = []

    def tool(self, name=None):
        self.names.append(name)
        return lambda fn: fn


def loc(path: str = "", repo: str = "", symbol: str = "") -> dict:
    """One snippet location, in the shape the record stores."""
    return {"repo": repo, "path": path, "symbol": symbol}


def design_token_stub(name, value_by_mode, group_name=None, purpose=None,
                      order_index=0, supersedes=None) -> SimpleNamespace:
    """A design-token row as the cascade / stylesheet code reads it."""
    return SimpleNamespace(
        name=name, value_by_mode=value_by_mode, group_name=group_name,
        purpose=purpose, order_index=order_index, supersedes=supersedes or [],
    )


@contextmanager
def http_sink(reply: bytes = b'{"context":"","note_ids":[]}'):
    """A throwaway local HTTP listener for hook end-to-end tests: yields
    ``(port, seen)`` where ``seen`` collects every GET's parsed query string
    (one dict per request, in order). Lets the shell be tested end to end —
    the extraction, the encoding, the URL — without a Scribe instance.

    Three test modules each carried their own ``_Sink`` handler before #2904
    consolidated them here; pass ``reply`` for the body the hook should see.
    """
    import http.server
    import threading
    import urllib.parse

    seen: list[dict] = []

    class _Sink(http.server.BaseHTTPRequestHandler):
        def do_GET(self):
            seen.append(urllib.parse.parse_qs(urllib.parse.urlparse(self.path).query))
            self.send_response(200)
            self.send_header("Content-Type", "application/json")
            self.end_headers()
            self.wfile.write(reply)

        def log_message(self, *a):
            pass

    server = http.server.HTTPServer(("127.0.0.1", 0), _Sink)
    threading.Thread(target=server.serve_forever, daemon=True).start()
    try:
        yield server.server_port, seen
    finally:
        server.shutdown()
        server.server_close()


def writepath_cfg(**over):
    """A complete `get_writepath_config` stand-in, built from the registry (#4102).

    DERIVED, NOT LITERAL, and the reason is a failure mode this file already
    warned about in prose without being able to prevent: the write-path hint
    drives three arms, each of which reads its numbers out of the config dict
    inside a fail-open `except`. A dict missing one key does not raise where a
    reader would see it — the arm silently becomes a no-op, which is
    indistinguishable from the arm working and finding nothing.

    So the keys come from `retrieval_surfaces.SURFACES`. A seventh surface, or a
    rename, changes this helper for free and cannot quietly disable an arm in
    ten hand-written dicts that each looked complete on the day they were typed.

    NOT EVERY KEY IS A SURFACE, and that gap already bit once (#4214). The
    checkpoint bar has no entry in SURFACES on purpose — everything in that
    table is a floor/budget pair belonging to one QUERY, and the checkpoint
    runs none, it re-reads hits the rule arms already produced. Adding it there
    would give it a phantom budget. So it is taken from the module constant
    instead, which keeps the "one literal, in the product" property even though
    the derivation differs. `test_the_config_stand_in_carries_every_key_the_
    real_one_does` is what makes the next addition fail loudly here rather than
    silently no-op an arm, which is the whole claim this docstring makes.
    """
    from scribe.services.plugin_context import _CHECKPOINT_DEFAULT
    from scribe.services.retrieval_surfaces import SURFACES

    cfg = {
        "enabled": True,
        "threshold": SURFACES["write_path"].floor_default,
        "top_k": SURFACES["write_path"].budget_default,
        "rule_threshold": SURFACES["write_path_rule"].floor_default,
        "rule_top_k": SURFACES["write_path_rule"].budget_default,
        "tool_rule_threshold": SURFACES["pre_tool_rule"].floor_default,
        "tool_rule_top_k": SURFACES["pre_tool_rule"].budget_default,
        "checkpoint_threshold": _CHECKPOINT_DEFAULT,
    }
    cfg.update(over)
    return cfg