skills/alirezarezvani__claude-skills
1
0
Fork 0
mirror of https://github.com/alirezarezvani/claude-skills.git synced 2026-09-19 03:31:11 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat(tooling): JSON-output sample gate (G9) + --sample fixtures (#654)

Browse Source 
Implements issue #654 Option A (embedded-sample convention) plus the
verification harness the issue asked for:

- scripts/smoke_json_output.py — new advisory gate (G9) that discovers
  every tool whose --help advertises JSON output, runs <tool> --sample
  <json-flag>, and asserts the stdout parses as JSON. Tools advertising
  JSON without --sample are reported as 'uncovered' (a backlog, not a
  failure) so the gate can be adopted incrementally; --strict flips that
  to a hard failure once coverage is high. Wired into ci-quality-gate.yml
  alongside G8.
- Added --sample embedded fixtures to the 5 tools named in #654:
  error_budget_calculator, slo_review, blast_radius_calculator,
  audit_log_analyzer, api_linter. Their required args are now optional
  when --sample is passed; missing-arg behavior is unchanged otherwise.
- Fixed 4 tools the new gate surfaced (prompt_rater, coach_tip_classifier,
  cheat_code_filter, redaction_linter): their --sample path printed human
  text and ignored --json; it now honors the JSON flag.
- Synced the 3 dual-published standalone copies (slo-architect x2,
  chaos-engineering) so the drift guard stays green.

Gate now reports 16 tools covered, 16 verified, 0 failures.

