Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
61 changes: 42 additions & 19 deletions src/cf_quality/exemptions.py
Original file line number Diff line number Diff line change
Expand Up @@ -61,6 +61,7 @@
import subprocess # fold-in scripts run via sys.executable, fixed argv, no shell (S603 gated below)
import sys
import tokenize
from collections.abc import Sequence
from pathlib import Path
from typing import Any

Expand Down Expand Up @@ -167,28 +168,40 @@ def _type_ignore_suppressions(
]


def _scan_src(root: Path) -> tuple[list[Suppression], list[GateViolation]]:
def _surface_files(surface: Sequence[Path]) -> list[Path]:
"""Every ``*.py`` the scanner opens, from an ALREADY-resolved surface."""
files: list[Path] = []
for base in surface:
files.extend([base] if base.is_file() else sorted(base.rglob("*.py")))
return files


def _scan_src(
root: Path, files: Sequence[Path] | None = None
) -> tuple[list[Suppression], list[GateViolation]]:
"""Scan the discovered Python surface for suppression comments.

The ``(suppressions, violations)`` ARITY is a published contract, not an
implementation detail: a consumer's drift-proof pin unpacks exactly two
values from this to cross-check its mirrored resolver (see
:func:`_matches`). The anchor audit therefore takes its surface from
:mod:`cf_quality.exemption_surface` rather than riding a third element.

``files`` threads in the file list a caller already resolved, so the surface
is discovered ONCE and the count a caller reports is this very walk. Omitted
— the one-argument shape the consumer pin calls — it resolves its own.
"""
if files is None:
files = _surface_files(discover_scan_paths(root))
suppressions: list[Suppression] = []
violations: list[GateViolation] = []
for base in discover_scan_paths(root):
files = [base] if base.is_file() else sorted(base.rglob("*.py"))
for file_path in files:
rel_path = file_path.relative_to(root).as_posix()
spans = symbol_spans(file_path)
for line, text in _comment_tokens(file_path):
found, broken = _classify_comment(
rel_path, line, text, enclosing_symbol(spans, line)
)
suppressions.extend(found)
violations.extend(broken)
for file_path in files:
rel_path = file_path.relative_to(root).as_posix()
spans = symbol_spans(file_path)
for line, text in _comment_tokens(file_path):
found, broken = _classify_comment(rel_path, line, text, enclosing_symbol(spans, line))
suppressions.extend(found)
violations.extend(broken)
return suppressions, violations


