"""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 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. Two files had spelled this out identically before a third was about to (#3207 — derive before the third copy), 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 (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 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 (#2932), so anything short of the real tool proves nothing. Consolidated 2026-09-21 from three byte-identical copies in test_contract_around_the_change, test_hook_json_reader and test_session_slippage_readout. """ 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 (milestone 312). 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. Consolidated 2026-09-21 from two byte-identical copies in test_integration_rule_move and test_integration_rule_verification. """ 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)