'
assert class_references("t/p.html", html) == {"row": 1, "col-2": 1, "y": 1}
# Not a template-bearing file: nothing, however it reads.
assert class_references("a/z.py", 'html = \'
\'') == {}
def test_scan_archive_returns_definitions_and_references_from_one_walk():
tree = dict(TREE)
tree["web/Card.vue"] = (
b'
x
\n'
b"\n"
)
scan = scan_archive(_tarball(tree))
assert isinstance(scan, ArchiveScan)
assert [(d.path, d.kind, d.name) for d in scan.definitions] == TREE_SHAPES + [
("web/Card.vue", "css", "card"),
]
# Only files whose markup names a class appear; the .py/.css files don't.
assert scan.references == {"web/Card.vue": {"btn": 1, "card": 1}}
assert shapes_from_archive(_tarball(tree)) == [(d.path, d.kind, d.name) for d in scan.definitions]
# --- 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_type_import_specifiers_are_not_definitions():
"""#2904: `import { type Foo, bar }` is the same two words as `type Foo =`
and defines nothing; only a `type` line with a declaration after the
name counts (TS alias, Go/Rust type)."""
from scribe.services.coverage import extract_shapes
src = (
'import { type DesignSystem, fetchDesignSystems } from "@/api/designSystems";\n'
'import { type Project } from "./x";\n'
"type Baz = { a: number };\n"
"type Wide
= T | null;\n"
"type Point struct {\n\tX int\n}\n"
)
assert extract_shapes(src) == [("sym", "Baz"), ("sym", "Wide"), ("sym", "Point")]
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_bind_repo_tool_takes_a_ref():
"""#2873: the binding names the branch the ledger follows."""
from scribe.mcp.server import build_mcp_server
tool = build_mcp_server()._tool_manager.get_tool("bind_repo")
assert "ref" in tool.parameters.get("properties", {})
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//coverage" in rules
assert "/api/projects//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 in ("/api/v1/repos/alice/widget/archive/main.tar.gz",
"/api/v1/repos/alice/widget/archive/dev.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, "scoped": 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}
]
# #2899: a first computation has no previous stamp — nothing is "new".
assert coverage["derive_new"] == {"count": 0, "examples": []}
# 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)]
# A CSS rule's fingerprint is its declarations (#2872): the same body
# under another selector is the same shape to the derive grouping.
css = ".closed-msg {\n text-align: center;\n padding: 0.5rem 0;\n}\n.error-block {\n text-align: center;\n padding: 0.5rem 0;\n}\n.other {\n text-align: left;\n}\n"
d = {x.name: x for x in extract_definitions(css)}
assert d["closed-msg"].body_sha == d["error-block"].body_sha != d["other"].body_sha
# One-line rules hash their own declarations — never the empty string
# (first deploy grouped 68 unrelated one-liners as one copy) — and a
# SINGLE declaration is not a shape (#2903): it keeps its selector in the
# hash, so `.a { color: red }` groups only with another `.a`, never with
# `.b { color: red }`. Two declarations and up stay selector-agnostic.
one = ".a { color: red; }\n\n.b { color: red; }\n\n.c { color: blue; }\n\n.a {\n color: red;\n}\n"
e = {x.name: x for x in extract_definitions(one)}
import hashlib
assert e["a"].body_sha != e["b"].body_sha != e["c"].body_sha
assert e["a"].body_sha != hashlib.sha1(b"").hexdigest()[:16]
two = ".a {\n color: red;\n margin: 0;\n}\n.b {\n color: red;\n margin: 0;\n}\n"
f = {x.name: x for x in extract_definitions(two)}
assert f["a"].body_sha == f["b"].body_sha
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
# Milestone 302: a css top copy says what renders it; unused classes
# join the standing block only when measured (None = no evidence).
line = coverage_line({**base, "unclassified": 0, "proposed": 0, "derive_groups": [
{"group": "name:css:error-msg", "label": ".error-msg", "files": 6,
"consumers": {"count": 6, "paths": ["a.vue"]}}], "unused_css": 3})
assert "top copy .error-msg ×6 files · used by 6 templates" in line
assert "3 unused classes" in line
line = coverage_line({**base, "unclassified": 0, "proposed": 0, "derive_groups": [
{"group": "name:css:x", "label": ".x", "files": 2,
"consumers": {"count": 1, "paths": ["a.vue"]}}], "unused_css": None})
assert "top copy .x ×2 files · used by 1 template" in line and "unused" not in line
# #2874: the next action on the line — biggest canon queue, widest copy.
line = coverage_line({
**base, "proposed": 40, "top_canon": {"snippet_id": 2844, "count": 78},
"derive_groups": [{"group": "dup:abc", "label": "closed-msg (identical body)", "files": 3}],
})
assert "top canon #2844 ×78" in line and "top copy closed-msg (identical body) ×3 files" in line
def test_coverage_line_shows_standing_work_even_with_nothing_unclassified():
"""#2899: since the scoped bucket a ledger can be fully accounted and
still carry derive groups / proposals / divergence — the line names
them as `standing:` instead of hiding them behind the todo count, and
names the drift since the previous refresh first-copy-first."""
from scribe.services.coverage import coverage_line
base = {
"total": 4693, "accounted": 4693, "unclassified": 0,
"counts": {"canonical": 37, "instance": 977, "variant": 73, "exempt": 1797, "scoped": 1809},
"computed_at": "2026-08-22T00:00:00+00:00", "largest_gaps": [],
}
quiet = coverage_line(base)
assert "unclassified" not in quiet and "standing" not in quiet
line = coverage_line({
**base,
"derive_groups": [{"group": "dup:abc", "label": "log-empty (identical body)", "files": 4}],
"derive_new": {"count": 2, "examples": [
{"label": ".error-msg", "path": "frontend/src/components/InceptionCard.vue", "group": "dup:9f0"},
{"label": ".error-msg", "path": "frontend/src/components/Other.vue", "group": "dup:9f0"},
]},
"divergent": 1,
})
assert "; standing: 1 derive group, +2 new copies since last refresh: .error-msg in " \
"frontend/src/components/InceptionCard.vue, 1 DIVERGENT, top copy log-empty (identical body) ×4 files" in line
assert "unclassified" not in line
# One copy reads singular; with a todo the block keeps its old place.
one = coverage_line({**base, "derive_new": {"count": 1, "examples": []}})
assert one.endswith("; standing: +1 new copy since last refresh")
todo = coverage_line({**base, "unclassified": 3, "accounted": 4690, "proposed": 2,
"derive_new": {"count": 1, "examples": [{"label": "x", "path": "a.py"}]},
"largest_gaps": [{"dir": "src", "unclassified": 3, "total": 9}]})
assert "; 3 unclassified (2 proposed, +1 new copy since last refresh: x in a.py), largest: src" in todo
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)
def test_scoped_definitions_are_vue_script_setup_and_scoped_style_only():
"""#2869: one-offs by construction — every sym in a .vue and every css
rule inside \n"
"\n"
)
defs = extract_definitions(vue)
names = {(d.kind, d.name) for d in defs}
assert {("sym", "load"), ("sym", "save"), ("css", "card"), ("css", "title"), ("css", "global-toast")} <= names
scoped = scoped_definitions("frontend/src/views/A.vue", vue, defs)
assert scoped == {("sym", "load"), ("sym", "save"), ("css", "card"), ("css", "title")}
# Definitions know their line, which is what the scoped-style range uses.
assert next(d for d in defs if d.name == "card").line > next(d for d in defs if d.name == "save").line
# Not a .vue: nothing is scoped, whatever it contains.
assert scoped_definitions("frontend/src/assets/components.css", ".card {\n x: 1;\n}\n",
extract_definitions(".card {\n x: 1;\n}\n")) == set()
assert scoped_definitions("src/a.py", "def load():\n pass\n", extract_definitions("def load():\n pass\n")) == set()
@pytest.mark.integration
async def test_binding_ref_is_the_branch_the_ledger_follows(seeded):
"""#2873: a binding that names a ref is read at that ref (not the forge's
default branch); "" clears it; None on a re-bind leaves it standing."""
from scribe.services.coverage import compute_coverage
from scribe.services.repo_bindings import bindings_for_project, set_binding
uid, pid = seeded["uid"], seeded["pid"]
b = await set_binding(uid, "https://git.example.com/alice/widget.git", pid, "dev")
assert b.ref == "dev"
coverage = await compute_coverage(uid, pid, selector=_selector(_tarball(TREE)))
assert coverage["repos"][0]["ref"] == "dev"
# A re-bind without a ref keeps it; "" clears it back to the default branch.
b = await set_binding(uid, "https://git.example.com/alice/widget.git", pid)
assert b.ref == "dev"
b = await set_binding(uid, "https://git.example.com/alice/widget.git", pid, "")
assert b.ref is None
assert [x.ref for x in await bindings_for_project(uid, pid)] == [None]
coverage = await compute_coverage(uid, pid, selector=_selector(_tarball(TREE)))
assert coverage["repos"][0]["ref"] == "main"