FabledGrimoire/tests/helpers.py

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

Each of these serves several test modules; 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.

    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. Matching raw text reports a phrase absent
    that is plainly there.

    Used by every test that pins the docstring CONTRACT rather than its
    wording. The tool docstring is the agent-facing contract, 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.

    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


def session_returning(note):
    """``make_mock_session`` whose every ``execute()`` yields `note` (or None).

    The commonest shape above the bare session: a service that loads one
    record by id and acts on it, and it belongs beside `make_mock_session`
    for the same reason that one does.

    For a service making SEVERAL different reads, set
    ``session.execute.side_effect`` to a list of results instead.
    """
    session = make_mock_session()
    result = MagicMock()
    result.scalars.return_value.first.return_value = note
    session.execute = AsyncMock(return_value=result)
    return session


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: 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,
        # 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,
        # A truthy mock would read as a live claim.
        "claimed_by": None, "claimed_at": None, "claim_touched_at": None,
        "claim_session": 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,
        # A truthy mock would read as a live claim.
        "claimed_by": None, "claimed_at": None, "claim_touched_at": None,
        "claim_session": 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 is the subject alone and the trigger lives in `data`, because
    that is the record the product creates — the trigger joins the title only
    in the embedded document.

    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")
    attrs.setdefault("data", {
        "what": "Give absolutely-positioned siblings an explicit stacking order",
        "when_to_apply": "placing two absolutely-positioned elements in the same area",
    })
    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.

    Shared here because 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, _moments=None):
        return rule.to_dict()
    return patch("scribe.mcp.tools.rulebooks.rulebooks_svc.rule_detail", _detail)


class FakeMCP:
    """Stand-in for the MCPServer 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":[]}',
              by_path: dict[str, bytes] | None = None):
    """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 Grimoire instance.

    Pass ``reply`` for the body the hook should see.

    ``by_path`` is for a hook that calls more than one route: each request is
    answered with the reply for its path (``reply`` when none is listed), and
    every entry in ``seen`` then also carries ``_path`` and, for a POST,
    ``_body`` — the raw request body, decoded.
    """
    import http.server
    import threading
    import urllib.parse

    seen: list[dict] = []

    class _Sink(http.server.BaseHTTPRequestHandler):
        def _answer(self, body: str | None):
            parsed = urllib.parse.urlparse(self.path)
            entry = urllib.parse.parse_qs(parsed.query)
            out = reply
            if by_path is not None:
                entry["_path"] = parsed.path
                out = by_path.get(parsed.path, reply)
                if body is not None:
                    entry["_body"] = body
            seen.append(entry)
            self.send_response(200)
            self.send_header("Content-Type", "application/json")
            self.end_headers()
            self.wfile.write(out)

        def do_GET(self):
            self._answer(None)

        def do_POST(self):
            size = int(self.headers.get("Content-Length") or 0)
            self._answer(self.rfile.read(size).decode("utf-8", "replace"))

        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.

    DERIVED, NOT LITERAL, because of a failure mode prose cannot 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
    hand-written dicts that each looked complete on the day they were typed.

    NOT EVERY KEY IS A SURFACE. 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


def need_tools(*tools):
    """Skip unless every named executable is on PATH.

    For the hook tests, which drive `plugin/hooks/*.sh` through a real shell
    and therefore depend on whatever that shell reaches for — `jq`, `awk`,
    `git`, `curl`. Those are present on the CI image and routinely absent from
    a developer's box, and the honest answer there is "not exercised", not a
    failure: a red test would say the hook is broken when nothing about the
    hook was ever run.

    A skip rather than a stub on purpose. Stubbing `jq` would test the stub —
    these tests exist precisely because the shell's behaviour is the thing in
    question, so anything short of the real tool proves nothing.
    """
    import shutil

    import pytest

    for t in tools:
        if shutil.which(t) is None:
            pytest.skip(f"hook runtime tool {t!r} not installed")


async def rule_row(rule_id: int):
    """Read a Rule straight from Postgres, outside whatever session the code
    under test used.

    THE POINT IS THE SEPARATE SESSION. An integration test that asserts on the
    object the service just returned is asserting on that session's identity
    map, which can hold a value the database never accepted — a column the
    write never reached, a default the ORM supplied rather than the schema.
    Opening a new session forces a real read and is the only way `verify_with`
    being NULL rather than "" is distinguishable at all.

    Import is lazy because this module is imported by unit tests that have no
    database and must not pay for one — the same reason `plugin_config` defers
    its service imports.
    """
    from scribe.models import async_session
    from scribe.models.rulebook import Rule

    async with async_session() as s:
        return await s.get(Rule, rule_id)


# Software-only vocabulary. Grimoire's product text speaks for any work a person
# drives through an agent — configuration, infrastructure, documents — so a
# catalog or a shipped shape that reads as software-only is wrong on most of
# the installs it ships to. The actions particular to one kind of work belong
# in data (mappings, rulebooks, preferences), not in the product's words.
DEV_ONLY = (r"\bCI\b", r"\bcommit", r"\bpull request", r"\bcode\b", r"\btest",
            r"\bgit\b", r"\brepo(s|sitor\w*)?\b", r"\bbranch", r"\bcompil", r"\bbuild\b")


def dev_only_hits(text: str) -> list[str]:
    """The DEV_ONLY patterns this text contains, case-insensitively."""
    import re

    return [w for w in DEV_ONLY if re.search(w, text, re.IGNORECASE)]


def skill_text(name: str) -> str:
    """Everything a bundled skill states: its SKILL.md, then each reference file.

    A skill keeps what matters on most turns in SKILL.md and moves what matters
    at one moment into sibling files it links (Agent Skills progressive
    disclosure). Both are the skill's own statement, so a test asking
    "does this skill still say X" reads them together rather than pinning X to
    whichever file it happens to sit in today.
    """
    import pathlib

    folder = pathlib.Path(__file__).resolve().parents[1] / "plugin" / "skills" / name
    refs = sorted(p for p in folder.glob("*.md") if p.name != "SKILL.md")
    return "\n\n".join(p.read_text() for p in [folder / "SKILL.md", *refs])


def forge_tarball(files: dict[str, bytes], top: str = "widget") -> bytes:
    """A gzipped tar shaped like a forge archive: every file under one
    top-level directory (repo-ref/), which the scan strips."""
    import io
    import tarfile

    buf = io.BytesIO()
    with tarfile.open(fileobj=buf, mode="w:gz") as tar:
        for path, data in files.items():
            info = tarfile.TarInfo(f"{top}/{path}")
            info.size = len(data)
            tar.addfile(info, io.BytesIO(data))
    return buf.getvalue()