feat(trim): R4 delete scripts/audit/ (no consumer found)

Verified no consumer of scripts/audit/ exists:

- grep across scripts/, plugins/, tests/, catalog/, .github/ for any
  import of scripts/audit or scripts/audit — only references are
  scripts/audit/ itself (intra-package) and tests/test_audit_* (the
  tests we're deleting).
- scripts/heretek_cli.py and scripts/catalog_updater.py have no
  'audit' subcommand.

Deleted:
- scripts/audit/ (7 .py + __init__ + __pycache__)
- tests/test_audit_*.py (8 files)

Verified:
- pytest: 414 pass (was 458; 44 audit tests removed)
- python scripts/validate.py: clean
- python scripts/generate_marketplace.py: clean (no drift)

PR2 of 5 in 2026-08-10-heretek-harness-trim spec.

Co-Authored-By: Claude <[email protected]>
This commit is contained in:
John Smith
2026-08-10 23:56:58 -04:00
co-authored by Claude
parent 7a90754921
commit 76132d0020
15 changed files with 0 additions and 1889 deletions
View File
-134
View File
@@ -1,134 +0,0 @@
"""Finding dataclass + load/save helpers for the harness self-audit.
One Finding = one card per the spec's per-finding schema. Cluster
agents emit YAML or JSON lists; this module loads them, validates each
card via `audit.validate`, and exposes a dataclass for synthesis / issues
to consume without re-parsing the schema each time.
JSON is the canonical output format (so downstream tools and the GitHub
issue builder don't have to handle two parsers).
"""
from __future__ import annotations
import json
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any
from ruamel.yaml import YAML
from audit import validate
SEVERITY_ORDER = ["critical", "high", "medium", "low", "info"]
@dataclass(frozen=True)
class Evidence:
code_refs: list[str]
file: str
line_range: list[int]
metric: str
@classmethod
def from_dict(cls, d: dict[str, Any]) -> "Evidence":
return cls(
code_refs=list(d["code_refs"]),
file=d["file"],
line_range=list(d["line_range"]),
metric=d["metric"],
)
def to_dict(self) -> dict[str, Any]:
return {
"code_refs": list(self.code_refs),
"file": self.file,
"line_range": list(self.line_range),
"metric": self.metric,
}
@dataclass(frozen=True)
class Finding:
finding_id: str
cluster: str
principle: str
severity: str
adversarial_posture: str
evidence: Evidence
failure_scenario: str
recommended_action: str
rationale: str
principle_reference: str
drift_signals: list[str] = field(default_factory=list)
@classmethod
def from_dict(cls, d: dict[str, Any]) -> "Finding":
return cls(
finding_id=d["finding_id"],
cluster=d["cluster"],
principle=d["principle"],
severity=d["severity"],
adversarial_posture=d["adversarial_posture"],
evidence=Evidence.from_dict(d["evidence"]),
failure_scenario=d["failure_scenario"],
recommended_action=d["recommended_action"],
rationale=d["rationale"],
principle_reference=d["principle_reference"],
drift_signals=list(d.get("drift_signals", [])),
)
def to_dict(self) -> dict[str, Any]:
return {
"finding_id": self.finding_id,
"cluster": self.cluster,
"principle": self.principle,
"severity": self.severity,
"adversarial_posture": self.adversarial_posture,
"evidence": self.evidence.to_dict(),
"failure_scenario": self.failure_scenario,
"recommended_action": self.recommended_action,
"rationale": self.rationale,
"principle_reference": self.principle_reference,
"drift_signals": list(self.drift_signals),
}
def _load_instance(path: Path) -> Any:
text = path.read_text()
if path.suffix.lower() in {".yaml", ".yml"}:
result = YAML(typ="safe").load(text)
if result is None:
raise ValueError(f"empty or whitespace-only YAML in {path}")
return result
return json.loads(text)
def load_findings(path: Path) -> list[Finding]:
"""Read a YAML or JSON file containing one finding or a list; validate each card."""
instance = _load_instance(path)
findings_in = instance if isinstance(instance, list) else [instance]
invalid_indexes: list[str] = []
out: list[Finding] = []
for i, d in enumerate(findings_in):
errors = validate.validate_finding(d)
if errors:
invalid_indexes.append(f"#{i}: " + "; ".join(errors))
continue
out.append(Finding.from_dict(d))
if invalid_indexes:
raise ValueError(
f"{path}: {len(invalid_indexes)} invalid finding(s): " + " | ".join(invalid_indexes)
)
return out
def save_findings(items: list[Finding], path: Path) -> None:
"""Write findings as a JSON array (canonical output format)."""
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(json.dumps([f.to_dict() for f in items], indent=2, sort_keys=True) + "\n")
def severity_rank(finding: Finding) -> int:
"""Lower rank = more severe. critical=0, high=1, medium=2, low=3, info=4."""
return SEVERITY_ORDER.index(finding.severity)
-178
View File
@@ -1,178 +0,0 @@
"""CLI driver for the harness self-audit toolkit.
Subcommands
-----------
emit-prompts Write one <letter>.txt prompt per cluster to --out-dir.
synthesize Run synthesis pass (validate + dedupe + report + JSON).
build-issues Read synthesis JSON, write GitHub-MCP-ready payload file.
run-all Convenience: synthesize only (MCP is operator-driven).
The driver does NOT call GitHub MCP directly -- payload creation is the
safe, testable part; the actual mcp__github__github-issue_write calls
happen in the operator's session when MCP is available.
"""
from __future__ import annotations
import argparse
import json
from pathlib import Path
from audit import findings as findings_mod
from audit import issues as issues_mod
from audit import prompts as prompts_mod
from audit import synthesis as synthesis_mod
# ---------------------------------------------------------------------------
# Subcommand: emit-prompts
# ---------------------------------------------------------------------------
def _cmd_emit_prompts(args: argparse.Namespace) -> int:
"""Write one ``<letter>.txt`` per cluster to *out_dir*."""
out_dir: Path = args.out_dir
out_dir.mkdir(parents=True, exist_ok=True)
for letter in sorted(prompts_mod.CLUSTERS):
text = prompts_mod.render_prompt(letter, args.repo_root, args.commit_sha)
(out_dir / f"{letter}.txt").write_text(text)
return 0
# ---------------------------------------------------------------------------
# Subcommand: synthesize
# ---------------------------------------------------------------------------
def _cmd_synthesize(args: argparse.Namespace) -> int:
"""Run synthesis; write report + JSON to *output_dir*."""
sonar_exclusions: set[str] | None = None
if args.sonar_exclusions_file:
raw = Path(args.sonar_exclusions_file).read_text()
sonar_exclusions = {
line.strip()
for line in raw.splitlines()
if line.strip() and not line.strip().startswith("#")
}
result = synthesis_mod.synthesize(
cluster_results_dir=args.cluster_results,
output_dir=args.output_dir,
commit_sha=args.commit_sha,
sonar_exclusions=sonar_exclusions,
)
print(
f"Synthesis complete: {len(result.findings)} findings, "
f"{result.duplicate_count} duplicates removed"
)
return 0
# ---------------------------------------------------------------------------
# Subcommand: build-issues
# ---------------------------------------------------------------------------
def _cmd_build_issues(args: argparse.Namespace) -> int:
"""Read synthesis JSON, write GitHub-MCP-ready payload file."""
loaded = findings_mod.load_findings(args.findings_json)
payloads = issues_mod.build_issue_payloads(loaded)
# If --report-link provided, append to each body
if args.report_link:
for p in payloads:
p.body += f"\n\n---\n*Audit report: {args.report_link}*"
out_path: Path = args.output
out_path.parent.mkdir(parents=True, exist_ok=True)
out_path.write_text(
json.dumps(
[{"title": p.title, "body": p.body, "labels": p.labels} for p in payloads],
indent=2,
sort_keys=True,
)
+ "\n"
)
print(f"Wrote {len(payloads)} issue payloads to {out_path}")
print(f" repo: {issues_mod.REPO} (use GitHub MCP, not gh CLI)")
return 0
# ---------------------------------------------------------------------------
# Subcommand: run-all
# ---------------------------------------------------------------------------
def _cmd_run_all(args: argparse.Namespace) -> int:
"""Convenience: synthesize only (issue creation is operator-driven).
Derives ``--commit-sha`` from ``git rev-parse HEAD`` in *repo_root*
and expects cluster results pre-staged at ``<repo-root>/audit/cluster-results/``.
"""
import subprocess
repo_root: Path = args.repo_root
sha = subprocess.check_output(
["git", "-C", str(repo_root), "rev-parse", "HEAD"],
text=True,
).strip()
cluster_results = repo_root / "audit" / "cluster-results"
synth_args = argparse.Namespace(
cluster_results=cluster_results,
output_dir=args.output_dir,
commit_sha=sha,
sonar_exclusions_file=args.sonar_exclusions_file,
)
return _cmd_synthesize(synth_args)
# ---------------------------------------------------------------------------
# CLI entry-point
# ---------------------------------------------------------------------------
def main(argv: list[str] | None = None) -> int:
"""Parse *argv* (or ``sys.argv[1:]``) and dispatch to a subcommand."""
parser = argparse.ArgumentParser(
prog="harness_self",
description="CLI driver for the heretek harness self-audit.",
)
sub = parser.add_subparsers(dest="command", required=True)
# -- emit-prompts -------------------------------------------------------
ep = sub.add_parser("emit-prompts", help="Write one prompt file per cluster.")
ep.add_argument("--repo-root", type=Path, required=True)
ep.add_argument("--commit-sha", required=True)
ep.add_argument("--out-dir", type=Path, required=True)
# -- synthesize ---------------------------------------------------------
sp = sub.add_parser("synthesize", help="Run synthesis pass.")
sp.add_argument("--cluster-results", type=Path, required=True)
sp.add_argument("--output-dir", type=Path, required=True)
sp.add_argument("--commit-sha", required=True)
sp.add_argument("--sonar-exclusions-file", type=Path, default=None)
# -- build-issues -------------------------------------------------------
bp = sub.add_parser("build-issues", help="Build GitHub-MCP-ready payload.")
bp.add_argument("--findings-json", type=Path, required=True)
bp.add_argument("--output", type=Path, required=True)
bp.add_argument("--report-link", default=None)
# -- run-all ------------------------------------------------------------
ra = sub.add_parser(
"run-all",
help="Convenience: synthesize only (MCP is operator-driven).",
)
ra.add_argument("--repo-root", type=Path, required=True)
ra.add_argument("--output-dir", type=Path, required=True)
ra.add_argument("--sonar-exclusions-file", type=Path, default=None)
args = parser.parse_args(argv)
dispatch = {
"emit-prompts": _cmd_emit_prompts,
"synthesize": _cmd_synthesize,
"build-issues": _cmd_build_issues,
"run-all": _cmd_run_all,
}
return dispatch[args.command](args)
if __name__ == "__main__":
raise SystemExit(main())
-156
View File
@@ -1,156 +0,0 @@
"""Build GitHub issue payloads from audit findings.
Per spec §Auto-issue creation rules: HIGH+CRITICAL only, 5/cluster cap,
1 umbrella issue per cluster for overflow. This module only builds payloads;
the actual mcp__github__github-issue_write calls happen in the driver or
by the operator -- keeping this module network-free and testable.
"""
from __future__ import annotations
from dataclasses import dataclass, field
from typing import List
from audit.findings import Finding
REPO: str = "Heretek-AI/heretek-claude-harness"
BASE_LABELS: List[str] = [
"audit",
"harness-self-audit",
"principles-audit",
"audit-2026-08-09",
]
_AUDITABLE_SEVERITIES: set[str] = {"critical", "high"}
# Short slug for issue title prefix, keyed by cluster name.
_CLUSTER_SHORT: dict[str, str] = {
"Readability & quality bar": "readability",
"Design & architecture": "design",
"Correctness & safety": "correctness",
"Testing & verification": "testing",
"Operations & docs": "ops-docs",
}
@dataclass
class IssuePayload:
"""Ready-to-file GitHub issue payload."""
title: str
body: str
labels: List[str] = field(default_factory=list)
def _title_for(f: Finding) -> str:
"""Build issue title from a single finding."""
short = _CLUSTER_SHORT.get(f.cluster, f.cluster.lower().replace(" ", "-")[:20])
return f"[audit:spec-1:{short}] {f.principle}"
def _body_for(f: Finding) -> str:
"""Build markdown issue body for a single finding."""
priority = "P0" if f.severity == "critical" else "P1"
lines = [
f"**Severity:** {f.severity} (`{priority}`)",
f"**Cluster:** {f.cluster}",
f"**Finding ID:** {f.finding_id}",
"",
"## Principle",
f"{f.principle} ({f.principle_reference})",
"",
"## Evidence",
f"- File: `{f.evidence.file}` lines {f.evidence.line_range[0]}-{f.evidence.line_range[1]}",
f"- Metric: {f.evidence.metric}",
f"- Code refs: {', '.join(f.evidence.code_refs)}",
"",
"## Failure scenario",
f.failure_scenario,
"",
"## Recommended action",
f.recommended_action,
"",
"## Rationale",
f.rationale,
]
if f.drift_signals:
lines += ["", "## Drift signals"] + [f"- {s}" for s in f.drift_signals]
return "\n".join(lines)
def _umbrella_body(cluster: str, overflow_count: int) -> str:
"""Build markdown body for an umbrella issue bundling overflow findings."""
lines = [
f"## Umbrella issue: {cluster}",
"",
f"This cluster has **{overflow_count} additional findings** beyond the "
f"5-per-cluster cap. Each finding is summarized below; the operator "
f"should file individual issues or address them in a batch PR.",
]
return "\n".join(lines)
def build_issue_payloads(
findings: List[Finding],
*,
cap: int = 5,
) -> List[IssuePayload]:
"""Build GitHub issue payloads from audit findings.
Groups HIGH+CRITICAL findings by cluster, emits up to ``cap`` individual
issues per cluster (sorted critical-first then by file), and one umbrella
issue per cluster for any overflow.
"""
# Filter to auditable severities only.
auditable = [f for f in findings if f.severity in _AUDITABLE_SEVERITIES]
if not auditable:
return []
# Group by cluster.
clusters: dict[str, list[Finding]] = {}
for f in auditable:
clusters.setdefault(f.cluster, []).append(f)
result: list[IssuePayload] = []
for cluster_name, cluster_findings in sorted(clusters.items()):
# Rank by severity (critical first), then by file for determinism.
cluster_findings.sort(
key=lambda f: (
0 if f.severity == "critical" else 1,
f.evidence.file,
f.evidence.line_range[0],
)
)
kept = cluster_findings[:cap]
overflow = cluster_findings[cap:]
for f in kept:
priority_label = "P0" if f.severity == "critical" else "P1"
result.append(
IssuePayload(
title=_title_for(f),
body=_body_for(f),
labels=BASE_LABELS + [priority_label],
)
)
if overflow:
short = _CLUSTER_SHORT.get(
cluster_name,
cluster_name.lower().replace(" ", "-")[:20],
)
result.append(
IssuePayload(
title=(
f"[audit:spec-1:{short}] {len(overflow)} additional findings "
f"from harness self-audit"
),
body=_umbrella_body(cluster_name, len(overflow)),
labels=BASE_LABELS + ["P1"],
)
)
return result
-306
View File
@@ -1,306 +0,0 @@
"""Cluster lane prompt templates for the harness self-audit.
Five constants — one per cluster (A-E) per the spec's methodology table.
Each template is the prompt an Explore-agent receives; it embeds the
per-finding schema so the agent returns valid cards. Placeholders:
{repo_root} — absolute path to the repo under audit
{commit_sha} — git HEAD commit SHA at audit time (snapshot per spec)
Use `render_prompt(letter, repo_root, commit_sha)` to get a ready-to-paste
prompt, or run the CLI:
python scripts/audit/prompts.py <A|B|C|D|E> --repo-root PATH --commit-sha SHA
"""
from __future__ import annotations
import argparse
import sys
from dataclasses import dataclass
from pathlib import Path
from typing import Iterable
@dataclass(frozen=True)
class ClusterPrompt:
letter: str
title: str
principles: tuple[str, ...]
evidence_strategy: str
template: str
# ---------- Cluster A: Readability & quality bar ----------
_A_TEMPLATE = """\
You are auditing the heretek-claude-harness repository for Cluster A:
**Readability & quality bar**.
Repo: {repo_root}
Snapshot commit: {commit_sha}
**Principles in scope**
- Small, focused functions
- Clear, consistent naming
- Comments explain WHY, not WHAT
- DRY, KISS, YAGNI
- No dead code, no copy-paste duplication
**Evidence-collection strategy**
- File size, function size, cyclomatic + cognitive complexity
- Lint output (ruff, pylint if present)
- Line counts, duplication scan
- For each candidate finding, cite a numeric metric and the tool that produced it.
**Posture: ADVERSARIAL.** Assume each principle is being violated until evidence
proves otherwise. Prefer false positives over missed findings; synthesis will
re-verify.
**Output format** — one YAML document per finding (or one YAML list of findings).
Return ONLY findings; do not write prose.
```yaml
- finding_id: A-NNN # sequential, zero-padded
cluster: "Readability & quality bar"
principle: "<one-line principle statement>"
severity: critical|high|medium|low|info
adversarial_posture: violated|partial|justified
evidence:
code_refs: ["<path>:<start_line>-<end_line>"]
file: "<path>"
line_range: [<start>, <end>]
metric: "<numeric metric + tool that produced it>"
failure_scenario: "<concrete inputs/state -> wrong output/crash>"
recommended_action: refactor|document|suppress|accept|escalate
rationale: "<one sentence: the why, not the what>"
principle_reference: "Code quality > <subsection> > <principle>"
drift_signals: [] # any "this was already flagged in PR #N" notes
```
If a principle is fully met across the in-scope code, return an empty list. Do
NOT fabricate findings to fill quotas.
"""
# ---------- Cluster B: Design & architecture ----------
_B_TEMPLATE = """\
You are auditing the heretek-claude-harness repository for Cluster B:
**Design & architecture**.
Repo: {repo_root}
Snapshot commit: {commit_sha}
**Principles in scope**
- SOLID (especially SRP, OCP, DIP)
- Composition over inheritance
- Tell, don't ask
- Law of Demeter
- Separation of concerns; loose coupling, high cohesion
**Evidence-collection strategy**
- Import graph (look for cycles, deep inheritance)
- Class/method counts per file (god-file detection: > 500 LOC + many responsibilities)
- Base-class fanout (a base used by > 5 concrete classes may be over-reaching)
- Module boundaries (does `scripts/scanners/` import from `tests/`? vice versa?)
**Posture: ADVERSARIAL.** Assume each principle is being violated until evidence
proves otherwise.
**Output format** — same per-finding YAML card as Cluster A. Use finding_id
prefix `B-NNN`. Cluster field: `"Design & architecture"`.
"""
# ---------- Cluster C: Correctness & safety ----------
_C_TEMPLATE = """\
You are auditing the heretek-claude-harness repository for Cluster C:
**Correctness & safety**.
Repo: {repo_root}
Snapshot commit: {commit_sha}
**Principles in scope**
- Explicit error handling (no silent exception swallowing)
- Input validation at boundaries
- Type hints on public functions
- Idempotent operations where possible
- Immutability over mutation when reasonable
- Defensive programming only where genuine threat exists
**Evidence-collection strategy**
- try/except scan: bare `except:` or `except Exception: pass` patterns
- Type-hint coverage scan on public functions (rough: missing `->` on > 30% of
public callables in a module is a cluster-C finding)
- Mutability patterns: top-level mutable defaults, module-level dict/list
mutations
- Parameter validation markers: functions that consume `Path` / `int` /
untyped `dict` from external sources without validation
**Posture: ADVERSARIAL.** Assume each principle is being violated until evidence
proves otherwise.
**Output format** — same per-finding YAML card. Use finding_id prefix `C-NNN`.
Cluster field: `"Correctness & safety"`.
"""
# ---------- Cluster D: Testing & verification ----------
_D_TEMPLATE = """\
You are auditing the heretek-claude-harness repository for Cluster D:
**Testing & verification**.
Repo: {repo_root}
Snapshot commit: {commit_sha}
**Principles in scope**
- Test pyramid (many unit, fewer integration, few E2E)
- Test isolation (no shared mutable state between tests)
- Coverage of critical paths (not 100% for its own sake)
- TDD evidence in commit history (red-green-refactor visible in `git log -- <file>`)
- No flaky-test markers (sleeps, time-dependent assertions, network mocks)
- No test smells (asserts on private attrs, oversized fixtures)
**Evidence-collection strategy**
- pytest markers: are `integration` tests actually marked? (run
`grep -rE "@pytest.mark.integration" tests/`)
- conftest.py review: shared fixtures, autouse patterns, module-level state
- Test counts per source module (rough: < 1 test per 50 LOC of source = smell)
- Fixture audit: oversized fixtures used by > 20 tests = god-fixture
- Commit history: spot-check 3 source files for TDD-shaped commit pairs
**Posture: ADVERSARIAL.** Assume each principle is being violated until evidence
proves otherwise.
**Output format** — same per-finding YAML card. Use finding_id prefix `D-NNN`.
Cluster field: `"Testing & verification"`.
"""
# ---------- Cluster E: Operations & docs ----------
_E_TEMPLATE = """\
You are auditing the heretek-claude-harness repository for Cluster E:
**Operations & docs**.
Repo: {repo_root}
Snapshot commit: {commit_sha}
**Principles in scope**
- Observability: structured logs, no bare `print()` in library code
- CI/CD maturity: workflows run on every PR, fail loud
- Feature flags / rollback: changes have a way back
- Security hygiene: pinned deps, no shell-injection sinks, secret-handling
- Doc freshness: README matches reality, ADRs for non-trivial decisions
- Onboarding: a new contributor can be productive in < 1 day
**Evidence-collection strategy**
- log/print scan: `grep -rnE "^\\s*print\\(" scripts/` (library code should
use the `logging` module)
- Requirements freshness: `requirements.txt` pins, `requirements.lock.txt`
presence, last-updated date
- GH workflow audit (READABILITY only — does the YAML parse? does each step
have a name? timeout set? — NOT enforcement behavior, that's Spec 3)
- Docstring coverage on public functions
- README freshness (does install command work? do plugin counts match?)
- Look for missing ADRs on non-trivial decisions
**Posture: ADVERSARIAL.** Assume each principle is being violated until evidence
proves otherwise.
**Output format** — same per-finding YAML card. Use finding_id prefix `E-NNN`.
Cluster field: `"Operations & docs"`.
"""
CLUSTERS: dict[str, ClusterPrompt] = {
"A": ClusterPrompt(
letter="A",
title="Readability & quality bar",
principles=(
"Small, focused functions",
"Clear, consistent naming",
"Comments explain WHY, not WHAT",
"DRY, KISS, YAGNI",
"No dead code, no copy-paste duplication",
),
evidence_strategy="file/function size, complexity metrics, lint, duplication scan",
template=_A_TEMPLATE,
),
"B": ClusterPrompt(
letter="B",
title="Design & architecture",
principles=(
"SOLID (SRP, OCP, DIP)",
"Composition over inheritance",
"Tell, don't ask",
"Law of Demeter",
"Separation of concerns; loose coupling, high cohesion",
),
evidence_strategy="import graph, god-file detection, base-class fanout, module boundaries",
template=_B_TEMPLATE,
),
"C": ClusterPrompt(
letter="C",
title="Correctness & safety",
principles=(
"Explicit error handling",
"Input validation at boundaries",
"Type hints on public functions",
"Idempotent operations",
"Immutability over mutation",
"Defensive programming where threat exists",
),
evidence_strategy="try/except scan, type-hint coverage, mutability patterns, validation markers",
template=_C_TEMPLATE,
),
"D": ClusterPrompt(
letter="D",
title="Testing & verification",
principles=(
"Test pyramid",
"Test isolation",
"Coverage of critical paths",
"TDD evidence in commit history",
"No flaky-test markers",
"No test smells",
),
evidence_strategy="pytest markers, conftest review, test counts per module, fixture audit, commit history",
template=_D_TEMPLATE,
),
"E": ClusterPrompt(
letter="E",
title="Operations & docs",
principles=(
"Observability",
"CI/CD maturity",
"Feature flags / rollback",
"Security hygiene",
"Doc freshness",
"Onboarding",
),
evidence_strategy="log/print scan, requirements freshness, GH workflow audit, docstring coverage, README freshness, ADR presence",
template=_E_TEMPLATE,
),
}
def render_prompt(letter: str, repo_root: Path, commit_sha: str) -> str:
"""Substitute placeholders and return the ready-to-paste prompt."""
cp = CLUSTERS[letter] # raises KeyError for unknown letters
return cp.template.replace("{repo_root}", str(repo_root)).replace("{commit_sha}", commit_sha)
def main(argv: Iterable[str] | None = None) -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument(
"cluster",
choices=sorted(CLUSTERS.keys()),
help="Cluster letter (A, B, C, D, or E).",
)
parser.add_argument("--repo-root", type=Path, required=True)
parser.add_argument("--commit-sha", required=True, help="Audit snapshot commit SHA.")
args = parser.parse_args(argv)
print(render_prompt(args.cluster, args.repo_root, args.commit_sha))
return 0
if __name__ == "__main__":
sys.exit(main())
-218
View File
@@ -1,218 +0,0 @@
"""Synthesis pass: load cluster results, dedupe, rank, write report + JSON.
Loads all 5 cluster result files (A-E) via audit.findings.load_findings
(schema-validated), filters SonarCloud exclusions, dedupes by
(file, line_range) keeping the higher-severity entry, sorts by severity,
and writes both a markdown report and JSON findings to
output_dir/audit-harness-self-<date>.{md,json}. Coverage gaps section
flags missing clusters and zero-finding clusters.
"""
from __future__ import annotations
from dataclasses import dataclass
from datetime import date
from pathlib import Path
from audit import findings
CLUSTER_LETTERS = ["A", "B", "C", "D", "E"]
@dataclass
class SynthesisResult:
findings: list[findings.Finding]
report_path: Path
json_path: Path
coverage_gaps: list[str]
duplicate_count: int
def _load_cluster_results(cluster_results_dir: Path) -> list[findings.Finding]:
"""Load all YAML/JSON cluster result files from *cluster_results_dir*.
Files are named <cluster_letter>.{yaml,yml,json}. Each file is
loaded and validated via findings.load_findings.
"""
all_findings: list[findings.Finding] = []
for letter in CLUSTER_LETTERS:
for ext in (".yaml", ".yml", ".json"):
path = cluster_results_dir / f"{letter}{ext}"
if path.exists():
all_findings.extend(findings.load_findings(path))
break
return all_findings
def _dedupe(items: list[findings.Finding]) -> tuple[list[findings.Finding], int]:
"""Collapse findings with identical (file, line_range). Keep the higher-severity one."""
seen: dict[tuple[str, tuple[int, int]], findings.Finding] = {}
dup_count = 0
for f in items:
key = (f.evidence.file, tuple(f.evidence.line_range))
if key in seen:
dup_count += 1
# Keep the more severe one (lower rank = more severe).
if findings.severity_rank(f) < findings.severity_rank(seen[key]):
seen[key] = f
else:
seen[key] = f
deduped = sorted(seen.values(), key=findings.severity_rank)
return deduped, dup_count
def _ranked_table(items: list[findings.Finding]) -> str:
"""Return a markdown table of findings sorted by severity."""
lines = [
"| # | ID | Severity | File | Line Range | Principle |",
"|---|---|---|---|---|---|",
]
for i, f in enumerate(items, 1):
lr = f.evidence.line_range
lines.append(
f"| {i} | {f.finding_id} | {f.severity} | "
f"`{f.evidence.file}` | {lr[0]}-{lr[1]} | {f.principle} |"
)
return "\n".join(lines)
def _present_cluster_files(cluster_results_dir: Path) -> set[str]:
"""Detect which cluster letters have a result file present."""
present: set[str] = set()
for letter in CLUSTER_LETTERS:
for ext in (".yaml", ".yml", ".json"):
if (cluster_results_dir / f"{letter}{ext}").exists():
present.add(letter)
break
return present
def _present_cluster_findings(items: list[findings.Finding]) -> set[str]:
"""Detect which cluster letters have at least one finding (by finding_id prefix)."""
present: set[str] = set()
for f in items:
for letter in CLUSTER_LETTERS:
if f.finding_id.startswith(letter):
present.add(letter)
break
return present
def _coverage_gaps_section(cluster_results_dir: Path, items: list[findings.Finding]) -> list[str]:
"""Detect missing clusters and zero-finding clusters."""
present_files = _present_cluster_files(cluster_results_dir)
present_findings = _present_cluster_findings(items)
gaps: list[str] = []
for letter in CLUSTER_LETTERS:
if letter not in present_files:
gaps.append(f"Cluster {letter}: result file missing")
elif letter not in present_findings:
gaps.append(f"Cluster {letter}: zero findings")
return gaps
def _report_markdown(
items: list[findings.Finding],
commit_sha: str,
duplicate_count: int,
coverage_gaps: list[str],
) -> str:
"""Build the full markdown report."""
crit = sum(1 for f in items if f.severity == "critical")
high = sum(1 for f in items if f.severity == "high")
med = sum(1 for f in items if f.severity == "medium")
low = sum(1 for f in items if f.severity == "low")
info = sum(1 for f in items if f.severity == "info")
sections = [
"# Harness Self-Audit Report",
"",
f"**Snapshot:** `{commit_sha}`",
f"**Date:** {date.today().isoformat()}",
f"**Total findings (after dedup):** {len(items)}",
f"**Duplicates removed:** {duplicate_count}",
"",
"## Severity Summary",
"",
f"- Critical: {crit}",
f"- High: {high}",
f"- Medium: {med}",
f"- Low: {low}",
f"- Info: {info}",
"",
]
if items:
sections.append("## Findings")
sections.append("")
sections.append(_ranked_table(items))
sections.append("")
if coverage_gaps:
sections.append("## Coverage Gaps")
sections.append("")
for gap in coverage_gaps:
sections.append(f"- {gap}")
sections.append("")
sections.extend(
[
"## Re-verification Checklist",
"",
"Synthesis flagged findings that need operator review. "
"The actual metric re-check is executed by the operator.",
"",
"- [ ] Every `critical` finding inspected directly at the cited file:line",
"- [ ] Every `high` finding inspected unless numeric metric + tool output present",
"- [ ] Severity counts reviewed and adjusted if needed",
"",
]
)
return "\n".join(sections)
def synthesize(
cluster_results_dir: Path,
output_dir: Path,
commit_sha: str,
sonar_exclusions: set[str] | None = None,
) -> SynthesisResult:
"""Run the full synthesis pass.
Loads cluster results, filters SonarCloud exclusions, dedupes,
sorts by severity, writes report + JSON.
"""
output_dir.mkdir(parents=True, exist_ok=True)
all_findings = _load_cluster_results(cluster_results_dir)
# Filter SonarCloud exclusions
if sonar_exclusions:
all_findings = [f for f in all_findings if f.evidence.file not in sonar_exclusions]
deduped, dup_count = _dedupe(all_findings)
# Determine output paths
today = date.today().isoformat()
report_path = output_dir / f"audit-harness-self-{today}.md"
json_path = output_dir / f"audit-harness-self-{today}.json"
# Write JSON
findings.save_findings(deduped, json_path)
# Coverage gaps
coverage_gaps = _coverage_gaps_section(cluster_results_dir, all_findings)
# Write report
report = _report_markdown(deduped, commit_sha, dup_count, coverage_gaps)
report_path.write_text(report)
return SynthesisResult(
findings=deduped,
report_path=report_path,
json_path=json_path,
coverage_gaps=coverage_gaps,
duplicate_count=dup_count,
)
-89
View File
@@ -1,89 +0,0 @@
"""Validate audit finding cards against the finding JSON Schema.
Run as CLI:
python scripts/audit/validate.py [--schema PATH] <file1.yaml|json> [file2 ...]
Exit codes: 0 if every card validates, 1 if any fails. Errors are printed
to stderr; per-file summary goes to stdout.
Library API:
validate_finding(finding: dict) -> list[str]
validate_findings(findings: list[dict]) -> list[tuple[int, list[str]]]
"""
from __future__ import annotations
import argparse
import json
import sys
from pathlib import Path
from typing import Iterable
import jsonschema
from ruamel.yaml import YAML
REPO_ROOT = Path(__file__).resolve().parents[2]
DEFAULT_SCHEMA = REPO_ROOT / "tests" / "schemas" / "audit_finding.schema.json"
SCHEMA_PATH: Path = DEFAULT_SCHEMA # tests may monkeypatch
def _load_instance(path: Path) -> dict:
text = path.read_text()
if path.suffix.lower() in {".yaml", ".yml"}:
result = YAML(typ="safe").load(text)
if result is None:
raise ValueError(f"empty or whitespace-only YAML in {path}")
return result
return json.loads(text)
def validate_finding(finding: dict) -> list[str]:
"""Return a list of error messages for one finding; empty = valid."""
schema = json.loads(SCHEMA_PATH.read_text())
validator = jsonschema.Draft202012Validator(schema)
return [f"{e.json_path}: {e.message}" for e in validator.iter_errors(finding)]
def validate_findings(findings: list[dict]) -> list[tuple[int, list[str]]]:
"""Validate a list; return [(index, errors)] for every invalid entry."""
return [(i, errs) for i, f in enumerate(findings) for errs in [validate_finding(f)] if errs]
def main(argv: Iterable[str] | None = None) -> int:
global SCHEMA_PATH # noqa: PLW0603
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--schema", type=Path, default=SCHEMA_PATH)
parser.add_argument("files", nargs="+", type=Path)
args = parser.parse_args(argv)
SCHEMA_PATH = args.schema.resolve()
any_errors = False
for path in args.files:
file_has_errors = False
resolved = path.resolve()
try:
instance = _load_instance(resolved)
except ValueError as exc:
print(f"{path}: parse error: {exc}", file=sys.stderr)
any_errors = True
file_has_errors = True
continue
# A file may contain a single finding OR a list of findings.
findings = instance if isinstance(instance, list) else [instance]
for idx, errors in validate_findings(findings):
file_has_errors = True
print(f"{path}#{idx}:", file=sys.stderr)
for e in errors:
print(f" - {e}", file=sys.stderr)
if not file_has_errors:
print(f"{path}: OK ({len(findings)} finding(s))")
else:
any_errors = True
return 1 if any_errors else 0
if __name__ == "__main__":
sys.exit(main())
-90
View File
@@ -1,90 +0,0 @@
"""End-to-end test: fixture cluster results -> synthesize -> build-issues -> payload sanity."""
import json
import sys
from pathlib import Path
import pytest
sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "scripts"))
from audit import findings, harness_self, issues, synthesis, validate # noqa: E402
@pytest.fixture(autouse=True)
def _set_schema(schemas_dir: Path) -> None:
validate.SCHEMA_PATH = schemas_dir / "audit_finding.schema.json"
FIXTURES = Path(__file__).resolve().parent / "fixtures" / "audit" / "cluster_results"
def test_e2e_synthesize_then_build_issues(tmp_path: Path) -> None:
"""Run the full pipeline against fixture cluster results."""
# Step 1: synthesize
result = synthesis.synthesize(
cluster_results_dir=FIXTURES,
output_dir=tmp_path / "out",
commit_sha="e2etest0",
)
assert len(result.findings) == 3
# Verify the report mentions the snapshot SHA.
md = result.report_path.read_text()
assert "e2etest0" in md
assert "Critical: 1" in md
assert "High: 1" in md
assert "Medium: 1" in md
# Step 2: build-issues (via the synthesis JSON)
fs = [findings.Finding.from_dict(d) for d in json.loads(result.json_path.read_text())]
payloads = issues.build_issue_payloads(fs)
# 1 critical + 1 high -> 2 issues; medium excluded.
assert len(payloads) == 2
titles = {p.title for p in payloads}
assert any("Small, focused functions" in t for t in titles)
assert any("Single Responsibility" in t for t in titles)
def test_e2e_run_all_via_driver(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
"""Call the driver's synthesize subcommand end-to-end via the CLI."""
out_dir = tmp_path / "out"
monkeypatch.setattr(
sys,
"argv",
[
"harness_self.py",
"synthesize",
"--cluster-results",
str(FIXTURES),
"--output-dir",
str(out_dir),
"--commit-sha",
"e2etest1",
],
)
assert harness_self.main() == 0
assert any(out_dir.glob("audit-harness-self-*.md"))
assert any(out_dir.glob("audit-harness-self-*.json"))
def test_e2e_emit_prompts_then_synthesize(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
"""emit-prompts + synthesize via the driver."""
prompts_dir = tmp_path / "prompts"
monkeypatch.setattr(
sys,
"argv",
[
"harness_self.py",
"emit-prompts",
"--repo-root",
str(tmp_path),
"--commit-sha",
"e2etest2",
"--out-dir",
str(prompts_dir),
],
)
assert harness_self.main() == 0
for letter in ("A", "B", "C", "D", "E"):
assert (prompts_dir / f"{letter}.txt").is_file()
assert "Readability" in (prompts_dir / "A.txt").read_text()
-79
View File
@@ -1,79 +0,0 @@
"""Tests for scripts/audit/findings.py."""
import json
import sys
from pathlib import Path
import pytest
sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "scripts"))
from audit import findings, validate # noqa: E402
@pytest.fixture(autouse=True)
def _set_schema(schemas_dir: Path) -> None:
validate.SCHEMA_PATH = schemas_dir / "audit_finding.schema.json"
def _sample_dict() -> dict:
return {
"finding_id": "A-001",
"cluster": "Readability & quality bar",
"principle": "Small, focused functions",
"severity": "high",
"adversarial_posture": "violated",
"evidence": {
"code_refs": ["scripts/x.py:1-10"],
"file": "scripts/x.py",
"line_range": [1, 10],
"metric": "x",
},
"failure_scenario": "x",
"recommended_action": "refactor",
"rationale": "x",
"principle_reference": "x",
"drift_signals": [],
}
def test_finding_from_dict_round_trips() -> None:
d = _sample_dict()
f = findings.Finding.from_dict(d)
assert f.finding_id == "A-001"
assert f.severity == "high"
assert f.evidence.line_range == [1, 10]
assert f.to_dict() == d
def test_load_findings_accepts_yaml(fixtures_dir: Path) -> None:
p = fixtures_dir / "audit" / "valid_finding.yaml"
loaded = findings.load_findings(p)
assert len(loaded) == 1
assert loaded[0].finding_id == "A-001"
def test_load_findings_rejects_invalid(fixtures_dir: Path) -> None:
p = fixtures_dir / "audit" / "invalid_finding_bad_enum.yaml"
with pytest.raises(ValueError, match="invalid"):
findings.load_findings(p)
def test_save_findings_writes_json(tmp_path: Path) -> None:
f = findings.Finding.from_dict(_sample_dict())
out = tmp_path / "out.json"
findings.save_findings([f], out)
assert out.is_file()
data = json.loads(out.read_text())
assert data[0]["finding_id"] == "A-001"
def test_severity_rank_orders_critical_first() -> None:
f_crit = findings.Finding.from_dict({**_sample_dict(), "severity": "critical"})
f_high = findings.Finding.from_dict({**_sample_dict(), "severity": "high"})
f_info = findings.Finding.from_dict({**_sample_dict(), "severity": "info"})
assert (
findings.severity_rank(f_crit)
< findings.severity_rank(f_high)
< findings.severity_rank(f_info)
)
-182
View File
@@ -1,182 +0,0 @@
"""Tests for scripts/audit/harness_self.py."""
import json
import sys
from pathlib import Path
import pytest
sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "scripts"))
from audit import harness_self, validate # noqa: E402
@pytest.fixture(autouse=True)
def _set_schema(schemas_dir: Path) -> None:
validate.SCHEMA_PATH = schemas_dir / "audit_finding.schema.json"
# -- emit-prompts ----------------------------------------------------------
def test_emit_prompts_writes_five_files(tmp_path: Path, repo_root: Path) -> None:
"""emit-prompts creates exactly one .txt per cluster letter A-E."""
out_dir = tmp_path / "prompts"
rc = harness_self.main(
[
"emit-prompts",
"--repo-root",
str(repo_root),
"--commit-sha",
"abcd1234",
"--out-dir",
str(out_dir),
]
)
assert rc == 0
letters = sorted(f.stem for f in out_dir.iterdir())
assert letters == ["A", "B", "C", "D", "E"]
for f in out_dir.iterdir():
assert f.suffix == ".txt"
def test_emit_prompts_substitutes_placeholders(
tmp_path: Path,
repo_root: Path,
) -> None:
"""Rendered prompt contains the literal repo_root and commit_sha."""
out_dir = tmp_path / "prompts"
harness_self.main(
[
"emit-prompts",
"--repo-root",
str(repo_root),
"--commit-sha",
"deadbeef",
"--out-dir",
str(out_dir),
]
)
content = (out_dir / "A.txt").read_text()
assert str(repo_root) in content
assert "deadbeef" in content
# -- synthesize ------------------------------------------------------------
def test_synthesize_writes_report_and_json(
tmp_path: Path,
repo_root: Path,
) -> None:
"""synthesize delegates to audit.synthesis and produces report + JSON."""
# Put one minimal finding in cluster A
cluster_dir = tmp_path / "clusters"
cluster_dir.mkdir()
(cluster_dir / "A.json").write_text(
json.dumps(
[
{
"finding_id": "A-001",
"cluster": "Readability & quality bar",
"principle": "Small, focused functions",
"severity": "high",
"adversarial_posture": "violated",
"evidence": {
"code_refs": ["scripts/x.py:1-10"],
"file": "scripts/x.py",
"line_range": [1, 10],
"metric": "x",
},
"failure_scenario": "x",
"recommended_action": "refactor",
"rationale": "x",
"principle_reference": "x",
"drift_signals": [],
}
]
)
+ "\n"
)
out_dir = tmp_path / "synth-out"
rc = harness_self.main(
[
"synthesize",
"--cluster-results",
str(cluster_dir),
"--output-dir",
str(out_dir),
"--commit-sha",
"cafe0000",
]
)
assert rc == 0
# Should produce a report .md and a findings .json
md_files = list(out_dir.glob("*.md"))
json_files = list(out_dir.glob("*.json"))
assert md_files, "no .md report written"
assert json_files, "no .json findings written"
content = json_files[0].read_text()
assert "A-001" in content
# -- build-issues ----------------------------------------------------------
def test_build_issues_writes_payload_file(tmp_path: Path) -> None:
"""build-issues reads findings JSON and writes IssuePayload list."""
findings_path = tmp_path / "findings.json"
findings_path.write_text(
json.dumps(
[
{
"finding_id": "A-001",
"cluster": "Readability & quality bar",
"principle": "Small, focused functions",
"severity": "critical",
"adversarial_posture": "violated",
"evidence": {
"code_refs": ["scripts/x.py:1-10"],
"file": "scripts/x.py",
"line_range": [1, 10],
"metric": "x",
},
"failure_scenario": "x",
"recommended_action": "refactor",
"rationale": "x",
"principle_reference": "x",
"drift_signals": [],
}
]
)
+ "\n"
)
out_path = tmp_path / "issues-to-create.json"
rc = harness_self.main(
[
"build-issues",
"--findings-json",
str(findings_path),
"--output",
str(out_path),
]
)
assert rc == 0
payload = json.loads(out_path.read_text())
assert isinstance(payload, list)
assert len(payload) >= 1
assert "title" in payload[0]
assert "body" in payload[0]
assert "labels" in payload[0]
# -- argparse rejection ----------------------------------------------------
def test_unknown_subcommand_exits_nonzero(monkeypatch: pytest.MonkeyPatch) -> None:
"""An unknown subcommand should exit with a non-zero status (argparse rejects)."""
monkeypatch.setattr(sys, "argv", ["harness_self.py", "bogus-cmd"])
with pytest.raises(SystemExit) as exc_info:
harness_self.main()
# argparse with required=True subparsers calls sys.exit(2) for unknown subcommand
assert exc_info.value.code != 0
-118
View File
@@ -1,118 +0,0 @@
"""Tests for scripts/audit/issues.py."""
import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "scripts"))
from audit.findings import Evidence, Finding # noqa: E402
from audit.issues import ( # noqa: E402
BASE_LABELS,
build_issue_payloads,
)
def _f(
severity: str,
fid: str,
cluster: str,
principle: str = "test principle",
) -> Finding:
"""Minimal Finding helper for tests."""
return Finding(
finding_id=fid,
cluster=cluster,
principle=principle,
severity=severity,
adversarial_posture="violated",
evidence=Evidence(
code_refs=["scripts/x.py:1-10"],
file="scripts/x.py",
line_range=[1, 10],
metric="x",
),
failure_scenario="x",
recommended_action="refactor",
rationale="x",
principle_reference="x",
)
# --- 7 tests per brief ---
def test_skips_medium_low_info() -> None:
"""medium, low, info findings produce zero payloads."""
fs = [
_f("medium", "A-001", "Readability & quality bar"),
_f("low", "A-002", "Readability & quality bar"),
_f("info", "A-003", "Readability & quality bar"),
]
assert build_issue_payloads(fs) == []
def test_emits_one_payload_per_critical_high() -> None:
"""Each critical and high finding gets its own IssuePayload."""
fs = [
_f("critical", "A-001", "Readability & quality bar"),
_f("high", "A-002", "Readability & quality bar"),
]
payloads = build_issue_payloads(fs)
assert len(payloads) == 2
ids = {p.labels[-1] for p in payloads}
assert "P0" in ids
assert "P1" in ids
def test_caps_at_five_per_cluster() -> None:
"""8 critical findings in cluster A yields 5 individual + 1 umbrella with overflow=3."""
fs = [_f("critical", f"A-{i:03d}", "Readability & quality bar") for i in range(1, 9)]
payloads = build_issue_payloads(fs)
# 5 individual + 1 umbrella = 6
assert len(payloads) == 6
# Last one is the umbrella
umbrella = payloads[-1]
assert "3 additional findings" in umbrella.title
def test_umbrella_title_shows_overflow_count() -> None:
"""Umbrella issue title includes the correct overflow count."""
fs = [_f("critical", f"A-{i:03d}", "Readability & quality bar") for i in range(1, 11)]
payloads = build_issue_payloads(fs)
umbrella = payloads[-1]
assert "5 additional findings" in umbrella.title
def test_per_cluster_cap_not_global() -> None:
"""Cap is per-cluster: 5 in A + 5 in B each produce 5 issues, no umbrella."""
fs_a = [_f("critical", f"A-{i:03d}", "Readability & quality bar") for i in range(1, 6)]
fs_b = [_f("critical", f"B-{i:03d}", "Design & architecture") for i in range(1, 6)]
payloads = build_issue_payloads(fs_a + fs_b)
# Exactly 10 — no umbrella because neither cluster exceeds cap
assert len(payloads) == 10
assert not any("additional findings" in p.title for p in payloads)
def test_labels_include_p0_for_critical() -> None:
"""Critical finding gets P0 priority label."""
fs = [_f("critical", "A-001", "Readability & quality bar")]
payloads = build_issue_payloads(fs)
assert len(payloads) == 1
p = payloads[0]
assert "P0" in p.labels
assert "P1" not in p.labels
for lbl in BASE_LABELS:
assert lbl in p.labels
def test_labels_include_p1_for_high() -> None:
"""High finding gets P1 priority label."""
fs = [_f("high", "A-001", "Readability & quality bar")]
payloads = build_issue_payloads(fs)
assert len(payloads) == 1
p = payloads[0]
assert "P1" in p.labels
assert "P0" not in p.labels
for lbl in BASE_LABELS:
assert lbl in p.labels
-52
View File
@@ -1,52 +0,0 @@
"""Tests for scripts/audit/prompts.py."""
import sys
from pathlib import Path
import pytest
sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "scripts"))
from audit import prompts # noqa: E402
def test_all_five_clusters_present() -> None:
assert set(prompts.CLUSTERS.keys()) == {"A", "B", "C", "D", "E"}
@pytest.mark.parametrize("letter", ["A", "B", "C", "D", "E"])
def test_each_cluster_has_required_fields(letter: str) -> None:
cp = prompts.CLUSTERS[letter]
assert cp.letter == letter
assert cp.title
assert len(cp.principles) >= 1
assert cp.evidence_strategy
assert cp.template
# Template must mention {repo_root} and {commit_sha}.
assert "{repo_root}" in cp.template
assert "{commit_sha}" in cp.template
def test_render_prompt_substitutes_placeholders(tmp_path: Path) -> None:
rendered = prompts.render_prompt("A", tmp_path, "deadbeef")
assert str(tmp_path) in rendered
assert "deadbeef" in rendered
def test_render_prompt_rejects_unknown_letter(tmp_path: Path) -> None:
with pytest.raises(KeyError):
prompts.render_prompt("Z", tmp_path, "deadbeef")
def test_cli_prints_prompt(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture
) -> None:
monkeypatch.setattr(
sys,
"argv",
["prompts.py", "A", "--repo-root", str(tmp_path), "--commit-sha", "deadbeef"],
)
assert prompts.main() == 0
out = capsys.readouterr().out
assert "deadbeef" in out
assert "{repo_root}" not in out # substituted
-42
View File
@@ -1,42 +0,0 @@
"""Tests for the audit_finding.schema.json JSON Schema."""
import json
from pathlib import Path
import jsonschema
import pytest
def test_schema_exists(schemas_dir: Path) -> None:
schema_path = schemas_dir / "audit_finding.schema.json"
assert schema_path.is_file(), f"missing schema at {schema_path}"
def test_valid_finding_passes(schemas_dir: Path, fixtures_dir: Path) -> None:
schema = json.loads((schemas_dir / "audit_finding.schema.json").read_text())
example = (fixtures_dir / "audit" / "valid_finding.yaml").read_text()
# YAML, not JSON — we use ruamel to parse.
from ruamel.yaml import YAML
parsed = YAML(typ="safe").load(example)
jsonschema.validate(instance=parsed, schema=schema) # must not raise
def test_missing_required_field_fails(schemas_dir: Path, fixtures_dir: Path) -> None:
schema = json.loads((schemas_dir / "audit_finding.schema.json").read_text())
bad = (fixtures_dir / "audit" / "invalid_finding_missing_severity.yaml").read_text()
from ruamel.yaml import YAML
parsed = YAML(typ="safe").load(bad)
with pytest.raises(jsonschema.ValidationError):
jsonschema.validate(instance=parsed, schema=schema)
def test_bad_enum_value_fails(schemas_dir: Path, fixtures_dir: Path) -> None:
schema = json.loads((schemas_dir / "audit_finding.schema.json").read_text())
bad = (fixtures_dir / "audit" / "invalid_finding_bad_enum.yaml").read_text()
from ruamel.yaml import YAML
parsed = YAML(typ="safe").load(bad)
with pytest.raises(jsonschema.ValidationError):
jsonschema.validate(instance=parsed, schema=schema)
-112
View File
@@ -1,112 +0,0 @@
"""Tests for scripts/audit/synthesis.py."""
import json
import sys
from pathlib import Path
import pytest
sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "scripts"))
from audit import synthesis, validate # noqa: E402
@pytest.fixture(autouse=True)
def _set_schema(schemas_dir: Path) -> None:
validate.SCHEMA_PATH = schemas_dir / "audit_finding.schema.json"
def _finding(
severity: str = "high",
fid: str = "A-001",
file: str = "scripts/x.py",
lines=(1, 10),
) -> dict:
return {
"finding_id": fid,
"cluster": "Readability & quality bar",
"principle": "Small, focused functions",
"severity": severity,
"adversarial_posture": "violated",
"evidence": {
"code_refs": [f"{file}:{lines[0]}-{lines[1]}"],
"file": file,
"line_range": list(lines),
"metric": "x",
},
"failure_scenario": "x",
"recommended_action": "refactor",
"rationale": "x",
"principle_reference": "x",
"drift_signals": [],
}
def test_synthesize_dedupes_across_clusters(tmp_path: Path, schemas_dir: Path) -> None:
"""Two findings with identical (file, line_range) collapse to one."""
cluster_dir = tmp_path / "in"
cluster_dir.mkdir()
(cluster_dir / "A.yaml").write_text(
"[" + json.dumps(_finding(fid="A-001", severity="high")) + "]"
)
(cluster_dir / "B.yaml").write_text(
"["
+ json.dumps(
{
**_finding(fid="B-001", severity="medium"),
"cluster": "Design & architecture",
}
)
+ "]"
)
result = synthesis.synthesize(
cluster_results_dir=cluster_dir,
output_dir=tmp_path / "out",
commit_sha="deadbeef",
)
assert result.duplicate_count == 1
assert len(result.findings) == 1
# Higher-severity finding wins.
assert result.findings[0].severity == "high"
def test_synthesize_writes_report_and_json(tmp_path: Path, schemas_dir: Path) -> None:
cluster_dir = tmp_path / "in"
cluster_dir.mkdir()
(cluster_dir / "A.yaml").write_text("[" + json.dumps(_finding()) + "]")
out_dir = tmp_path / "out"
result = synthesis.synthesize(cluster_dir, out_dir, commit_sha="deadbeef")
assert result.report_path.is_file()
assert result.json_path.is_file()
data = json.loads(result.json_path.read_text())
assert len(data) == 1
md = result.report_path.read_text()
assert "harness-self-audit" in md or "Harness Self-Audit" in md
assert "deadbeef" in md
def test_synthesize_skips_sonar_excluded(tmp_path: Path, schemas_dir: Path) -> None:
cluster_dir = tmp_path / "in"
cluster_dir.mkdir()
(cluster_dir / "A.yaml").write_text("[" + json.dumps(_finding(file="scripts/excl.py")) + "]")
result = synthesis.synthesize(
cluster_dir,
tmp_path / "out",
commit_sha="deadbeef",
sonar_exclusions={"scripts/excl.py"},
)
assert result.findings == []
def test_synthesize_records_coverage_gap_for_missing_cluster(
tmp_path: Path, schemas_dir: Path
) -> None:
cluster_dir = tmp_path / "in"
cluster_dir.mkdir()
# Only cluster A present; B, C, D, E missing.
(cluster_dir / "A.yaml").write_text("[" + json.dumps(_finding()) + "]")
result = synthesis.synthesize(cluster_dir, tmp_path / "out", commit_sha="deadbeef")
gap_clusters = " ".join(result.coverage_gaps)
for letter in ("B", "C", "D", "E"):
assert letter in gap_clusters
-133
View File
@@ -1,133 +0,0 @@
"""Tests for scripts/audit/validate.py."""
import json as json_mod
import sys
from pathlib import Path
import pytest
sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "scripts"))
from audit import validate # noqa: E402
def _load_yaml(p: Path) -> dict:
from ruamel.yaml import YAML
return YAML(typ="safe").load(p.read_text())
def _set_schema(monkeypatch: pytest.MonkeyPatch, schemas_dir: Path) -> None:
monkeypatch.setattr(validate, "SCHEMA_PATH", schemas_dir / "audit_finding.schema.json")
def test_validate_finding_accepts_valid_card(
monkeypatch: pytest.MonkeyPatch, schemas_dir: Path, fixtures_dir: Path
) -> None:
_set_schema(monkeypatch, schemas_dir)
valid = _load_yaml(fixtures_dir / "audit" / "valid_finding.yaml")
errors = validate.validate_finding(valid)
assert errors == [], f"expected no errors, got: {errors}"
def test_validate_finding_rejects_missing_severity(
monkeypatch: pytest.MonkeyPatch, schemas_dir: Path, fixtures_dir: Path
) -> None:
_set_schema(monkeypatch, schemas_dir)
bad = _load_yaml(fixtures_dir / "audit" / "invalid_finding_missing_severity.yaml")
errors = validate.validate_finding(bad)
assert len(errors) >= 1
assert any("severity" in e for e in errors)
def test_validate_finding_rejects_bad_enum(
monkeypatch: pytest.MonkeyPatch, schemas_dir: Path, fixtures_dir: Path
) -> None:
_set_schema(monkeypatch, schemas_dir)
bad = _load_yaml(fixtures_dir / "audit" / "invalid_finding_bad_enum.yaml")
errors = validate.validate_finding(bad)
assert len(errors) >= 1
def test_validate_findings_indexes_errors(
monkeypatch: pytest.MonkeyPatch, schemas_dir: Path, fixtures_dir: Path
) -> None:
_set_schema(monkeypatch, schemas_dir)
findings = [
_load_yaml(fixtures_dir / "audit" / "valid_finding.yaml"),
_load_yaml(fixtures_dir / "audit" / "invalid_finding_bad_enum.yaml"),
]
indexed = validate.validate_findings(findings)
assert indexed == [] or all(isinstance(t, tuple) for t in indexed)
# The bad one at index 1 should produce errors.
bad_index, bad_errors = next(iter([t for t in indexed if t[1]]))
assert bad_index == 1
assert len(bad_errors) >= 1
def test_cli_exits_zero_on_valid(
monkeypatch: pytest.MonkeyPatch,
schemas_dir: Path,
fixtures_dir: Path,
tmp_path: Path,
) -> None:
_set_schema(monkeypatch, schemas_dir)
# Write a single-finding YAML file the CLI can read.
single = tmp_path / "single.yaml"
single.write_text((fixtures_dir / "audit" / "valid_finding.yaml").read_text())
monkeypatch.setattr(
sys,
"argv",
[
"validate.py",
"--schema",
str(schemas_dir / "audit_finding.schema.json"),
str(single),
],
)
assert validate.main() == 0
def test_cli_accepts_json_input(
monkeypatch: pytest.MonkeyPatch,
schemas_dir: Path,
fixtures_dir: Path,
tmp_path: Path,
) -> None:
_set_schema(monkeypatch, schemas_dir)
# Write a valid finding as JSON (not YAML).
valid = _load_yaml(fixtures_dir / "audit" / "valid_finding.yaml")
single = tmp_path / "single.json"
single.write_text(json_mod.dumps(valid))
monkeypatch.setattr(
sys,
"argv",
[
"validate.py",
"--schema",
str(schemas_dir / "audit_finding.schema.json"),
str(single),
],
)
assert validate.main() == 0
def test_cli_exits_one_on_invalid(
monkeypatch: pytest.MonkeyPatch,
schemas_dir: Path,
fixtures_dir: Path,
tmp_path: Path,
) -> None:
_set_schema(monkeypatch, schemas_dir)
single = tmp_path / "bad.yaml"
single.write_text((fixtures_dir / "audit" / "invalid_finding_bad_enum.yaml").read_text())
monkeypatch.setattr(
sys,
"argv",
[
"validate.py",
"--schema",
str(schemas_dir / "audit_finding.schema.json"),
str(single),
],
)
assert validate.main() == 1