generated from john/python-template
Enforce the four unenforced reviewer checks with guard tests
The reviewer skill recorded four deterministic checks as unenforced or partial. Add tests so they fail the build instead of relying on a reviewer noticing. tests/test_model_contract_guards.py: - Status vocabulary: flags string literals compared against or assigned to status/purpose attributes, plus a narrower sweep that requires every status-valued literal in the package to be a known non-status use. - Relationship loading: every Relationship must declare lazy='raise' except documented exceptions, and the exception set must match the Relationship Loading Contract in docs/schema.md. - Schema fidelity: the Field-Accurate Table Contracts tables must match db/models.py on table coverage, field names, and declaration order, and the Authoritative Enumerations section must match the enum members. tests/test_orphan_sweep.py: - Locks the set of unreferenced public definitions. Route handlers registered by decorator are exempt, string entrypoint references count, and tests/ and tools/ count as consumers. KNOWN_ORPHANS records the four current orphans with rationale; a new one fails the build. Each guard was mutation-tested: reverting the fix below, dropping a documented field, widening a lazy strategy, and adding a stranded function each fail their respective test. Also fix the one violation the status guard found: sources_page.py compared attempt.status.value to the literal 'transcribed' instead of JobSourceStatus.TRANSCRIBED, which would survive an enum rename. Co-authored-by: Copilot App <[email protected]>
This commit is contained in:
Jim Lancaster
co-authored by
Copilot App
Copilot App
parent
5566f48fc0
commit
c6ed3126e0
@@ -0,0 +1,278 @@
|
||||
"""Deterministic guards for the model and persistence contracts.
|
||||
|
||||
These cover three checks that `.github/skills/python-code-reviewer/skill.md` requires
|
||||
on every review but that previously had no automated enforcement:
|
||||
|
||||
* **Status vocabulary conformance** - status comparisons and assignments must use the
|
||||
`JobStatus` / `JobSourceStatus` / `JobPurpose` enums rather than string literals.
|
||||
* **Relationship loading contract** - relationships declare ``lazy="raise"`` so read
|
||||
paths must eager-load explicitly, and the documented exceptions in `docs/schema.md`
|
||||
must match the code exactly.
|
||||
* **Schema contract fidelity** - the "Field-Accurate Table Contracts" tables in
|
||||
`docs/schema.md` must list exactly the fields each SQLModel table declares.
|
||||
|
||||
See `docs/requirements.md` (REQ-4-102), `docs/schema.md` ("Relationship Loading
|
||||
Contract"), and `.github/instructions/services.instructions.md`.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import ast
|
||||
import re
|
||||
from pathlib import Path
|
||||
|
||||
from sqlmodel import SQLModel
|
||||
|
||||
from transcription.db import models as models_module
|
||||
from transcription.db.models import JobPurpose
|
||||
from transcription.db.models import JobSourceStatus
|
||||
from transcription.db.models import JobStatus
|
||||
|
||||
PROJECT_ROOT = Path(__file__).resolve().parents[1]
|
||||
SOURCE_DIR = PROJECT_ROOT / "src" / "transcription"
|
||||
MODELS_PATH = SOURCE_DIR / "db" / "models.py"
|
||||
SCHEMA_DOC = PROJECT_ROOT / "docs" / "schema.md"
|
||||
|
||||
STATUS_ENUMS = (JobStatus, JobSourceStatus, JobPurpose)
|
||||
STATUS_VALUES = frozenset(member.value for enum in STATUS_ENUMS for member in enum)
|
||||
|
||||
# Attribute names that carry a status enum. A string literal compared against or
|
||||
# assigned to one of these is a stringly-typed status, even if it happens to match.
|
||||
STATUS_ATTRIBUTES = frozenset({"status", "purpose"})
|
||||
|
||||
# `JobSource.execution_attempts` loads attempt evidence on demand rather than raising,
|
||||
# because evidence is fetched deliberately by the services that own it. Documented in
|
||||
# `docs/schema.md` under "Relationship Loading Contract".
|
||||
DOCUMENTED_LOADING_EXCEPTIONS = {"execution_attempts": "noload"}
|
||||
|
||||
|
||||
def _python_files() -> list[Path]:
|
||||
return sorted(SOURCE_DIR.rglob("*.py"))
|
||||
|
||||
|
||||
def _relative(path: Path) -> str:
|
||||
return path.relative_to(PROJECT_ROOT).as_posix()
|
||||
|
||||
|
||||
def _is_status_target(node: ast.expr) -> bool:
|
||||
"""True for `x.status`, `x.purpose`, and their `.value` unwrappings."""
|
||||
if isinstance(node, ast.Attribute):
|
||||
if node.attr in STATUS_ATTRIBUTES:
|
||||
return True
|
||||
if node.attr == "value":
|
||||
return _is_status_target(node.value)
|
||||
return False
|
||||
|
||||
|
||||
def _string_constant(node: ast.expr) -> str | None:
|
||||
return node.value if isinstance(node, ast.Constant) and isinstance(node.value, str) else None
|
||||
|
||||
|
||||
def _status_literal_violations(tree: ast.Module) -> list[tuple[int, str]]:
|
||||
found: list[tuple[int, str]] = []
|
||||
for node in ast.walk(tree):
|
||||
if isinstance(node, ast.Compare):
|
||||
operands = [node.left, *node.comparators]
|
||||
if not any(_is_status_target(operand) for operand in operands):
|
||||
continue
|
||||
for operand in operands:
|
||||
literal = _string_constant(operand)
|
||||
if literal is not None:
|
||||
found.append((node.lineno, literal))
|
||||
elif isinstance(node, ast.Call):
|
||||
for keyword in node.keywords:
|
||||
if keyword.arg not in STATUS_ATTRIBUTES:
|
||||
continue
|
||||
literal = _string_constant(keyword.value)
|
||||
if literal is not None:
|
||||
found.append((node.lineno, literal))
|
||||
return found
|
||||
|
||||
|
||||
def test_status_enums_expose_expected_vocabulary():
|
||||
"""Guard the guard: the scan below is meaningless if the enums are empty."""
|
||||
assert {member.value for member in JobStatus} == {
|
||||
"queued",
|
||||
"processing",
|
||||
"transcribed",
|
||||
"partial_success",
|
||||
"failed",
|
||||
}
|
||||
assert {member.value for member in JobSourceStatus} == {
|
||||
"pending",
|
||||
"transcribed",
|
||||
"failed",
|
||||
"cancelled",
|
||||
}
|
||||
assert {member.value for member in JobPurpose} == {"transcription", "retranscription"}
|
||||
|
||||
|
||||
def test_no_stringly_typed_status_comparisons_or_assignments():
|
||||
"""Status handling must go through the enums, never raw strings.
|
||||
|
||||
`attempt.status.value == "transcribed"` silently survives an enum rename and
|
||||
compares a projection of the value rather than the value itself.
|
||||
"""
|
||||
violations: dict[str, list[tuple[int, str]]] = {}
|
||||
for path in _python_files():
|
||||
tree = ast.parse(path.read_text(encoding="utf-8"))
|
||||
found = _status_literal_violations(tree)
|
||||
if found:
|
||||
violations[_relative(path)] = found
|
||||
assert violations == {}
|
||||
|
||||
|
||||
def test_status_string_literals_outside_models_are_accounted_for():
|
||||
"""Any bare status-valued literal in the package must be a known non-status use.
|
||||
|
||||
This is deliberately narrower than the comparison scan: it catches literals that
|
||||
merely *look* like statuses, so a genuine new one cannot slip in unnoticed.
|
||||
"""
|
||||
allowed = {
|
||||
# `JobStatus` / `JobSourceStatus` / `JobPurpose` member definitions.
|
||||
"src/transcription/db/models.py",
|
||||
# `WorkerHealthState` is a separate Literal vocabulary that reuses "failed".
|
||||
"src/transcription/worker.py",
|
||||
# Distribution name lookups for `transcription`, not the JobPurpose member.
|
||||
"src/transcription/config.py",
|
||||
"src/transcription/providers/evidence.py",
|
||||
# UI placeholder copy where a value is absent, not a status render.
|
||||
"src/transcription/ui/pages/jobs_page.py",
|
||||
}
|
||||
unexpected: dict[str, list[tuple[int, str]]] = {}
|
||||
for path in _python_files():
|
||||
relative = _relative(path)
|
||||
if relative in allowed:
|
||||
continue
|
||||
tree = ast.parse(path.read_text(encoding="utf-8"))
|
||||
found = [
|
||||
(node.lineno, node.value)
|
||||
for node in ast.walk(tree)
|
||||
if isinstance(node, ast.Constant) and isinstance(node.value, str) and node.value in STATUS_VALUES
|
||||
]
|
||||
if found:
|
||||
unexpected[relative] = found
|
||||
assert unexpected == {}
|
||||
|
||||
|
||||
def _relationship_loading_strategies() -> dict[str, dict[str, str | None]]:
|
||||
"""Map each model attribute defined via `Relationship(...)` to its lazy strategy."""
|
||||
tree = ast.parse(MODELS_PATH.read_text(encoding="utf-8"))
|
||||
strategies: dict[str, dict[str, str | None]] = {}
|
||||
for class_node in tree.body:
|
||||
if not isinstance(class_node, ast.ClassDef):
|
||||
continue
|
||||
for statement in class_node.body:
|
||||
if not isinstance(statement, ast.AnnAssign) or statement.value is None:
|
||||
continue
|
||||
call = statement.value
|
||||
if not isinstance(call, ast.Call) or getattr(call.func, "id", None) != "Relationship":
|
||||
continue
|
||||
attribute = statement.target.id if isinstance(statement.target, ast.Name) else "<unknown>"
|
||||
lazy: str | None = None
|
||||
for keyword in call.keywords:
|
||||
if keyword.arg != "sa_relationship_kwargs" or not isinstance(keyword.value, ast.Dict):
|
||||
continue
|
||||
for key, value in zip(keyword.value.keys, keyword.value.values, strict=True):
|
||||
if isinstance(key, ast.Constant) and key.value == "lazy" and isinstance(value, ast.Constant):
|
||||
lazy = value.value
|
||||
strategies.setdefault(class_node.name, {})[attribute] = lazy
|
||||
return strategies
|
||||
|
||||
|
||||
def test_relationships_are_discovered():
|
||||
"""Guard the guard: the loading rules below are meaningless if nothing is scanned."""
|
||||
strategies = _relationship_loading_strategies()
|
||||
assert {"Document", "Job", "JobSource", "Source"} <= set(strategies)
|
||||
assert sum(len(attributes) for attributes in strategies.values()) >= 25
|
||||
|
||||
|
||||
def test_relationships_declare_lazy_raise_except_documented_cases():
|
||||
"""REQ-4-102: relationships raise on implicit load so read shape stays explicit."""
|
||||
violations: dict[str, str | None] = {}
|
||||
for model_name, attributes in _relationship_loading_strategies().items():
|
||||
for attribute, lazy in attributes.items():
|
||||
expected = DOCUMENTED_LOADING_EXCEPTIONS.get(attribute, "raise")
|
||||
if lazy != expected:
|
||||
violations[f"{model_name}.{attribute}"] = lazy
|
||||
assert violations == {}
|
||||
|
||||
|
||||
def test_documented_loading_exceptions_match_schema_doc():
|
||||
"""The exception list is only trustworthy while `docs/schema.md` agrees with it."""
|
||||
schema_text = SCHEMA_DOC.read_text(encoding="utf-8")
|
||||
contract = schema_text.split("## Relationship Loading Contract", 1)[1]
|
||||
for attribute, strategy in DOCUMENTED_LOADING_EXCEPTIONS.items():
|
||||
assert attribute in contract, f"{attribute} is exempted in code but not documented"
|
||||
assert f'`lazy="{strategy}"`' in contract
|
||||
|
||||
|
||||
def _documented_table_fields() -> dict[str, list[str]]:
|
||||
schema_text = SCHEMA_DOC.read_text(encoding="utf-8")
|
||||
section = schema_text.split("## Field-Accurate Table Contracts", 1)[1]
|
||||
documented: dict[str, list[str]] = {}
|
||||
for block in re.split(r"\n### ", section)[1:]:
|
||||
heading = block.splitlines()[0].strip().strip("`")
|
||||
documented[heading] = re.findall(r"^\| `([^`]+)` \|", block, re.MULTILINE)
|
||||
return documented
|
||||
|
||||
|
||||
def _table_models() -> dict[str, type[SQLModel]]:
|
||||
return {
|
||||
name: attribute
|
||||
for name, attribute in vars(models_module).items()
|
||||
if isinstance(attribute, type)
|
||||
and issubclass(attribute, SQLModel)
|
||||
and attribute is not SQLModel
|
||||
and getattr(attribute, "__table__", None) is not None
|
||||
}
|
||||
|
||||
|
||||
def test_schema_doc_documents_every_table_model():
|
||||
"""Every persisted table needs a field contract, and vice versa."""
|
||||
documented = set(_documented_table_fields())
|
||||
actual = set(_table_models())
|
||||
assert actual, "no table models discovered"
|
||||
assert documented - actual == set(), "schema.md documents tables that no longer exist"
|
||||
assert actual - documented == set(), "schema.md is missing tables that exist in models.py"
|
||||
|
||||
|
||||
def test_schema_doc_field_contracts_match_models():
|
||||
"""Check 6: `docs/schema.md` stays field-accurate with `db/models.py`."""
|
||||
documented = _documented_table_fields()
|
||||
drift: dict[str, dict[str, list[str]]] = {}
|
||||
for name, model in _table_models().items():
|
||||
expected = set(model.model_fields)
|
||||
listed = set(documented.get(name, []))
|
||||
if expected != listed:
|
||||
drift[name] = {
|
||||
"undocumented_fields": sorted(expected - listed),
|
||||
"stale_doc_entries": sorted(listed - expected),
|
||||
}
|
||||
assert drift == {}
|
||||
|
||||
|
||||
def test_schema_doc_lists_fields_in_declaration_order():
|
||||
"""Ordering drift is how a doc silently stops being reviewable against the model."""
|
||||
documented = _documented_table_fields()
|
||||
misordered = {
|
||||
name: {"documented": documented[name], "declared": list(model.model_fields)}
|
||||
for name, model in _table_models().items()
|
||||
if documented.get(name, []) != list(model.model_fields)
|
||||
}
|
||||
assert misordered == {}
|
||||
|
||||
|
||||
def test_schema_doc_enumerations_match_status_enums():
|
||||
"""The "Authoritative Enumerations" section must list the real members."""
|
||||
schema_text = SCHEMA_DOC.read_text(encoding="utf-8")
|
||||
section = schema_text.split("## Authoritative Enumerations", 1)[1].split("\n## ", 1)[0]
|
||||
drift: dict[str, dict[str, list[str]]] = {}
|
||||
for enum in STATUS_ENUMS:
|
||||
block = re.split(rf"\n### {enum.__name__}\n", section)
|
||||
assert len(block) == 2, f"{enum.__name__} has no section in docs/schema.md"
|
||||
listed = re.findall(r"^- `([^`]+)`", block[1].split("\n### ", 1)[0], re.MULTILINE)
|
||||
expected = [member.value for member in enum]
|
||||
if listed != expected:
|
||||
drift[enum.__name__] = {"documented": listed, "declared": expected}
|
||||
assert drift == {}
|
||||
@@ -0,0 +1,146 @@
|
||||
"""Deterministic orphan sweep for the `transcription` package.
|
||||
|
||||
`.github/skills/python-code-reviewer/skill.md` requires every review to report
|
||||
orphaned code as removed, retained-with-justification, or uncertain-follow-up. This
|
||||
guard makes that sweep reproducible: it locks the current set of unreferenced public
|
||||
definitions, so a newly stranded function fails the build instead of accumulating
|
||||
silently, and deleting a known orphan requires deleting its entry here.
|
||||
|
||||
The sweep is intentionally conservative. It only considers module-level public
|
||||
definitions, and it honours the dynamic-wiring exceptions the skill calls out:
|
||||
framework route registration, string-based entrypoint references, and use from
|
||||
`tests/` or `tools/`.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import ast
|
||||
from pathlib import Path
|
||||
|
||||
PROJECT_ROOT = Path(__file__).resolve().parents[1]
|
||||
SOURCE_DIR = PROJECT_ROOT / "src" / "transcription"
|
||||
|
||||
# Every tree that may legitimately consume package API.
|
||||
REFERENCE_ROOTS = (SOURCE_DIR, PROJECT_ROOT / "tests", PROJECT_ROOT / "tools")
|
||||
|
||||
# Decorators that hand a callable to a framework registry, making the definition
|
||||
# reachable without any in-repo reference to its name.
|
||||
REGISTRATION_DECORATOR_PREFIXES = ("router.", "app.", "ui.page")
|
||||
|
||||
# Confirmed orphans, retained by decision rather than by reference. Each entry needs a
|
||||
# rationale. Removing the code means removing the entry; adding an entry means an
|
||||
# explicit decision to keep unreferenced code.
|
||||
KNOWN_ORPHANS: dict[str, str] = {
|
||||
"BenchmarkManifest": (
|
||||
"Benchmark manifest model in benchmarking.py with no current caller. "
|
||||
"Uncertain - follow-up: confirm whether the benchmarking entrypoint is still "
|
||||
"intended before removing."
|
||||
),
|
||||
"dispose_all_engines": (
|
||||
"Engine lifecycle helper in db/engine.py. Uncertain - follow-up: operational "
|
||||
"teardown utility with no runtime or test caller."
|
||||
),
|
||||
"refresh_engine": (
|
||||
"Engine lifecycle helper in db/engine.py. Uncertain - follow-up: paired with "
|
||||
"dispose_all_engines and equally unreferenced."
|
||||
),
|
||||
"summarize_error": (
|
||||
"Error-presentation helper in ui/components/error_presenter.py that no page or "
|
||||
"component calls. Uncertain - follow-up: superseded by the presenter's other "
|
||||
"entrypoints."
|
||||
),
|
||||
}
|
||||
|
||||
|
||||
def _source_files() -> list[Path]:
|
||||
return sorted(SOURCE_DIR.rglob("*.py"))
|
||||
|
||||
|
||||
def _is_registered_with_framework(node: ast.FunctionDef | ast.AsyncFunctionDef | ast.ClassDef) -> bool:
|
||||
return any(
|
||||
ast.unparse(decorator).startswith(REGISTRATION_DECORATOR_PREFIXES) for decorator in node.decorator_list
|
||||
)
|
||||
|
||||
|
||||
def _public_definitions() -> dict[str, str]:
|
||||
"""Public module-level definitions, mapped to `path:line`."""
|
||||
definitions: dict[str, str] = {}
|
||||
for path in _source_files():
|
||||
tree = ast.parse(path.read_text(encoding="utf-8"))
|
||||
for node in tree.body:
|
||||
if not isinstance(node, ast.FunctionDef | ast.AsyncFunctionDef | ast.ClassDef):
|
||||
continue
|
||||
if node.name.startswith("_") or _is_registered_with_framework(node):
|
||||
continue
|
||||
definitions[node.name] = f"{path.relative_to(PROJECT_ROOT).as_posix()}:{node.lineno}"
|
||||
return definitions
|
||||
|
||||
|
||||
def _referenced_names() -> tuple[set[str], str]:
|
||||
"""Names referenced anywhere, plus every string literal joined for dotted lookups."""
|
||||
names: set[str] = set()
|
||||
literals: list[str] = []
|
||||
for root in REFERENCE_ROOTS:
|
||||
for path in sorted(root.rglob("*.py")):
|
||||
# This module names every known orphan in `KNOWN_ORPHANS`; counting those
|
||||
# strings as references would make the allowlist self-satisfying.
|
||||
if path == Path(__file__).resolve():
|
||||
continue
|
||||
tree = ast.parse(path.read_text(encoding="utf-8"))
|
||||
for node in ast.walk(tree):
|
||||
if isinstance(node, ast.Name):
|
||||
names.add(node.id)
|
||||
elif isinstance(node, ast.Attribute):
|
||||
names.add(node.attr)
|
||||
elif isinstance(node, ast.ImportFrom):
|
||||
for alias in node.names:
|
||||
names.add(alias.name)
|
||||
names.add(alias.asname or alias.name)
|
||||
elif isinstance(node, ast.Constant) and isinstance(node.value, str):
|
||||
literals.append(node.value)
|
||||
return names, "\n".join(literals)
|
||||
|
||||
|
||||
def _orphans() -> dict[str, str]:
|
||||
definitions = _public_definitions()
|
||||
names, literal_blob = _referenced_names()
|
||||
return {
|
||||
name: location
|
||||
for name, location in definitions.items()
|
||||
# A definition is referenced if its name is used directly, or appears inside a
|
||||
# string such as "transcription.__main__:create_cli_app".
|
||||
if name not in names and name not in literal_blob
|
||||
}
|
||||
|
||||
|
||||
def test_public_definitions_are_discovered():
|
||||
"""Guard the guard: the sweep is meaningless if nothing is scanned."""
|
||||
definitions = _public_definitions()
|
||||
assert len(definitions) >= 200
|
||||
assert "create_app" in definitions
|
||||
|
||||
|
||||
def test_framework_registered_routes_are_exempt():
|
||||
"""Route handlers are reachable via decorator registration, not by name."""
|
||||
definitions = _public_definitions()
|
||||
assert "healthz_route" not in definitions
|
||||
assert "read_document_source_media" not in definitions
|
||||
|
||||
|
||||
def test_no_unexpected_orphaned_definitions():
|
||||
"""Check 10: no public definition becomes unreferenced without a recorded decision."""
|
||||
unexpected = {name: location for name, location in _orphans().items() if name not in KNOWN_ORPHANS}
|
||||
assert unexpected == {}
|
||||
|
||||
|
||||
def test_known_orphans_are_still_orphaned():
|
||||
"""Keep the allowlist honest: an entry that regained callers must be removed."""
|
||||
current = set(_orphans())
|
||||
stale = sorted(name for name in KNOWN_ORPHANS if name not in current)
|
||||
assert stale == [], "these definitions are referenced again; drop them from KNOWN_ORPHANS"
|
||||
|
||||
|
||||
def test_known_orphans_document_a_rationale():
|
||||
"""An allowlist without reasons is just suppressed output."""
|
||||
missing = sorted(name for name, reason in KNOWN_ORPHANS.items() if len(reason.strip()) < 40)
|
||||
assert missing == []
|
||||
Reference in New Issue
Block a user