Expand Down Expand Up @@ -285,9 +298,12 @@ def _ratchet_report(entry_count: int, frozen_count: int) -> tuple[list[GateViola
return violations, lines


def check(root: Path) -> tuple[list[GateViolation], list[str]]:
"""Run checks (a)-(e) against a repo root; returns (violations, report lines)."""
suppressions, violations = _scan_src(root)
def check(root: Path) -> tuple[list[GateViolation], list[str], dict[str, int]]:
"""Run checks (a)-(e) — (violations, report lines, the counts measured)."""
surface = tuple(discover_scan_paths(root))
files = _surface_files(surface)
suppressions, violations = _scan_src(root, files)
counts = {"suppressions": len(suppressions), "files_scanned": len(files), "entries": 0}
config = _load_config(root)
if config is None:
if suppressions:
Expand All @@ -296,16 +312,17 @@ def check(root: Path) -> tuple[list[GateViolation], list[str]]:
message="gated suppressions found in src/ but exemptions.json is missing",
context={"suppressions": len(suppressions)},
)
return violations, ["no exemptions.json and no gated suppressions — nothing to register"]
quiet = "no exemptions.json and no gated suppressions — nothing to register"
return violations, [quiet], counts
entries, frozen_count = config
surface = tuple(discover_scan_paths(root))
counts["entries"] = len(entries)
anchors = exemption_anchors.audit(root, entries, suppressions, surface)
match_violations, registered_lines = _match_suppressions(suppressions, entries, anchors.rotted)
violations.extend(anchors.violations)
violations.extend(match_violations)
ratchet_violations, lines = _ratchet_report(len(entries), frozen_count)
violations.extend(ratchet_violations)
return violations, [*lines, *anchors.notices, *registered_lines]
return violations, [*lines, *anchors.notices, *registered_lines], counts


def _unregistered_violation(
Expand Down Expand Up @@ -446,7 +463,7 @@ def main(argv: list[str] | None = None) -> int:
args = parser.parse_args(argv)
root = Path(args.root).resolve()
try:
violations, lines = check(root)
violations, lines, counts = check(root)
fold_in_lines, fold_in_exit = _run_fold_ins(root)
except GateError as error:
return print_verdict("cf-exemptions", [], error)
Expand All @@ -456,6 +473,12 @@ def main(argv: list[str] | None = None) -> int:
"cf-exemptions",
violations,
notices=notices,
measured=(
f"— read {counts['files_scanned']} Python file(s), found "
f"{counts['suppressions']} suppression(s) against "
f"{counts['entries']} registered entry(ies)"
),
evidence=counts,
clean_summary="cf-exemptions: OK",
fail_summary=f"cf-exemptions: FAIL ({len(violations)} violation(s))",
)
Expand Down
41 changes: 29 additions & 12 deletions src/cf_quality/file_budget.py
Original file line number Diff line number Diff line change
Expand Up @@ -52,6 +52,7 @@
from typing import Any

from cf_quality.errors import GateError, GateViolation
from cf_quality.reporting import print_verdict

NEW_FILE_BUDGET = 500
SKIP_DIR_NAMES = {"__pycache__", "node_modules", "build", "dist", "venv"}
Expand Down Expand Up @@ -229,8 +230,8 @@ def _check_packages(budget: Budget, measured: dict[str, int]) -> list[GateViolat
return violations


def check_tree(root: Path, budget: Budget) -> tuple[list[GateViolation], list[str]]:
"""Measure the tree against the baseline; return (violations, notices)."""
def check_tree(root: Path, budget: Budget) -> tuple[list[GateViolation], list[str], int]:
"""Measure the tree; return (violations, notices, the count of files measured)."""
measured = {
path.relative_to(root).as_posix(): measure_file(path) for path in iter_python_files(root)
}
Expand All @@ -246,7 +247,7 @@ def check_tree(root: Path, budget: Budget) -> tuple[list[GateViolation], list[st
if entry.frozen_lines is not None and rel not in measured:
notices.append(f"shrink: {rel} {entry.frozen_lines} -> deleted (drop from baseline)")
violations.extend(_check_packages(budget, measured))
return violations, notices
return violations, notices, len(measured)


def init_tree(root: Path) -> dict[str, Any]:
Expand All @@ -264,15 +265,31 @@ def init_tree(root: Path) -> dict[str, Any]:


def _run_check(root: Path, config: Path) -> int:
violations, notices = check_tree(root, load_budget(config))
for notice in notices:
print(notice)
if violations:
report = {"gate": "cf-file-budget", "violations": [v.to_dict() for v in violations]}
print(json.dumps(report, indent=2))
return 1
print("cf-file-budget: clean")
return 0
budget = load_budget(config)
violations, notices, files = check_tree(root, budget)
# Only an entry carrying a line count is FROZEN (shrink-only). A
# declared-not-banned entry is a purpose with no ceiling — _check_file
# treats it exactly like an undeclared file — so counting the whole
# baseline as "frozen" would claim a ratchet that is not being enforced.
frozen = sum(1 for entry in budget.files.values() if entry.frozen_lines is not None)
return print_verdict(
"cf-file-budget",
violations,
notices=notices,
measured=(
f"— measured {files} file(s) against {frozen} frozen file entry(ies), "
f"{len(budget.files) - frozen} declared-not-banned entry(ies) "
f"and {len(budget.packages)} frozen package budget(s)"
),
evidence={
"files_measured": files,
"frozen_files": frozen,
"declared_files": len(budget.files) - frozen,
"frozen_packages": len(budget.packages),
},
clean_summary="cf-file-budget: clean",
fail_summary=f"cf-file-budget: FAIL ({len(violations)} violation(s))",
)


def _run_init(root: Path, config: Path) -> int:
Expand Down
14 changes: 7 additions & 7 deletions src/cf_quality/gate_runner.py
Original file line number Diff line number Diff line change
Expand Up @@ -189,22 +189,23 @@ def _structured_verdict(gate: str, data: Mapping[str, Any]) -> GateVerdict:
"""Build a verdict from a cf-* gate's JSON — full shape or the subset.

Honours an ``error`` field so a gate that exits 2 under its own GateError
contract resolves to a GateError verdict; the subset (file_budget) has none.
contract resolves to a GateError verdict; a partial emitter's has none.
"""
violations = [_violation_from_dict(item) for item in data.get("violations", [])]
return GateVerdict(
gate=data.get("gate", gate),
violations=violations,
violations=[_violation_from_dict(item) for item in data.get("violations", [])],
error=_error_from_dict(data.get("error")),
notices=list(data.get("notices", [])),
evidence=dict(data.get("evidence", {})),
)


def _verdict_from_proc(gate: str, proc: subprocess.CompletedProcess[str]) -> GateVerdict:
"""Parse a cf-* gate's output into a verdict, robust to its three wire shapes.

1. a full GateVerdict JSON (or any object carrying ``error``) → use it;
2. the ``{gate, violations}`` subset (file_budget) → verdict, error=None;
3. non-JSON human text (e.g. sticky-check) → exit-code semantics: rc 0 is
2. the ``{gate, violations}`` subset (a partial emitter) → verdict, error=None;
3. non-JSON human text (a foreign emitter) → exit-code semantics: rc 0 is
clean, any non-zero is one violation carrying the output (NOT a GateError
— unparseable output is not a gate-could-not-run condition).
"""
Expand All @@ -225,8 +226,7 @@ def _run_cf_gate(
) -> GateVerdict:
"""Run a cf-* console gate in JSON mode and parse its verdict."""
sub_env = {**env, "CF_QUALITY_JSON": "1"}
proc = _exec([str(_tool(gate)), *args], cwd, sub_env)
return _verdict_from_proc(gate, proc)
return _verdict_from_proc(gate, _exec([str(_tool(gate)), *args], cwd, sub_env))


# --- the stages (declared once, mirroring quality-gate.yml) -----------------
Expand Down
40 changes: 33 additions & 7 deletions src/cf_quality/import_contract.py
Original file line number Diff line number Diff line change
Expand Up @@ -259,17 +259,19 @@ def _is_dynamic_import(call: ast.Call) -> bool:
return isinstance(func, ast.Attribute) and func.attr == "import_module"


def scan_dynamic_imports(root: Path) -> list[GateViolation]:
"""PASS 3 — module-level dynamic imports inside contract-protected layers."""
def scan_dynamic_imports(root: Path) -> tuple[list[GateViolation], int]:
"""PASS 3 — module-level dynamic imports; findings and the modules scanned."""
root = root.resolve()
table = _load_contract_table(root)
if table is None:
return []
return [], 0
source_root = resolve_source_root(root)
protected = _named_top_level(_contracts(table)) & set(_top_level_packages(source_root))
violations: list[GateViolation] = []
scanned = 0
for package in sorted(protected):
for path in sorted((source_root / package).rglob("*.py")):
scanned += 1
rel = path.relative_to(root).as_posix()
violations.extend(
GateViolation(
Expand All @@ -284,7 +286,7 @@ def scan_dynamic_imports(root: Path) -> list[GateViolation]:
for call in _module_level_calls(_read_tree(path))
if _is_dynamic_import(call)
)
return violations
return violations, scanned


def _run_lint_imports(root: Path) -> tuple[int, str]:
Expand Down Expand Up @@ -353,12 +355,29 @@ def _edge_violations(code: str, summary: str, output: str) -> list[GateViolation
return [_contract_violation(code, f"{summary}: {edge}", {"edge": edge}) for edge in edges]


def _measured(clauses: int, modules: int) -> str:
"""The denominator — two DIFFERENT surfaces, each named for what it is.

``clauses`` is what PASS 1 linted in the committed contract file; ``modules``
is what PASS 3's AST walk opened (packages both named in a contract and
present as top-level). PASS 2 delegates to lint-imports and reports neither,
so the line states each number separately rather than saying one covers the
other — a contract naming no top-level package genuinely lints clauses while
scanning zero modules, and that is not a hole, it is two surfaces.
"""
return f"— linted {clauses} contract clause(s); scanned {modules} module(s) for dynamic imports"


def _missing_contract_verdict(root: Path) -> int:
packages = _top_level_packages(resolve_source_root(root))
measured = _measured(0, 0)
evidence = {"contract_clauses": 0, "modules_scanned": 0}
if not packages:
return print_verdict(
"cf-import-contract",
[],
measured=measured,
evidence=evidence,
clean_summary="cf-import-contract: OK (no top-level packages, no contract required)",
)
violation = _contract_violation(
Expand All @@ -367,7 +386,7 @@ def _missing_contract_verdict(root: Path) -> int:
"contract ([tool.importlinter] in pyproject.toml)",
{"packages": packages},
)
return print_verdict("cf-import-contract", [violation])
return print_verdict("cf-import-contract", [violation], measured=measured, evidence=evidence)


_CLEAN_SUMMARY = (
Expand Down Expand Up @@ -411,10 +430,17 @@ def _run_gate(root: Path) -> int:
return _missing_contract_verdict(root)
violations = lint_contract(root)
violations += _pass_two(root, tolerate_config_error=bool(violations))
violations += scan_dynamic_imports(root)
dynamic, modules = scan_dynamic_imports(root)
violations += dynamic
clauses = len(_contracts(table))
notices: list[str] = [] if violations else _honored_notices(table)
return print_verdict(
"cf-import-contract", violations, notices=notices, clean_summary=_CLEAN_SUMMARY
"cf-import-contract",
violations,
notices=notices,
measured=_measured(clauses, modules),
evidence={"contract_clauses": clauses, "modules_scanned": modules},
clean_summary=_CLEAN_SUMMARY,
)


Expand Down
10 changes: 6 additions & 4 deletions src/cf_quality/mirror_check.py
Original file line number Diff line number Diff line change
Expand Up @@ -279,8 +279,8 @@ def check_mirrors(
*,
max_pin_age_days: int = DEFAULT_MAX_PIN_AGE_DAYS,
today: dt.date | None = None,
) -> list[GateViolation]:
"""Check every declared mirror in ``repo/MIRRORS.md``; return all findings."""
) -> tuple[list[GateViolation], int]:
"""Check every declared mirror in ``repo/MIRRORS.md``findings and row count."""
mirrors_path = repo / "MIRRORS.md"
if not mirrors_path.is_file():
raise GateError(
Expand All @@ -293,7 +293,7 @@ def check_mirrors(
violations: list[GateViolation] = []
for row in rows:
violations.extend(_check_row(repo, row, max_pin_age_days, effective_today))
return violations
return violations, len(rows)


def render_template() -> str:
Expand Down Expand Up @@ -345,12 +345,14 @@ def main(argv: list[str] | None = None) -> int:
path = init_mirrors(repo)
print(f"wrote {path}")
return 0
violations = check_mirrors(repo, max_pin_age_days=args.max_pin_age_days)
violations, rows = check_mirrors(repo, max_pin_age_days=args.max_pin_age_days)
except GateError as error:
return print_verdict("cf-mirror-check", [], error)
return print_verdict(
"cf-mirror-check",
violations,
measured=f"— checked {rows} declared mirror row(s) in MIRRORS.md",
evidence={"mirror_rows": rows},
clean_summary="cf-mirror-check: OK",
fail_summary=f"cf-mirror-check: FAIL ({len(violations)} violation(s))",
)
Expand Down
Loading