From d49e4ad106b04df48d1858ac97ef880be06fa91f Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Mon, 21 Sep 2026 18:27:05 -0400 Subject: [PATCH] refactor(tests): derive the two duplicated test helpers into tests/helpers.py (#4276) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Derive-first, ahead of the tests shape pay-down. An AST pass over all 2628 definitions under `tests/` found exactly three helper bodies duplicated across files. Two are real copies and are consolidated here; the third is not, and is left alone. `need_tools(*tools)` — byte-identical in three hook-test modules, each skipping when `jq`/`awk`/`git` is absent from PATH. Now snippet #4277. The `import shutil` each file carried existed only to serve it and goes with it. `rule_row(rule_id)` — byte-identical in two integration modules, reading a Rule back through a SEPARATE session so the assertion is about what Postgres holds rather than what the writing session's identity map remembers. Now snippet #4278. Its imports are lazy, because `tests/helpers.py` is imported by unit tests that have no database, which is the same reason `plugin_config` defers its service imports. NOT consolidated: `_side(uid, k, d="")` in test_services_plugin_context and test_write_path_trigger. The body is identical but it closes over a module-local `stored` dict, so it is not self-contained and "moving" it would mean inventing a parameter neither call site wants. That is convention plumbing — two tests independently writing the same one-line side_effect — and it is dismissed in the ledger rather than lifted. Worth recording for the next pass: a repeated NAME is not a family. `_row` is defined in five modules and only two of those share a body; the other three (test_list_rows_brief, test_calibration_stamp, test_shape_ledger) build entirely different objects. Grouping by name would have consolidated three things that have nothing in common. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01821k5B3Ysecp9fNYs92Kuy --- tests/helpers.py | 52 +++++++++++++++++++++ tests/test_contract_around_the_change.py | 15 ++---- tests/test_hook_json_reader.py | 15 ++---- tests/test_integration_rule_move.py | 13 ++---- tests/test_integration_rule_verification.py | 23 ++++----- tests/test_session_slippage_readout.py | 17 +++---- 6 files changed, 81 insertions(+), 54 deletions(-) diff --git a/tests/helpers.py b/tests/helpers.py index 8bc80f1..f3b6090 100644 --- a/tests/helpers.py +++ b/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) diff --git a/tests/test_contract_around_the_change.py b/tests/test_contract_around_the_change.py index 839ae69..919b3b1 100644 --- a/tests/test_contract_around_the_change.py +++ b/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) diff --git a/tests/test_hook_json_reader.py b/tests/test_hook_json_reader.py index fcdc371..c94b16c 100644 --- a/tests/test_hook_json_reader.py +++ b/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) diff --git a/tests/test_integration_rule_move.py b/tests/test_integration_rule_move.py index d37c4f7..93ef55b 100644 --- a/tests/test_integration_rule_move.py +++ b/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"]) diff --git a/tests/test_integration_rule_verification.py b/tests/test_integration_rule_verification.py index 9e94f52..2c3e01e 100644 --- a/tests/test_integration_rule_verification.py +++ b/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 diff --git a/tests/test_session_slippage_readout.py b/tests/test_session_slippage_readout.py index 73b656d..29f0a15 100644 --- a/tests/test_session_slippage_readout.py +++ b/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)