https://claude.ai/code/session_01CUWsrUNZP9jpxvAwq67UiT
This commit is contained in:
Claude
2026-06-11 15:29:39 +00:00
parent 5c395451a3
commit 028dc13b35
14 changed files with 473 additions and 78 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.github/workflows/ci-quality-gate.yml
+5
View File
@@ -113,6 +113,11 @@ jobs:
run: |
python3 scripts/smoke_scripts.py
- name: JSON-output sample gate (gate G9 — advisory)
continue-on-error: true
run: |
python3 scripts/smoke_json_output.py
- name: Counter derivation check (gate G3 — advisory)
continue-on-error: true
run: |
engineering/chaos-engineering/skills/chaos-engineering/scripts/blast_radius_calculator.py
+13 -4
View File
@@ -70,9 +70,11 @@ def render_text(result):
def main():
ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
ap.add_argument("--traffic-share", type=float, required=True, help="Fraction (0-1) of traffic affected")
ap.add_argument("--user-pop", type=int, required=True, help="Total user population")
ap.add_argument("--duration-min", type=int, required=True, help="Experiment duration in minutes")
ap.add_argument("--traffic-share", type=float, help="Fraction (0-1) of traffic affected")
ap.add_argument("--user-pop", type=int, help="Total user population")
ap.add_argument("--duration-min", type=int, help="Experiment duration in minutes")
ap.add_argument("--sample", action="store_true",
help="Run with embedded sample inputs (5%% traffic, 100k users, 30 min)")
ap.add_argument("--baseline-availability", type=float, default=0.999, help="Baseline availability (default: 0.999)")
ap.add_argument("--expected-impact-availability", type=float, default=0.95, dest="impact_avail",
help="Availability under fault (default: 0.95)")
@@ -81,9 +83,16 @@ def main():
ap.add_argument("--format", choices=["text", "json"], default="text")
args = ap.parse_args()
if args.sample:
traffic_share, user_pop, duration_min = 0.05, 100000, 30
elif None not in (args.traffic_share, args.user_pop, args.duration_min):
traffic_share, user_pop, duration_min = args.traffic_share, args.user_pop, args.duration_min
else:
ap.error("--traffic-share, --user-pop and --duration-min are required (or use --sample)")
try:
result = calculate(
args.traffic_share, args.user_pop, args.duration_min,
traffic_share, user_pop, duration_min,
args.baseline_availability, args.impact_avail, args.monthly_budget_min,
)
except ValueError as e:
engineering/claude-coach/skills/claude-coach/scripts/cheat_code_filter.py
+7 -3
View File
@@ -118,12 +118,16 @@ def render_human(picks: list[Technique]) -> str:
return "\n".join(out)
def sample_run() -> int:
def sample_run(as_json: bool = False) -> int:
sample_path = DEFAULT_GLOSSARY
if not sample_path.exists():
print("Sample glossary not found; place references/cheat-codes.md alongside this script.", file=sys.stderr)
return 1
picks = rank(parse_glossary(sample_path), ["writing", "coding"], 5)
use_cases = ["writing", "coding"]
picks = rank(parse_glossary(sample_path), use_cases, 5)
if as_json:
print(json.dumps({"use_cases": use_cases, "picks": [asdict(t) for t in picks]}, indent=2))
return 0
print(render_human(picks))
return 0
@@ -138,7 +142,7 @@ def main(argv: list[str] | None = None) -> int:
args = parser.parse_args(argv)
if args.sample:
return sample_run()
return sample_run(args.json)
if not args.use_cases:
parser.error("--use-cases is required unless --sample is passed")
engineering/claude-coach/skills/claude-coach/scripts/coach_tip_classifier.py
+7 -4
View File
@@ -164,7 +164,7 @@ def render_human(d: Decision) -> str:
return "\n".join(out)
def sample_run() -> int:
def sample_run(as_json: bool = False) -> int:
cases = [
("Can you help me with my email?", False),
("Write a 200-word product description for a noise-cancelling headphone targeting remote workers, focused on the focus-time benefit, no marketing fluff.", False),
@@ -172,8 +172,11 @@ def sample_run() -> int:
("Can you make this better?", True),
("stop with the tips, just rewrite it", False),
]
for prompt, prev in cases:
d = classify(prompt, previous_tip_given=prev)
decisions = [classify(prompt, previous_tip_given=prev) for prompt, prev in cases]
if as_json:
print(json.dumps([asdict(d) for d in decisions], indent=2))
return 0
for d in decisions:
print(render_human(d))
print("-" * 60)
return 0
@@ -188,7 +191,7 @@ def main(argv: list[str] | None = None) -> int:
args = parser.parse_args(argv)
if args.sample:
return sample_run()
return sample_run(args.json)
if not args.prompt:
parser.error("--prompt is required unless --sample is passed")
engineering/claude-coach/skills/claude-coach/scripts/prompt_rater.py
+7 -4
View File
@@ -138,14 +138,17 @@ def render_human(r: Rating) -> str:
)
def sample_run() -> int:
def sample_run(as_json: bool = False) -> int:
samples = [
"Can you help me with my email?",
"Write a 200-word product description for a noise-cancelling headphone targeting remote workers, focused on the focus-time benefit, no marketing fluff.",
"thoughts?",
]
for s in samples:
r = rate(s)
ratings = [rate(s) for s in samples]
if as_json:
print(json.dumps([asdict(r) for r in ratings], indent=2))
return 0
for r in ratings:
print(render_human(r))
print("-" * 60)
return 0
@@ -159,7 +162,7 @@ def main(argv: list[str] | None = None) -> int:
args = parser.parse_args(argv)
if args.sample:
return sample_run()
return sample_run(args.json)
if not args.prompt:
parser.error("--prompt is required unless --sample is passed")
engineering/skills/api-design-reviewer/scripts/api_linter.py
+54 -12
View File
@@ -826,6 +826,35 @@ class APILinter:
return "\n".join(report_lines)
# Embedded sample OpenAPI spec — intentionally imperfect (a verb in a URL, a
# snake_case property) so --sample produces a representative report.
SAMPLE_OPENAPI_SPEC = {
"openapi": "3.0.0",
"info": {"title": "Sample API", "version": "1.0.0"},
"servers": [{"url": "https://api.example.com"}],
"paths": {
"/user-profiles/{userId}": {
"get": {
"summary": "Get a user profile",
"responses": {"200": {"description": "OK"}, "404": {"description": "Not found"}},
"parameters": [{"name": "userId", "in": "path", "required": True}],
}
},
"/user-profiles/create": {
"post": {
"summary": "Create a user profile (verb-in-URL anti-pattern)",
"responses": {"201": {"description": "Created"}},
}
},
},
"components": {
"schemas": {
"UserProfile": {"properties": {"first_name": {"type": "string"}}}
}
},
}
def main():
"""Main CLI entry point."""
parser = argparse.ArgumentParser(
@@ -836,13 +865,21 @@ Examples:
python api_linter.py openapi.json
python api_linter.py --format json openapi.json > report.json
python api_linter.py --raw-endpoints endpoints.json
python api_linter.py --sample --format json
"""
)
parser.add_argument(
'input_file',
nargs='?',
help='Input file: OpenAPI/Swagger JSON file or raw endpoints JSON'
)
parser.add_argument(
'--sample',
action='store_true',
help='Lint an embedded sample OpenAPI spec (no input file needed)'
)
parser.add_argument(
'--format',
@@ -863,17 +900,22 @@ Examples:
)
args = parser.parse_args()
# Load input file
try:
with open(args.input_file, 'r') as f:
input_data = json.load(f)
except FileNotFoundError:
print(f"Error: Input file '{args.input_file}' not found.", file=sys.stderr)
return 1
except json.JSONDecodeError as e:
print(f"Error: Invalid JSON in '{args.input_file}': {e}", file=sys.stderr)
return 1
# Load input data — from the embedded sample or the input file
if args.sample:
input_data = SAMPLE_OPENAPI_SPEC
else:
if not args.input_file:
parser.error("input_file is required (or use --sample)")
try:
with open(args.input_file, 'r') as f:
input_data = json.load(f)
except FileNotFoundError:
print(f"Error: Input file '{args.input_file}' not found.", file=sys.stderr)
return 1
except json.JSONDecodeError as e:
print(f"Error: Invalid JSON in '{args.input_file}': {e}", file=sys.stderr)
return 1
# Initialize linter and run analysis
linter = APILinter()
engineering/skills/chaos-engineering/scripts/blast_radius_calculator.py
+13 -4
View File
@@ -70,9 +70,11 @@ def render_text(result):
def main():
ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
ap.add_argument("--traffic-share", type=float, required=True, help="Fraction (0-1) of traffic affected")
ap.add_argument("--user-pop", type=int, required=True, help="Total user population")
ap.add_argument("--duration-min", type=int, required=True, help="Experiment duration in minutes")
ap.add_argument("--traffic-share", type=float, help="Fraction (0-1) of traffic affected")
ap.add_argument("--user-pop", type=int, help="Total user population")
ap.add_argument("--duration-min", type=int, help="Experiment duration in minutes")
ap.add_argument("--sample", action="store_true",
help="Run with embedded sample inputs (5%% traffic, 100k users, 30 min)")
ap.add_argument("--baseline-availability", type=float, default=0.999, help="Baseline availability (default: 0.999)")
ap.add_argument("--expected-impact-availability", type=float, default=0.95, dest="impact_avail",
help="Availability under fault (default: 0.95)")
@@ -81,9 +83,16 @@ def main():
ap.add_argument("--format", choices=["text", "json"], default="text")
args = ap.parse_args()
if args.sample:
traffic_share, user_pop, duration_min = 0.05, 100000, 30
elif None not in (args.traffic_share, args.user_pop, args.duration_min):
traffic_share, user_pop, duration_min = args.traffic_share, args.user_pop, args.duration_min
else:
ap.error("--traffic-share, --user-pop and --duration-min are required (or use --sample)")
try:
result = calculate(
args.traffic_share, args.user_pop, args.duration_min,
traffic_share, user_pop, duration_min,
args.baseline_availability, args.impact_avail, args.monthly_budget_min,
)
except ValueError as e:
engineering/skills/secrets-vault-manager/scripts/audit_log_analyzer.py
+44 -6
View File
@@ -278,6 +278,32 @@ def print_human(result, threshold):
print(" (* = off-hours)")
# Embedded synthetic audit log — exercises volume-spike + off-hours + failed-access
# detectors so --sample produces a non-trivial report without a real log file.
SAMPLE_ENTRIES = [
{"timestamp": "2026-03-20T03:14:00Z", "type": "request",
"auth": {"display_name": "approle-payment-svc"},
"request": {"path": "secret/data/production/payment/api-keys", "operation": "read"},
"response": {"status_code": 200}, "remote_address": "10.0.1.15"},
{"timestamp": "2026-03-20T03:15:00Z", "type": "request",
"auth": {"display_name": "approle-payment-svc"},
"request": {"path": "secret/data/production/payment/db", "operation": "read"},
"response": {"status_code": 200}, "remote_address": "10.0.1.99"},
{"timestamp": "2026-03-20T03:16:00Z", "type": "request",
"auth": {"display_name": "approle-payment-svc"},
"request": {"path": "secret/data/production/payment/jwt", "operation": "read"},
"response": {"status_code": 403}, "remote_address": "203.0.113.7"},
{"timestamp": "2026-03-20T03:17:00Z", "type": "request",
"auth": {"display_name": "approle-payment-svc"},
"request": {"path": "secret/data/production/payment/jwt", "operation": "read"},
"response": {"status_code": 403}, "remote_address": "203.0.113.7"},
{"timestamp": "2026-03-20T14:00:00Z", "type": "request",
"auth": {"display_name": "ci-runner"},
"request": {"path": "secret/data/ci/tokens", "operation": "read"},
"response": {"status_code": 200}, "remote_address": "10.0.2.20"},
]
def main():
parser = argparse.ArgumentParser(
description="Analyze Vault/cloud secret manager audit logs for anomalies.",
@@ -299,7 +325,7 @@ def main():
%(prog)s --log-file audit.json --threshold 3 --json
"""),
)
parser.add_argument("--log-file", required=True, help="Path to audit log file (JSON lines or JSON array)")
parser.add_argument("--log-file", help="Path to audit log file (JSON lines or JSON array)")
parser.add_argument(
"--threshold",
type=int,
@@ -307,23 +333,35 @@ def main():
help="Anomaly sensitivity threshold — lower = more sensitive (default: 5)",
)
parser.add_argument("--json", action="store_true", dest="json_output", help="Output as JSON")
parser.add_argument("--sample", action="store_true",
help="Analyze an embedded synthetic audit log")
args = parser.parse_args()
entries = load_logs(args.log_file)
if args.sample:
entries = SAMPLE_ENTRIES
log_file = "<embedded sample>"
threshold = 2
else:
if not args.log_file:
parser.error("--log-file is required (or use --sample)")
entries = load_logs(args.log_file)
log_file = args.log_file
threshold = args.threshold
if not entries:
print("No log entries found in file.", file=sys.stderr)
sys.exit(1)
result = analyze(entries, args.threshold)
result["log_file"] = args.log_file
result["threshold"] = args.threshold
result = analyze(entries, threshold)
result["log_file"] = log_file
result["threshold"] = threshold
result["analyzed_at"] = datetime.now().isoformat()
if args.json_output:
print(json.dumps(result, indent=2))
else:
print_human(result, args.threshold)
print_human(result, threshold)
if __name__ == "__main__":
engineering/skills/slo-architect/scripts/error_budget_calculator.py
+10 -2
View File
@@ -125,13 +125,21 @@ def render_text(result):
def main():
ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
ap.add_argument("--target", type=float, required=True, help="Target percent (e.g., 99.9)")
ap.add_argument("--target", type=float, help="Target percent (e.g., 99.9)")
ap.add_argument("--window-days", type=int, default=28, help="Window in days (default: 28)")
ap.add_argument("--format", choices=["text", "json"], default="text")
ap.add_argument("--sample", action="store_true", help="Run with embedded sample inputs (99.9%% / 28d)")
args = ap.parse_args()
if args.sample:
target, window_days = 99.9, 28
elif args.target is not None:
target, window_days = args.target, args.window_days
else:
ap.error("--target is required (or use --sample)")
try:
result = compute(args.target, args.window_days)
result = compute(target, window_days)
except ValueError as e:
print(f"ERROR: {e}", file=sys.stderr)
return 2
engineering/skills/slo-architect/scripts/slo_review.py
+25 -8
View File
@@ -64,8 +64,16 @@ def _has_cpu_as_sli(text):
return False
def audit_one(path):
text = _read(path)
# Embedded sample SLO doc — intentionally flawed (target too high, CPU-as-SLI,
# no error budget policy) so --sample exercises several finding paths.
SAMPLE_SLO_DOC = """# Checkout API SLO
target: 99.995%
window_days: 28
sli: cpu_usage below 80%
"""
def audit_text(text):
findings = []
target = _parse_target(text)
window_days = _parse_window_days(text)
@@ -103,6 +111,10 @@ def audit_one(path):
return findings
def audit_one(path):
return audit_text(_read(path))
def _walk(target):
if os.path.isfile(target):
yield target
@@ -140,15 +152,20 @@ def render_text(results):
def main():
ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
ap.add_argument("--slo-doc", required=True, help="Path to SLO doc or directory of docs")
ap.add_argument("--slo-doc", help="Path to SLO doc or directory of docs")
ap.add_argument("--format", choices=["text", "json"], default="text")
ap.add_argument("--sample", action="store_true", help="Audit an embedded sample SLO doc")
args = ap.parse_args()
if not os.path.exists(args.slo_doc):
print(f"ERROR: not found: {args.slo_doc}", file=sys.stderr)
return 2
results = audit(args.slo_doc)
if args.sample:
results = [{"path": "<embedded sample>", "findings": audit_text(SAMPLE_SLO_DOC)}]
else:
if not args.slo_doc:
ap.error("--slo-doc is required (or use --sample)")
if not os.path.exists(args.slo_doc):
print(f"ERROR: not found: {args.slo_doc}", file=sys.stderr)
return 2
results = audit(args.slo_doc)
if args.format == "json":
print(json.dumps(results, indent=2))
return 1 if any(f[0] == "FAIL" for r in results for f in r["findings"]) else 0
engineering/slo-architect/skills/slo-architect/scripts/error_budget_calculator.py
+10 -2
View File
@@ -125,13 +125,21 @@ def render_text(result):
def main():
ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
ap.add_argument("--target", type=float, required=True, help="Target percent (e.g., 99.9)")
ap.add_argument("--target", type=float, help="Target percent (e.g., 99.9)")
ap.add_argument("--window-days", type=int, default=28, help="Window in days (default: 28)")
ap.add_argument("--format", choices=["text", "json"], default="text")
ap.add_argument("--sample", action="store_true", help="Run with embedded sample inputs (99.9%% / 28d)")
args = ap.parse_args()
if args.sample:
target, window_days = 99.9, 28
elif args.target is not None:
target, window_days = args.target, args.window_days
else:
ap.error("--target is required (or use --sample)")
try:
result = compute(args.target, args.window_days)
result = compute(target, window_days)
except ValueError as e:
print(f"ERROR: {e}", file=sys.stderr)
return 2
engineering/slo-architect/skills/slo-architect/scripts/slo_review.py
+25 -8
View File
@@ -64,8 +64,16 @@ def _has_cpu_as_sli(text):
return False
def audit_one(path):
text = _read(path)
# Embedded sample SLO doc — intentionally flawed (target too high, CPU-as-SLI,
# no error budget policy) so --sample exercises several finding paths.
SAMPLE_SLO_DOC = """# Checkout API SLO
target: 99.995%
window_days: 28
sli: cpu_usage below 80%
"""
def audit_text(text):
findings = []
target = _parse_target(text)
window_days = _parse_window_days(text)
@@ -103,6 +111,10 @@ def audit_one(path):
return findings
def audit_one(path):
return audit_text(_read(path))
def _walk(target):
if os.path.isfile(target):
yield target
@@ -140,15 +152,20 @@ def render_text(results):
def main():
ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
ap.add_argument("--slo-doc", required=True, help="Path to SLO doc or directory of docs")
ap.add_argument("--slo-doc", help="Path to SLO doc or directory of docs")
ap.add_argument("--format", choices=["text", "json"], default="text")
ap.add_argument("--sample", action="store_true", help="Audit an embedded sample SLO doc")
args = ap.parse_args()
if not os.path.exists(args.slo_doc):
print(f"ERROR: not found: {args.slo_doc}", file=sys.stderr)
return 2
results = audit(args.slo_doc)
if args.sample:
results = [{"path": "<embedded sample>", "findings": audit_text(SAMPLE_SLO_DOC)}]
else:
if not args.slo_doc:
ap.error("--slo-doc is required (or use --sample)")
if not os.path.exists(args.slo_doc):
print(f"ERROR: not found: {args.slo_doc}", file=sys.stderr)
return 2
results = audit(args.slo_doc)
if args.format == "json":
print(json.dumps(results, indent=2))
return 1 if any(f[0] == "FAIL" for r in results for f in r["findings"]) else 0
productivity/handoff/skills/handoff/scripts/redaction_linter.py
+26 -21
View File
@@ -209,6 +209,27 @@ def _format_human(report: Report, mode: str, path: Path) -> str:
return "\n".join(lines)
def _report_json(report, mode: str, file_label: str) -> str:
return json.dumps(
{
"file": file_label,
"mode": mode,
"findings": [
{
"line": f.line_number,
"pattern": f.pattern_name,
"severity": f.severity,
"match": f.match,
"suggestion": f.suggestion,
}
for f in report.findings
],
"counts": report.by_severity(),
},
indent=2,
)
def main(argv: list[str] | None = None) -> int:
parser = argparse.ArgumentParser(description="Scan a handoff draft for secrets and PII.")
parser.add_argument("file", nargs="?", help="Path to the handoff markdown file.")
@@ -237,7 +258,10 @@ def main(argv: list[str] | None = None) -> int:
"Allowed: AKIAIOSFODNN7EXAMPLE <!-- handoff:allow secret -->\n"
)
report = scan_text(fixture)
print(_format_human(report, "strict", Path("<sample>")))
if args.json:
print(_report_json(report, "strict", "<sample>"))
else:
print(_format_human(report, "strict", Path("<sample>")))
return 1 if report.findings else 0
if args.mode == "off":
@@ -255,26 +279,7 @@ def main(argv: list[str] | None = None) -> int:
report = scan_file(path)
if args.json:
print(
json.dumps(
{
"file": str(path),
"mode": args.mode,
"findings": [
{
"line": f.line_number,
"pattern": f.pattern_name,
"severity": f.severity,
"match": f.match,
"suggestion": f.suggestion,
}
for f in report.findings
],
"counts": report.by_severity(),
},
indent=2,
)
)
print(_report_json(report, args.mode, str(path)))
else:
print(_format_human(report, args.mode, path))
scripts/smoke_json_output.py
Executable
+227
View File
@@ -0,0 +1,227 @@
#!/usr/bin/env python3
"""JSON-output verification gate for Python tools (audit gate G9).
Companion to smoke_scripts.py (which only asserts `--help` exits 0). Many tools
advertise `--json` or `--format json` in their help text but require positional
or required arguments before they can emit anything — so a bare-flag smoke test
reports false failures (see issue #654).
This harness verifies JSON output the way the tools are actually meant to run:
1. Discover every tool whose `--help` advertises JSON output AND an embedded
`--sample` fixture (the chosen convention — issue #654 Option A).
2. Run `<tool> --sample <json-flag>` and assert stdout parses as JSON.
Tools that advertise JSON output but do NOT yet expose `--sample` are reported
as "uncovered" — a to-do list for backporting the convention, not a failure
(so the gate can be adopted incrementally without going red on day one).
Pass --strict to treat uncovered JSON tools as failures once coverage is high.
Exit codes:
0 every --sample JSON tool produced valid JSON (and, with --strict, every
JSON-advertising tool exposes --sample)
1 one or more --sample JSON runs produced invalid JSON / errored
2 harness error
Usage:
python3 scripts/smoke_json_output.py # human-readable report
python3 scripts/smoke_json_output.py --json # machine-readable report
python3 scripts/smoke_json_output.py --strict # uncovered JSON tools fail
"""
from __future__ import annotations
import argparse
import concurrent.futures
import json
import os
import re
import subprocess
import sys
REPO_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
TIMEOUT_SECONDS = 20
# Reuse the same exclusion set + exceptions file as the --help gate.
EXCLUDE_DIRS = {
".git", ".codex", ".gemini", ".hermes", ".vibe", "docs", "audit",
"node_modules", "integrations", "eval-workspace", "site",
"__pycache__", ".venv", "venv",
}
EXCEPTIONS_FILE = os.path.join(REPO_ROOT, "scripts", "smoke_exceptions.txt")
# The smoke harnesses describe `--sample`/`--json` in their own help text but
# are gate runners, not analysis tools — never classify them as JSON tools.
SELF_SKIP = {"scripts/smoke_json_output.py", "scripts/smoke_scripts.py"}
# `--format json` is only a valid invocation when help shows json as a choice,
# e.g. `--format {text,json}`. A bare mention of the word "format" elsewhere in
# help must not trigger it (that misfires on tools that only accept `--json`).
_FORMAT_JSON_RE = re.compile(r"--format[ =]?\{[^}]*\bjson\b[^}]*\}")
def load_exceptions(path):
exceptions = {}
if not os.path.isfile(path):
return exceptions
with open(path, "r", encoding="utf-8") as f:
for raw in f:
line = raw.strip()
if not line or line.startswith("#"):
continue
rel = line.split("#", 1)[0].strip()
if rel:
exceptions[rel] = True
return exceptions
def find_python_files(root):
files = []
for dirpath, dirnames, filenames in os.walk(root):
dirnames[:] = sorted(d for d in dirnames if d not in EXCLUDE_DIRS)
for name in sorted(filenames):
if name.endswith(".py"):
files.append(os.path.relpath(
os.path.join(dirpath, name), root).replace(os.sep, "/"))
return files
def _help_text(abs_path):
try:
proc = subprocess.run(
[sys.executable, abs_path, "--help"],
stdin=subprocess.DEVNULL, capture_output=True, text=True,
timeout=TIMEOUT_SECONDS, cwd=os.path.dirname(abs_path),
)
except (subprocess.TimeoutExpired, OSError):
return ""
return (proc.stdout or "") + (proc.stderr or "") if proc.returncode == 0 else ""
def classify(rel_path):
"""Return (advertises_json, json_flag, has_sample) for one tool."""
if rel_path in SELF_SKIP:
return False, None, False
abs_path = os.path.join(REPO_ROOT, rel_path)
help_text = _help_text(abs_path)
if not help_text:
return False, None, False
low = help_text.lower()
# Prefer `--format json` only when help shows json as an actual choice;
# otherwise fall back to a plain `--json` flag.
json_flag = None
if _FORMAT_JSON_RE.search(low):
json_flag = ["--format", "json"]
elif re.search(r"(?<![\w-])--json(?![\w-])", low):
json_flag = ["--json"]
advertises_json = json_flag is not None
has_sample = "--sample" in low
return advertises_json, json_flag, has_sample
def verify_one(rel_path, json_flag):
"""Run `<tool> --sample <json_flag>` and check stdout parses as JSON."""
abs_path = os.path.join(REPO_ROOT, rel_path)
try:
proc = subprocess.run(
[sys.executable, abs_path, "--sample", *json_flag],
stdin=subprocess.DEVNULL, capture_output=True, text=True,
timeout=TIMEOUT_SECONDS, cwd=os.path.dirname(abs_path),
)
except subprocess.TimeoutExpired:
return rel_path, False, f"timeout after {TIMEOUT_SECONDS}s"
except OSError as exc:
return rel_path, False, f"could not execute: {exc}"
# A non-zero exit is acceptable only if the tool intentionally signals a
# finding through its exit code (e.g. blast_radius RED) — but it must still
# have emitted valid JSON on stdout.
out = (proc.stdout or "").strip()
if not out:
tail = (proc.stderr or "").strip().splitlines()
return rel_path, False, f"no stdout (exit {proc.returncode}): {tail[-1] if tail else ''}"[:200]
try:
json.loads(out)
except json.JSONDecodeError as exc:
return rel_path, False, f"stdout is not valid JSON: {exc}"
return rel_path, True, ""
def main(argv=None):
parser = argparse.ArgumentParser(description=__doc__,
formatter_class=argparse.RawDescriptionHelpFormatter)
parser.add_argument("--json", action="store_true",
help="emit a JSON report instead of human-readable output")
parser.add_argument("--strict", action="store_true",
help="treat JSON-advertising tools without --sample as failures")
parser.add_argument("--jobs", type=int, default=os.cpu_count() or 4,
help="parallel workers (default: CPU count)")
parser.add_argument("--root", default=REPO_ROOT, help="repo root")
args = parser.parse_args(argv)
try:
exceptions = load_exceptions(EXCEPTIONS_FILE)
except OSError as exc:
print(f"ERROR: cannot read exceptions file: {exc}", file=sys.stderr)
return 2
all_files = [f for f in find_python_files(args.root) if f not in exceptions]
# Phase 1: classify in parallel.
json_tools = {} # rel_path -> json_flag
uncovered = [] # advertises json but no --sample
with concurrent.futures.ThreadPoolExecutor(max_workers=args.jobs) as pool:
results = pool.map(lambda f: (f, *classify(f)), all_files)
for rel_path, advertises, json_flag, has_sample in results:
if not advertises:
continue
if has_sample:
json_tools[rel_path] = json_flag
else:
uncovered.append(rel_path)
uncovered.sort()
# Phase 2: verify covered tools in parallel.
failures = []
with concurrent.futures.ThreadPoolExecutor(max_workers=args.jobs) as pool:
for rel_path, ok, detail in pool.map(
lambda item: verify_one(item[0], item[1]), sorted(json_tools.items())):
if not ok:
failures.append({"file": rel_path, "detail": detail})
failures.sort(key=lambda f: f["file"])
covered = len(json_tools)
total_json = covered + len(uncovered)
coverage_pct = round(100 * covered / total_json, 1) if total_json else 100.0
if args.json:
print(json.dumps({
"json_advertising_tools": total_json,
"covered_by_sample": covered,
"coverage_pct": coverage_pct,
"verified_ok": covered - len(failures),
"failed": failures,
"uncovered": uncovered,
}, indent=2))
else:
print(f"JSON-advertising tools: {total_json}")
print(f"Covered by --sample: {covered} ({coverage_pct}%)")
print(f"Verified valid JSON: {covered - len(failures)}")
print(f"Failed: {len(failures)}")
print(f"Uncovered (no --sample): {len(uncovered)}")
if failures:
print("\nFAILURES:")
for f in failures:
print(f" {f['file']}\n {f['detail']}")
if uncovered:
print("\nUNCOVERED (advertise JSON but lack --sample — backport target):")
for f in uncovered:
print(f" {f}")
if failures:
return 1
if args.strict and uncovered:
return 1
return 0
if __name__ == "__main__":
sys.exit(main())