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


def compiled_sql(element) -> 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.
    """
    return str(element.compile(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 for its kind marker, user_id for the "shared by …"
    attribution, data for a snippet's language, deleted_at for trash."""
    return _with_defaults({
        "id": 1, "title": "t", "body": "", "tags": [], "user_id": 7,
        "note_type": "note", "is_task": False, "task_kind": "work",
        "data": None, "deleted_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,
    }, 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,
    }, 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, "tier": "always_on", "arose_from_id": None,
        # 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)


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()