tests/helpers.py

@@ -418,3 +418,55 @@ def writepath_cfg(**over):
     }
     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)

tests/test_contract_around_the_change.py

@@ -34,24 +34,19 @@ A FIXTURE REPO, NEVER THIS ONE. Asserting against Scribe's own files would
 make the test a description of today's tree, failing the next time someone
 renames something (rule 115's reasoning, one floor down).
 """
-import shutil
 import subprocess
 from pathlib import Path
 
 import pytest
 
+from tests.helpers import need_tools
+
 DEFS = Path(__file__).resolve().parents[1] / "plugin" / "hooks" / "scribe_defs.sh"
 HOOK = Path(__file__).resolve().parents[1] / "plugin" / "hooks" / "scribe_prior_art.sh"
 
 
-def _need(*tools):
-    for t in tools:
-        if shutil.which(t) is None:
-            pytest.skip(f"hook runtime tool {t!r} not installed")
-
-
 def run(script: str) -> str:
-    _need("bash", "awk", "git", "grep", "sed")
+    need_tools("bash", "awk", "git", "grep", "sed")
     r = subprocess.run(
         ["bash", "-c", f'set -uo pipefail\n. "{DEFS}"\n{script}'],
         capture_output=True, text=True,
@@ -63,7 +58,7 @@ def run(script: str) -> str:
 @pytest.fixture()
 def repo(tmp_path):
     """A small git repo: a definition with a reader, and one without."""
-    _need("git")
+    need_tools("git")
     # `other` lives in lib.py BESIDE widget, and nothing references it. That
     # placement is the point of the no-readers case: putting it in its own
     # file would leave that file as its reader, since only the file being
@@ -248,6 +243,6 @@ def test_the_hook_asks_the_contract_question_first():
 
 
 def test_the_hook_is_still_shell_valid():
-    _need("bash")
+    need_tools("bash")
     subprocess.run(["bash", "-n", str(HOOK)], check=True)
     subprocess.run(["bash", "-n", str(DEFS)], check=True)

tests/test_hook_json_reader.py

@@ -22,25 +22,20 @@ from __future__ import annotations
 
 import json
 import re
-import shutil
 import subprocess
 from pathlib import Path
 from urllib.parse import quote
 
 import pytest
 
+from tests.helpers import need_tools
+
 HOOKS = Path(__file__).resolve().parents[1] / "plugin" / "hooks"
 DEFS = HOOKS / "scribe_defs.sh"
 PARSER = HOOKS / "scribe_json.awk"
 TURN = HOOKS / "scribe_turn.awk"
 
 
-def _need(*tools):
-    for t in tools:
-        if shutil.which(t) is None:
-            pytest.skip(f"hook runtime tool {t!r} not installed")
-
-
 def sh(script: str, stdin: str = "") -> str:
     """Run a snippet with scribe_defs.sh sourced, under the hooks' own flags.
 
@@ -50,7 +45,7 @@ def sh(script: str, stdin: str = "") -> str:
     quietly normalise the characters it exists to check: the first version of
     this file did, and reported a round-trip failure that was entirely its own.
     """
-    _need("bash", "awk")
+    need_tools("bash", "awk")
     r = subprocess.run(
         ["bash", "-c", f'set -uo pipefail\n. "{DEFS}"\n{script}'],
         input=stdin.encode(), capture_output=True,
@@ -60,7 +55,7 @@ def sh(script: str, stdin: str = "") -> str:
 
 
 def flat(doc: str, mode: str = "whole") -> list[tuple[str, str, str]]:
-    _need("awk")
+    need_tools("awk")
     r = subprocess.run(["awk", "-v", f"mode={mode}", "-f", str(PARSER)],
                        input=doc.encode(), capture_output=True)
     assert r.returncode == 0, r.stderr.decode()
@@ -260,7 +255,7 @@ def test_urlenc_and_the_envelope_can_fail():
 # The transcript turn, which is the largest thing jq was doing here.
 
 def _turn(records: list[dict]) -> dict:
-    _need("awk")
+    need_tools("awk")
     doc = "\n".join(json.dumps(r) for r in records) + "\n"
     p1 = subprocess.run(["awk", "-v", "mode=lines", "-f", str(PARSER)],
                         input=doc, capture_output=True, text=True)

tests/test_integration_rule_move.py

@@ -20,7 +20,7 @@ from scribe.models.rulebook import Rule
 from scribe.services import canonical_systems as canonical_svc
 from scribe.services import rule_versions as rv_svc
 from scribe.services import rulebooks as rulebooks_svc
-from tests.helpers import ensure_user
+from tests.helpers import ensure_user, rule_row
 
 pytestmark = [pytest.mark.integration, pytest.mark.usefixtures("_dispose_engine")]
 
@@ -69,17 +69,12 @@ async def homes():
     return ids
 
 
-async def _row(rule_id: int) -> Rule:
-    async with async_session() as s:
-        return await s.get(Rule, rule_id)
-
-
 async def test_a_project_rule_becomes_global_and_keeps_everything(homes):
     owner, rule_id = homes["owner"], homes["rule"]
     moved = await rulebooks_svc.move_rule(rule_id, owner, topic_id=homes["topic"])
 
     assert moved.id == rule_id
-    row = await _row(rule_id)
+    row = await rule_row(rule_id)
     assert (row.topic_id, row.project_id) == (homes["topic"], None)
     assert len(await rv_svc.list_versions(rule_id)) == 1, "the move must not drop history"
     areas = await rulebooks_svc.list_rule_systems([rule_id])
@@ -100,7 +95,7 @@ async def test_a_global_rule_can_move_onto_a_project(homes):
     owner, rule_id = homes["owner"], homes["rule"]
     await rulebooks_svc.move_rule(rule_id, owner, topic_id=homes["topic"])
     await rulebooks_svc.move_rule(rule_id, owner, project_id=homes["other"])
-    row = await _row(rule_id)
+    row = await rule_row(rule_id)
     assert (row.topic_id, row.project_id) == (None, homes["other"])
 
 
@@ -126,7 +121,7 @@ async def test_refusals_happen_before_anything_is_written(homes):
     with pytest.raises(ValueError, match=f"rule {clash.id}"):
         await rulebooks_svc.move_rule(rule_id, owner, topic_id=homes["topic"])
 
-    row = await _row(rule_id)
+    row = await rule_row(rule_id)
     assert (row.topic_id, row.project_id) == (None, homes["home"])
 
 

tests/test_integration_rule_verification.py

@@ -23,7 +23,7 @@ import pytest_asyncio
 from scribe.models import async_session
 from scribe.models.rulebook import Rule
 from scribe.services import rulebooks as rulebooks_svc
-from tests.helpers import ensure_user
+from tests.helpers import ensure_user, rule_row
 
 pytestmark = [pytest.mark.integration, pytest.mark.usefixtures("_dispose_engine")]
 
@@ -57,13 +57,8 @@ async def constraint():
     return {"uid": uid, "rule": rule.id}
 
 
-async def _row(rule_id: int) -> Rule:
-    async with async_session() as s:
-        return await s.get(Rule, rule_id)
-
-
 async def test_the_check_and_its_expiry_persist(constraint):
-    row = await _row(constraint["rule"])
+    row = await rule_row(constraint["rule"])
     assert row.verify_with == "read the workflow's shell setting"
     assert row.expires_when == "the runner can be given a bash shell"
     assert row.verified_at is not None
@@ -79,7 +74,7 @@ async def test_an_empty_string_becomes_null_not_an_empty_column(constraint):
     await rulebooks_svc.update_rule(
         constraint["rule"], constraint["uid"], verify_with="", expires_when="",
     )
-    row = await _row(constraint["rule"])
+    row = await rule_row(constraint["rule"])
     assert row.verify_with is None
     assert row.expires_when is None
 
@@ -89,7 +84,7 @@ async def test_naming_a_field_in_clear_empties_it(constraint):
     await rulebooks_svc.update_rule(
         constraint["rule"], constraint["uid"], clear=["verify_with"],
     )
-    row = await _row(constraint["rule"])
+    row = await rule_row(constraint["rule"])
     assert row.verify_with is None
     # expires_when was NOT named, so it survives — clearing is per-field, and
     # a caller retiring one field must not lose the others.
@@ -107,7 +102,7 @@ async def test_rewording_the_check_drops_the_stamp(constraint):
         constraint["rule"], constraint["uid"],
         verify_with="read the runner's container shell, not the image's",
     )
-    row = await _row(constraint["rule"])
+    row = await rule_row(constraint["rule"])
     assert row.verified_at is None
 
 
@@ -115,7 +110,7 @@ async def test_clearing_the_check_drops_the_stamp(constraint):
     await rulebooks_svc.update_rule(
         constraint["rule"], constraint["uid"], clear=["verify_with"],
     )
-    row = await _row(constraint["rule"])
+    row = await rule_row(constraint["rule"])
     assert row.verified_at is None
 
 
@@ -132,7 +127,7 @@ async def test_editing_anything_else_leaves_the_stamp_alone(constraint):
             "applies to the build, not to `run:`.",
         expires_when="the runner grows a shell setting",
     )
-    row = await _row(constraint["rule"])
+    row = await rule_row(constraint["rule"])
     assert row.verified_at is not None
     assert row.why.startswith("act_runner picks the shell")
 
@@ -221,11 +216,11 @@ async def test_a_failed_check_writes_nothing(rulebook_of_three):
     not in a special condition — it is WRONG. Recording the failure would let
     it sit there being false with the sweep satisfied that someone looked.
     """
-    before = await _row(rulebook_of_three["stale"])
+    before = await rule_row(rulebook_of_three["stale"])
     await rulebooks_svc.mark_rule_verified(
         rulebook_of_three["stale"], rulebook_of_three["uid"], still_true=False,
     )
-    after = await _row(rulebook_of_three["stale"])
+    after = await rule_row(rulebook_of_three["stale"])
     assert after.verified_at == before.verified_at
 
 

tests/test_session_slippage_readout.py

@@ -32,13 +32,14 @@ from __future__ import annotations
 
 import json
 import os
-import shutil
 import subprocess
 import time
 from pathlib import Path
 
 import pytest
 
+from tests.helpers import need_tools
+
 ROOT = Path(__file__).resolve().parents[1]
 HOOKS = ROOT / "plugin" / "hooks"
 DEFS = HOOKS / "scribe_defs.sh"
@@ -47,14 +48,8 @@ RECORDER = HOOKS / "scribe_record_outcome.sh"
 HOOKS_JSON = HOOKS / "hooks.json"
 
 
-def _need(*tools):
-    for t in tools:
-        if shutil.which(t) is None:
-            pytest.skip(f"hook runtime tool {t!r} not installed")
-
-
 def sh(script: str) -> str:
-    _need("bash", "awk")
+    need_tools("bash", "awk")
     r = subprocess.run(
         ["bash", "-c", f'set -uo pipefail\n. "{DEFS}"\n{script}'],
         capture_output=True, text=True, timeout=30,
@@ -173,7 +168,7 @@ def test_a_session_no_rule_touched_says_nothing_at_all(tmp_path):
 # ── The hook that carries it ──────────────────────────────────────────────
 
 def run_precompact(event: dict, tmpdir: Path) -> subprocess.CompletedProcess:
-    _need("bash")
+    need_tools("bash")
     env = dict(os.environ)
     env["TMPDIR"] = str(tmpdir)
     return subprocess.run(["bash", str(PRECOMPACT)], input=json.dumps(event),
@@ -210,7 +205,7 @@ def test_the_compaction_hook_never_emits_a_json_envelope(tmp_path):
 # ── The recorder that makes `acted` mean anything ─────────────────────────
 
 def run_recorder(event: dict, tmpdir: Path) -> subprocess.CompletedProcess:
-    _need("bash")
+    need_tools("bash")
     env = {"PATH": os.environ["PATH"], "HOME": str(tmpdir), "TMPDIR": str(tmpdir)}
     return subprocess.run(["bash", str(RECORDER)], input=json.dumps(event),
                           capture_output=True, text=True, timeout=30, env=env)
@@ -257,7 +252,7 @@ def test_the_recorder_is_registered_on_the_rule_outcome_tool():
 
 
 def test_the_recorder_is_shell_valid():
-    _need("bash")
+    need_tools("bash")
     subprocess.run(["bash", "-n", str(RECORDER)], check=True)