Files
FabledScribe/tests/test_pattern_coverage.py
T
bvandeusenandClaude Fable 5 bbee0d0db1
CI & Build / Python lint (push) Successful in 3s
CI & Build / Plugin hooks (push) Successful in 8s
CI & Build / integration (push) Successful in 24s
CI & Build / TypeScript typecheck (push) Successful in 32s
CI & Build / Python tests (push) Successful in 54s
CI & Build / Build & push image (push) Successful in 18s
refactor(tests): one definition each for the copied fixtures and fakes (#2825, milestone 296 area 1)
The shape ledger showed the same test scaffolding defined over and over:
_bind_user x12 (byte-identical), _dispose_engine x10 in three wordings,
_no_supersession x3, _make_mock_session x7 in three subsets, a get-or-create
User helper x2 (+3 inlined), and fifteen hand-rolled MagicMock note factories
each re-explaining the same "an auto-MagicMock attribute is truthy" hazard
(note 2109).

Now: conftest.py carries _bind_user / _dispose_engine / _no_supersession as
opt-in fixtures (pytestmark = usefixtures(...) per module, so unit tests pay
nothing), and tests/helpers.py carries make_mock_session(), ensure_user() and
fake_note(**attrs) — the hazard documented once, real values on every
attribute the product reads. Call sites were rewritten by AST so titles with
dashes and commas survived; the three SimpleNamespace _note stand-ins that
only feed a single function stay local.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-08-21 11:03:48 -04:00

503 lines
20 KiB
Python

"""Pattern-library coverage (#2692) — the extractor that mirrors the hook,
the shape/record matcher, and the end-to-end measurement on real Postgres.
The extractor here and the hook's awk program (scribe_prior_art.sh ARM 1)
must agree on what counts as "a definition" — the metric and the write-path
backstop are two views of the same doctrine. The EXTRACTION_VECTORS below
deliberately reuse the definitions test_write_path_trigger stages for the
hook; extending one detector means extending both, and this comment is the
tripwire.
"""
import io
import json
import tarfile
import pytest
import pytest_asyncio
from scribe.services.coverage import (
coverage_line,
extract_shapes,
largest_gaps,
scannable,
shapes_from_archive,
)
from scribe.services.shape_ledger import location_covers
from tests.helpers import ensure_user
# --- unit: the definition extractor (shared vectors with the hook) -----------
EXTRACTION_VECTORS = [
# (id, source text, expected (kind, name) list)
("python", "def make_app():\n pass\nclass Config:\n pass\n",
[("sym", "make_app"), ("sym", "Config")]),
("python-dunder-skip", "class C:\n def __init__(self):\n pass\n",
[("sym", "C")]),
("go-func", "func Resolve(x int) error {\n\treturn nil\n}\n",
[("sym", "Resolve")]),
("go-method", "func (s *Scanner) Resolve(x int) error {\n\treturn nil\n}\n",
[("sym", "Resolve")]),
("kotlin-fun", "suspend fun refreshQueue(id: Long) {\n}\n",
[("sym", "refreshQueue")]),
("rust-fn", "pub async fn fetch_all() -> u32 {\n 0\n}\n",
[("sym", "fetch_all")]),
("go-type", "type ForgeAdapter struct {\n\tname string\n}\n",
[("sym", "ForgeAdapter")]),
("rust-pub-crate", "pub(crate) struct Widget {}\n",
[("sym", "Widget")]),
("js-export-default", "export default function App() {}\n",
[("sym", "App")]),
("js-arrow", "const useThing = (id) => id;\nlet fetcher = async () => 0;\n",
[("sym", "useThing"), ("sym", "fetcher")]),
# Every selector line in a group counts — .btn-ghost, and .btn-text { }
# both announce a class, exactly as the hook's awk sees them.
("css", ".btn-primary {\n color: red;\n}\n.btn-ghost,\n.btn-text { }\n",
[("css", "btn-primary"), ("css", "btn-ghost"), ("css", "btn-text")]),
# Call sites, imports, and impl blocks are NOT definitions — matching
# them would drown the metric exactly as it would drown the hook.
("non-definitions",
"make_app()\nimpl Widget {\nreturn fetch_all\nimport os\nx = 1\n",
[]),
("dedup-within-file", "def f():\n pass\ndef f():\n pass\n",
[("sym", "f")]),
]
@pytest.mark.parametrize(
("text", "expected"),
[(t, e) for _i, t, e in EXTRACTION_VECTORS],
ids=[i for i, _t, _e in EXTRACTION_VECTORS],
)
def test_extractor_agrees_with_the_hook_on_what_defines(text, expected):
assert extract_shapes(text) == expected
def test_scannable_gates_prose_vendored_and_sourcemaps():
assert scannable("src/app.py")
assert scannable("web/button.css")
assert scannable(".gitea/workflows/ci.yml") # config IS worth recording
assert not scannable("README.md")
assert not scannable("dist/bundle.js.map")
assert not scannable("node_modules/x/index.js")
assert not scannable("web/node_modules/y/util.ts")
# A FILE named like a skip-dir is not a directory hit.
assert scannable("src/vendor.py")
# --- unit: reading shapes out of a forge tarball -----------------------------
def _tarball(files: dict[str, bytes], top: str = "widget") -> bytes:
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()
TREE = {
"src/app.py": b"def make_app():\n pass\nclass Config:\n def __init__(self):\n pass\n",
"src/util.py": b"def helper():\n pass\n",
"web/button.css": b".btn {\n color: red;\n}\n",
"README.md": b"def not_code(): pass\n",
"node_modules/x/index.js": b"function vendored() {}\n",
"data.bin": b"\xff\xfe\x00\x01",
}
# What TREE holds once the gates run: 4 shapes, none from the skipped files.
TREE_SHAPES = [
("src/app.py", "sym", "make_app"),
("src/app.py", "sym", "Config"),
("src/util.py", "sym", "helper"),
("web/button.css", "css", "btn"),
]
def test_shapes_from_archive_strips_the_wrapper_and_gates_files():
assert shapes_from_archive(_tarball(TREE)) == TREE_SHAPES
# --- unit: the covering predicate (lives with the ledger since #2788) --------
def test_location_covers_by_exact_path_dir_prefix_and_css_dot():
assert location_covers("src/app.py", "make_app", "src/app.py", "make_app")
# dir prefix + css dot normalization
assert location_covers("web", ".btn", "web/button.css", "btn")
assert not location_covers("src/app.py", "make_app", "src/util.py", "make_app")
def test_a_symbol_less_record_covers_nothing():
"""A whole-file snippet makes no claim about any particular definition
inside it — crediting all of them would inflate the number for free."""
assert not location_covers("src/app.py", "", "src/app.py", "make_app")
def test_no_prefix_bleed_between_sibling_directories():
assert not location_covers("src/lib", "helper", "src/library/x.py", "helper")
def test_largest_gaps_ranks_by_unclassified_and_drops_clean_dirs():
accounted = [
("src/app.py", "sym", "make_app", True),
("src/app.py", "sym", "Config", False),
("src/util.py", "sym", "helper", False),
("web/button.css", "css", "btn", True),
]
gaps = largest_gaps(accounted)
assert gaps == [{"dir": "src", "unclassified": 2, "total": 3}]
def test_coverage_line_is_evidence_carrying_and_labeled_estimate():
line = coverage_line({
"total": 4573, "accounted": 3100, "unclassified": 1473,
"counts": {"canonical": 12, "instance": 2900, "variant": 0, "exempt": 188},
"estimate": True,
"computed_at": "2026-08-16T12:00:00+00:00",
"largest_gaps": [
{"dir": "internal/api", "unclassified": 40, "total": 60},
{"dir": "web/src/components", "unclassified": 25, "total": 30},
],
})
assert "3100/4573 shapes accounted for" in line
assert "12 canonical · 2900 instance · 188 exempt" in line # zero variant elided
assert "estimate" in line
assert "2026-08-16" in line
assert "1473 unclassified" in line
assert "internal/api, web/src/components" in line
def test_coverage_routes_are_registered():
from scribe.app import create_app
rules = {r.rule for r in create_app().url_map.iter_rules()}
assert "/api/projects/<int:project_id>/coverage" in rules
assert "/api/projects/<int:project_id>/coverage/refresh" in rules
# --- integration: the measurement end to end on real Postgres ----------------
def _forge(tar_bytes: bytes):
import httpx
from scribe.services.forge import GiteaForge
def handler(request: httpx.Request) -> httpx.Response:
path = request.url.path
if path == "/api/v1/repos/alice/widget":
return httpx.Response(200, json={"default_branch": "main"})
if path == "/api/v1/repos/alice/widget/archive/main.tar.gz":
return httpx.Response(200, content=tar_bytes)
return httpx.Response(404, json={"message": "not found"})
return GiteaForge(
"https://git.example.com", "tok", transport=httpx.MockTransport(handler)
)
def _selector(tar_bytes: bytes):
"""The keyring shape compute_coverage consumes since #2778 — one owner
keyring holding the mocked Gitea adapter."""
from scribe.services.forge import ForgeSelector
return ForgeSelector((_forge(tar_bytes),))
@pytest_asyncio.fixture
async def seeded(_dispose_engine):
"""User + project + binding + two snippets that cover 2 of TREE's 4 shapes."""
from scribe.models import async_session
from scribe.models.project import Project
from scribe.services import snippets as svc
from scribe.services.repo_bindings import set_binding
async with async_session() as s:
user = await ensure_user(s, "coverage_itest")
project = Project(user_id=user.id, title="Widget")
s.add(project)
await s.flush()
uid, pid = user.id, project.id
await s.commit()
await set_binding(uid, "https://git.example.com/alice/widget.git", pid)
await svc.create_snippet(
uid, name="cov_make_app", code="def make_app():\n pass\n",
language="python", repo="Widget", path="src/app.py",
symbol="make_app", project_id=pid,
)
await svc.create_snippet(
uid, name="cov_btn", code=".btn {\n color: red;\n}\n",
language="css", repo="Widget", path="web", symbol=".btn",
project_id=pid,
)
return {"uid": uid, "pid": pid}
@pytest.mark.integration
async def test_coverage_measures_the_tree_exactly_and_caches(seeded):
from scribe.services.coverage import (
cached_coverage,
compute_coverage,
refresh_coverage,
)
from sqlalchemy import select
from scribe.models import async_session
from scribe.models.code_shape import CodeShape
uid, pid = seeded["uid"], seeded["pid"]
selector = _selector(_tarball(TREE))
coverage = await compute_coverage(uid, pid, selector=selector)
assert coverage is not None
assert coverage["total"] == 4
assert coverage["accounted"] == 2
assert coverage["unclassified"] == 2
assert coverage["counts"] == {
"canonical": 2, "instance": 0, "variant": 0, "exempt": 0,
}
assert coverage["estimate"] is True
assert coverage["repos"] == [{
"repo": "git.example.com/alice/widget", "ref": "main",
"total": 4, "accounted": 2,
}]
assert coverage["largest_gaps"] == [
{"dir": "src", "unclassified": 2, "total": 3}
]
# The walk fed the LEDGER (#2788): every extracted shape has a row, the
# snippet reference locations are mechanically stamped canonical WITH
# their snippet id, and the rest sit in the todo state.
async with async_session() as s:
rows = (await s.execute(
select(CodeShape).where(CodeShape.project_id == pid)
)).scalars().all()
by_symbol = {r.symbol: r for r in rows}
assert set(by_symbol) == {"make_app", "Config", "helper", "btn"}
assert by_symbol["make_app"].status == "canonical"
assert by_symbol["make_app"].snippet_id is not None
assert by_symbol["make_app"].classified_by == "mechanical"
assert by_symbol["btn"].status == "canonical"
assert by_symbol["Config"].status == "unclassified"
assert by_symbol["helper"].status == "unclassified"
assert all(r.vanished_at is None for r in rows)
assert all(r.first_seen_commit for r in rows) # ref at minimum
# Idempotence: a second walk changes nothing about the readout.
again = await compute_coverage(uid, pid, selector=_selector(_tarball(TREE)))
assert (again["total"], again["accounted"]) == (4, 2)
# Nothing computed → nothing cached; refresh writes; the cache reads back
# byte-equal, because enter_project will serve exactly this.
assert await cached_coverage(uid, pid) is None
stored = await refresh_coverage(uid, pid, selector=selector)
assert (await cached_coverage(uid, pid)) == json.loads(json.dumps(stored))
@pytest.mark.integration
async def test_ledger_keeps_judgments_and_stamps_vanished_shapes(seeded):
"""The two survival rules (#2788): an agent's classification outlives
recompute, and a shape that leaves the tree is stamped vanished — kept
for history, dropped from the readout."""
from datetime import datetime, timezone
from sqlalchemy import select
from scribe.models import async_session
from scribe.models.code_shape import CodeShape
from scribe.services.coverage import compute_coverage
uid, pid = seeded["uid"], seeded["pid"]
await compute_coverage(uid, pid, selector=_selector(_tarball(TREE)))
# An agent judges `helper` a deliberate one-off.
async with async_session() as s:
helper = (await s.execute(select(CodeShape).where(
CodeShape.project_id == pid, CodeShape.symbol == "helper",
))).scalar_one()
helper.status = "exempt"
helper.reason = "test scaffolding, deliberately local"
helper.classified_by = "agent"
helper.classified_at = datetime.now(timezone.utc)
await s.commit()
# The tree moves on: util.py (helper) is gone entirely, app.py loses
# nothing. The judgment on `helper` must survive AS HISTORY (vanished,
# still exempt), never be reset by the sync.
smaller = {k: v for k, v in TREE.items() if k != "src/util.py"}
coverage = await compute_coverage(uid, pid, selector=_selector(_tarball(smaller)))
assert coverage["total"] == 3 # helper's row left the readout
assert coverage["accounted"] == 2
assert coverage["counts"]["exempt"] == 0 # vanished rows don't count
async with async_session() as s:
helper = (await s.execute(select(CodeShape).where(
CodeShape.project_id == pid, CodeShape.symbol == "helper",
))).scalar_one()
assert helper.vanished_at is not None
assert helper.status == "exempt" # the judgment is history, kept
assert helper.reason == "test scaffolding, deliberately local"
# And it returns: the shape reappearing clears the stamp, judgment intact.
coverage = await compute_coverage(uid, pid, selector=_selector(_tarball(TREE)))
assert coverage["total"] == 4
assert coverage["accounted"] == 3 # exempt counts as accounted again
assert coverage["counts"]["exempt"] == 1
@pytest.mark.integration
async def test_enter_project_surfaces_the_line_only_once_computed(seeded):
from scribe.mcp._context import _user_id_ctx
from scribe.mcp.tools.projects import enter_project
from scribe.services.coverage import refresh_coverage
uid, pid = seeded["uid"], seeded["pid"]
token = _user_id_ctx.set(uid)
try:
# Forge-less / never-computed instance: the key is present, null, and
# nothing else about the response changes.
before = await enter_project(project_id=pid)
assert before["pattern_coverage"] is None
await refresh_coverage(uid, pid, selector=_selector(_tarball(TREE)))
after = await enter_project(project_id=pid)
line = after["pattern_coverage"]
assert line.startswith(
"shape accounting: 2/4 shapes accounted for — 2 canonical "
"(estimate, computed "
)
assert line.endswith("; 2 unclassified, largest: src")
finally:
_user_id_ctx.reset(token)
@pytest.mark.integration
async def test_explicit_refresh_names_its_failures(seeded):
"""refresh_for_caller (#2802) raises fixable errors instead of silence:
an agent mid-task must learn WHY nothing measured — 'None' is exactly the
stranding the button-only path caused."""
from scribe.models import async_session
from scribe.services.coverage import refresh_for_caller
uid, pid = seeded["uid"], seeded["pid"]
# The owner has no forge connection rows → the error names the fix.
with pytest.raises(ValueError) as err:
await refresh_for_caller(uid, pid)
assert "Git Forges" in str(err.value)
# A stranger gets not-found/no-write, never a measurement.
async with async_session() as s:
other = await ensure_user(s, "coverage_outsider")
other_id = other.id
await s.commit()
with pytest.raises(ValueError) as err:
await refresh_for_caller(other_id, pid)
assert "no write access" in str(err.value)
@pytest.mark.integration
async def test_background_seed_is_quiet_without_a_forge(seeded):
"""refresh_if_stale (#2802) must exit silently for a forge-less owner —
rule #115's baseline — and treat a fresh cache as nothing-to-do."""
from scribe.services.coverage import refresh_coverage, refresh_if_stale
uid, pid = seeded["uid"], seeded["pid"]
# Absent cache + no forge rows: returns without raising, writes nothing.
await refresh_if_stale(uid, pid)
# Fresh cache: returns before ever consulting the keyring.
stored = await refresh_coverage(uid, pid, selector=_selector(_tarball(TREE)))
await refresh_if_stale(uid, pid, cached=stored)
@pytest.mark.integration
async def test_unservable_binding_measures_nothing(seeded):
"""A project bound only to a host the forge doesn't serve returns None —
the same silence as no forge at all, never an error."""
from scribe.services.coverage import compute_coverage
from scribe.services.repo_bindings import set_binding
from scribe.models import async_session
from scribe.models.project import Project
uid = seeded["uid"]
async with async_session() as s:
other = Project(user_id=uid, title="Elsewhere")
s.add(other)
await s.flush()
other_pid = other.id
await s.commit()
await set_binding(uid, "https://github.com/somebody/else.git", other_pid)
assert await compute_coverage(uid, other_pid, selector=_selector(_tarball(TREE))) is None
# --- #2792: fingerprints and the proposer's readout --------------------------
def test_extract_definitions_fingerprints_each_block():
"""The block rule across the language families the extractor knows: a
Python def ends at the next top-level statement, a braces/CSS block keeps
its closer, and comments/decorators don't move the hash."""
from scribe.services.coverage import extract_definitions
text = (
"import os\n\n"
"def a(x):\n # comment\n return x + 1\n\n\n"
"class B:\n def m(self):\n return 2\n\n"
".btn {\n color: red;\n}\n"
"export const f = (x) => {\n return x;\n};\n"
)
defs = {d.name: d for d in extract_definitions(text)}
assert set(defs) == {"a", "B", "m", "btn", "f"}
assert defs["a"].signature == "def a(x):"
assert defs["a"].body.startswith("def a(x):\n # comment\n return x + 1")
assert "class B" not in defs["a"].body
assert defs["B"].body.rstrip().endswith("return 2")
assert defs["btn"].body == ".btn {\n color: red;\n}"
assert defs["f"].body == "export const f = (x) => {\n return x;\n};"
assert all(len(d.body_sha) == 16 for d in defs.values())
# Comment changes don't change what the shape IS; code changes do.
again = {d.name: d for d in extract_definitions(text.replace("# comment", "# other"))}
assert again["a"].body_sha == defs["a"].body_sha
changed = {d.name: d for d in extract_definitions(text.replace("x + 1", "x + 2"))}
assert changed["a"].body_sha != defs["a"].body_sha
# And the identity view is unchanged for the hook mirror.
from scribe.services.coverage import extract_shapes
assert extract_shapes(text) == [(d.kind, d.name) for d in extract_definitions(text)]
def test_coverage_line_names_the_proposers_standing():
from scribe.services.coverage import coverage_line
base = {
"total": 100, "accounted": 10, "unclassified": 90,
"counts": {"canonical": 10, "instance": 0, "variant": 0, "exempt": 0},
"computed_at": "2026-08-21T00:00:00+00:00",
"largest_gaps": [{"dir": "src", "unclassified": 90, "total": 90}],
}
assert coverage_line(base).endswith("; 90 unclassified, largest: src")
line = coverage_line({**base, "proposed": 40, "derive_groups": [{"group": "a"}, {"group": "b"}]})
assert "; 90 unclassified (40 proposed, 2 derive groups), largest: src" in line
line = coverage_line({**base, "proposed": 0, "derive_groups": [{"group": "a"}]})
assert "(1 derive group)" in line
def test_coverage_line_names_divergence_and_recheck():
from scribe.services.coverage import coverage_line
base = {
"total": 100, "accounted": 40, "unclassified": 60,
"counts": {"canonical": 10, "instance": 30, "variant": 0, "exempt": 0},
"computed_at": "2026-08-21T00:00:00+00:00",
"largest_gaps": [{"dir": "src", "unclassified": 60, "total": 60}],
}
line = coverage_line({**base, "divergent": 2, "recheck": 1, "proposed": 5})
assert "; 60 unclassified (5 proposed, 2 DIVERGENT), largest: src" in line
assert line.endswith("; 1 judged shape changed since judged — recheck")
assert "DIVERGENT" not in coverage_line(base)
assert "recheck" not in coverage_line(base)