Nick
c3c95688f7
Seeds the pipeline with the REAL #192 CONVERGED case-mismatch payload: the safety-net convergence step exported Python-cased True/False from json.load while the steps compared env.CONVERGED == 'false' — both conditionals unreachable (discharged on the release-safety-net branch, public in PR #211's diff). Captured via the pipeline itself (--class env-case-mismatch --source ci, 0 redactions — the payload is already public and credential-free). The generated regression test asserts the boolean-export checker still FLAGS the recorded payload (checker_expectation: flags), so a recurrence of the class fails CI. Manifest gains checker_expectation: capture records whether the class's checker raises on the payload ('flags' — the raise IS the recurrence alarm) or must stay quiet ('passes'); the generated tests enforce the recorded expectation, so a checker that stops catching its recorded defect rots the suite red. Docs: bugfix-contract §10 gains the operational capture step + explicit trivial-bug exemption list; README dev section points at the pipeline; tests/fixtures/production/README.md documents the flow. Henxels: capture_failure.py required in scripts/; production fixtures governed (manifest.json + README.md required); payload.json and manifest.json travel together (changed_with both directions, verified to bite on a payload-only staged change).
493 lines
21 KiB
Python
493 lines
21 KiB
Python
#!/usr/bin/env python3
|
|
"""Production failure → sanitised regression fixture (#157, plan 2026-09-07-157).
|
|
|
|
One command, zero ceremony: a payload file (raw error text, JSON blob, or a
|
|
subsystem_health record excerpt) plus ``--class`` and ``--source`` become a
|
|
redacted fixture under ``tests/fixtures/production/<class>/<id>/payload.json``
|
|
and a manifest entry in ``tests/fixtures/production/manifest.json``. The
|
|
manifest carries the payload's sha256 so a fixture edited without re-running
|
|
capture fails the generated regression tests (tamper-evidence).
|
|
|
|
Redaction reuses the credential patterns discharged for the failure auto-filer
|
|
(#122, commit 9105ca8 — query-string credential params, URL userinfo/DSNs,
|
|
Authorization/bearer headers, assignment forms) so that a fixture sees exactly
|
|
what the filer would have persisted. The patterns are copied here because the
|
|
filer module ships on the failure-auto-filing branch; when that branch merges,
|
|
import ``rune.failure_auto_filer.redact`` directly and delete the copy.
|
|
``tests/test_capture_failure.py::test_redact_matches_filer_on_shared_cases``
|
|
pins parity on the filer's own test literals so the copies cannot drift.
|
|
|
|
Sanitisation is conservative (plan design decision 1): redact first, then run
|
|
an INDEPENDENT residual scan for credential-shaped spans. If anything
|
|
credential-shaped survives, capture aborts loudly (ResidualSecretError /
|
|
exit code 3) and persists nothing — a leaked fixture is an incident.
|
|
|
|
Usage:
|
|
python3 scripts/capture_failure.py payload.txt --class ci-crash \
|
|
--source ci [--id my-slug] [--fixtures-root tests/fixtures/production]
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
import re
|
|
import sys
|
|
from datetime import date, datetime, timezone
|
|
from pathlib import Path
|
|
|
|
EXIT_OK = 0
|
|
EXIT_USAGE = 2
|
|
EXIT_RESIDUAL_SECRET = 3
|
|
|
|
DEFAULT_FIXTURES_ROOT = "tests/fixtures/production"
|
|
VALID_SOURCES = ("issue", "ci", "manual")
|
|
SAFE_NAME_RE = re.compile(r"^[a-z0-9][a-z0-9._-]*$", re.IGNORECASE)
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Redaction — copied verbatim from src/rune/failure_auto_filer.py (9105ca8, #122)
|
|
# ---------------------------------------------------------------------------
|
|
# Credentials must never persist in fixtures. Patterns (order matters
|
|
# little; all are applied): query-string credential params, URL userinfo
|
|
# (scheme://user:pass@host — covers postgres/redis/mysql DSNs and basic
|
|
# auth), Authorization headers / bearer tokens, and assignment forms
|
|
# (password=..., secret=...). Only the credential SPAN is replaced — the
|
|
# surrounding error text (class, host, port, message shape) survives.
|
|
_QUERY_CREDS = re.compile(
|
|
r"([?&](?:token|password|passwd|secret|api_key|apikey|access_token|key|sig|signature|credential)=)[^&\s'\"]+",
|
|
re.IGNORECASE,
|
|
)
|
|
_URL_USERINFO = re.compile(r"\b([a-zA-Z][a-zA-Z0-9+.-]*://)([^\s:/@'\"]+)?:([^\s/@'\"]+)@")
|
|
_BEARER = re.compile(r"(?i)\b((?:authorization\s*:?\s*(?:bearer\s+)?|bearer\s+))[^\s,'\";]+")
|
|
_ASSIGNMENT = re.compile(
|
|
r"(?i)\b((?:password|passwd|secret|api_key|apikey|access_token|auth)\s*[=:]\s*)['\"]?([^\s'\"]{4,})"
|
|
)
|
|
|
|
# Scan passes get progressively less specific; report the FIRST matching pass
|
|
# per hit so the abort message names the tightest classification.
|
|
REDACTION_PASSES = (
|
|
("query-creds", _QUERY_CREDS),
|
|
("dsn-userinfo", _URL_USERINFO),
|
|
("bearer", _BEARER),
|
|
("assignment", _ASSIGNMENT),
|
|
)
|
|
|
|
|
|
def redact(text: str) -> str:
|
|
"""Strip credentials from an error string (no-op on clean text)."""
|
|
out = _QUERY_CREDS.sub(r"\1<redacted>", text)
|
|
out = _URL_USERINFO.sub(r"\1<redacted>:<redacted>@", out)
|
|
out = _BEARER.sub(r"\1<redacted>", out)
|
|
out = _ASSIGNMENT.sub(lambda m: f"{m.group(1)}<redacted>", out)
|
|
return out
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Residual-secret scan — deliberately independent of the redaction spans
|
|
# ---------------------------------------------------------------------------
|
|
# The scan must catch what redaction MISSES (e.g. a quote inside the value
|
|
# defeats the assignment span). It therefore re-derives credential shapes from
|
|
# scratch and only forgives values that are already the redaction marker or
|
|
# are too short to be a credential (1-char leftovers of truncation).
|
|
|
|
_SCAN_ASSIGN = re.compile(
|
|
r"(?i)\b((?:password|passwd|secret|api_key|apikey|access_token|auth)\s*[=:]\s*)['\"]?([^\s'\"]{2,})"
|
|
)
|
|
_SCAN_QUERY = re.compile(
|
|
r"(?i)((?:[?&])(?:token|password|passwd|secret|api_key|apikey|access_token|key|sig|signature|credential)=)"
|
|
r"([^\s&'\"]+)"
|
|
)
|
|
_SCAN_BEARER = re.compile(r"(?i)\b((?:authorization\s*:?\s*(?:bearer\s+)?|bearer\s+))([^\s,'\"]+)")
|
|
_SCAN_DSN = re.compile(r"\b([a-zA-Z][a-zA-Z0-9+.-]*://)([^\s/@'\"]*)@")
|
|
_REDACTED = re.compile(r"redacted", re.IGNORECASE)
|
|
|
|
|
|
def residual_secrets(text: str) -> list[str]:
|
|
"""Credential-shaped spans that survived redaction (empty list = clean)."""
|
|
hits: list[str] = []
|
|
for pattern in (_SCAN_ASSIGN, _SCAN_QUERY, _SCAN_BEARER, _SCAN_DSN):
|
|
for match in pattern.finditer(text):
|
|
groups = match.groups()
|
|
value = groups[-1] if groups else match.group(0)
|
|
if _REDACTED.search(value) or _REDACTED.search(match.group(0)):
|
|
continue
|
|
# truncation slivers (a single stray char) are noise, not secrets
|
|
if len(value) < 2:
|
|
continue
|
|
hits.append(match.group(0))
|
|
return hits
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Capture
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _assert_safe_name(name: str, what: str) -> None:
|
|
if not name or not SAFE_NAME_RE.match(name) or ".." in name:
|
|
raise ValueError(f"unsafe {what}: {name!r} (must be a flat slug, no path separators)")
|
|
|
|
|
|
def _original_shape(raw: str) -> str:
|
|
stripped = raw.strip()
|
|
if stripped.startswith(("{", "[")):
|
|
try:
|
|
json.loads(stripped)
|
|
return "json"
|
|
except json.JSONDecodeError:
|
|
return "text"
|
|
return "text"
|
|
|
|
|
|
def capture(
|
|
payload_path: str | Path,
|
|
failure_class: str,
|
|
source: str,
|
|
fixture_id: str | None = None,
|
|
fixtures_root: str | Path = DEFAULT_FIXTURES_ROOT,
|
|
now: datetime | None = None,
|
|
) -> dict:
|
|
"""Redact a payload, store the fixture, upsert the manifest entry.
|
|
|
|
Returns the manifest entry dict. Raises ResidualSecretError (persisting
|
|
nothing) when credential-shaped text survives redaction; ValueError for
|
|
bad failure_class/source/fixture_id.
|
|
"""
|
|
if source not in VALID_SOURCES:
|
|
raise ValueError(f"unknown --source {source!r} (expected one of {', '.join(VALID_SOURCES)})")
|
|
_assert_safe_name(failure_class, "failure class")
|
|
|
|
payload_file = Path(payload_path)
|
|
raw = payload_file.read_text(encoding="utf-8")
|
|
|
|
shape = _original_shape(raw)
|
|
redacted = redact(raw)
|
|
residuals = residual_secrets(redacted)
|
|
if residuals:
|
|
raise ResidualSecretError(
|
|
"residual credential-shaped text survived redaction — fixture NOT written. "
|
|
f"Redact manually (or broaden the redaction patterns) and retry: {residuals[:5]}"
|
|
)
|
|
|
|
kinds = [kind for kind, pattern in REDACTION_PASSES if pattern.search(raw)]
|
|
redactions_applied = max(redacted.count("<redacted>") - raw.count("<redacted>"), 0)
|
|
|
|
slug = fixture_id or _slugify(f"{failure_class}-{_headline(redacted)}")
|
|
_assert_safe_name(slug, "fixture id")
|
|
|
|
class_dir = Path(fixtures_root) / failure_class
|
|
fixture_dir = class_dir / slug
|
|
fixture_dir.mkdir(parents=True, exist_ok=True)
|
|
fixture_file = fixture_dir / "payload.json"
|
|
|
|
if fixture_file.exists():
|
|
stored = json.loads(fixture_file.read_text(encoding="utf-8"))
|
|
else:
|
|
stored = {}
|
|
stored.update(
|
|
{
|
|
"id": slug,
|
|
"failure_class": failure_class,
|
|
"source": source,
|
|
"date": (now or datetime.now(timezone.utc)).date().isoformat(),
|
|
"original_shape": shape,
|
|
"redactions_applied": redactions_applied,
|
|
"redaction_kinds": kinds,
|
|
"payload": redacted,
|
|
}
|
|
)
|
|
fixture_file.write_text(json.dumps(stored, indent=2, sort_keys=True) + "\n", encoding="utf-8")
|
|
|
|
sha256 = hashlib.sha256(redacted.encode("utf-8")).hexdigest()
|
|
# Does the class's checker FLAG this payload, or must it stay quiet?
|
|
# Defect-class payloads (the recorded payload exhibits the bug) are
|
|
# "flags": the generated regression test asserts the checker still
|
|
# catches them, so a recurrence fails CI. Everything else is "passes".
|
|
try:
|
|
check_payload(failure_class, redacted)
|
|
expectation = "passes"
|
|
except AssertionError:
|
|
expectation = "flags"
|
|
entry = {
|
|
"id": slug,
|
|
"class": failure_class,
|
|
"source": source,
|
|
"date": stored["date"],
|
|
"original_shape": shape,
|
|
"redactions_applied": redactions_applied,
|
|
"redaction_kinds": kinds,
|
|
"checker_expectation": expectation,
|
|
"sha256": sha256,
|
|
}
|
|
_upsert_manifest(Path(fixtures_root), entry)
|
|
return entry
|
|
|
|
|
|
class ResidualSecretError(RuntimeError):
|
|
"""A credential-shaped span survived redaction — capture aborted."""
|
|
|
|
|
|
def _headline(text: str) -> str:
|
|
"""First line, trimmed, for slug derivation."""
|
|
line = text.strip().splitlines()[0] if text.strip() else "failure"
|
|
return line[:60]
|
|
|
|
|
|
def _slugify(text: str) -> str:
|
|
slug = re.sub(r"[^a-z0-9]+", "-", text.lower()).strip("-")
|
|
slug = re.sub(r"-{2,}", "-", slug)
|
|
if not slug or not SAFE_NAME_RE.match(slug):
|
|
slug = f"fixture-{hashlib.sha256(text.encode()).hexdigest()[:8]}"
|
|
return slug[:80]
|
|
|
|
|
|
def _upsert_manifest(fixtures_root: Path, entry: dict) -> None:
|
|
manifest_file = fixtures_root / "manifest.json"
|
|
manifest = {"version": 1, "fixtures": []}
|
|
if manifest_file.exists():
|
|
manifest = json.loads(manifest_file.read_text(encoding="utf-8"))
|
|
manifest.setdefault("version", 1)
|
|
fixtures = [f for f in manifest.get("fixtures", []) if f.get("id") != entry["id"]]
|
|
fixtures.append(entry)
|
|
manifest["fixtures"] = sorted(fixtures, key=lambda f: (f.get("class", ""), f["id"]))
|
|
manifest_file.write_text(json.dumps(manifest, indent=2, sort_keys=True) + "\n", encoding="utf-8")
|
|
_ensure_generated_test(fixtures_root)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Per-class payload checkers — the generated regression tests dispatch here
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def check_payload(failure_class: str, payload: str) -> None:
|
|
"""Assert the system's CURRENT handling CATCHES the recorded payload.
|
|
|
|
One checker per failure class; a payload that its class's checker fails
|
|
to flag means the class's regression coverage has rotted. Unknown classes
|
|
are a no-op (generic fixture assertions still apply).
|
|
"""
|
|
checker = CLASS_CHECKERS.get(failure_class)
|
|
if checker is not None:
|
|
checker(payload)
|
|
|
|
|
|
def _check_env_case_mismatch(payload: str) -> None:
|
|
"""env-case-mismatch (#192): a workflow boolean env export emitting
|
|
Python-cased True/False while shell steps compare lowercase 'true'/
|
|
'false' — conditionals unreachable. The recorded payload must still be
|
|
flagged by the boolean-export guard, so a recurrence fails CI.
|
|
|
|
Mirrors tests/test_workflow_script_contract.py::test_boolean_env_exports_are_lowercase
|
|
(the #192 discharge, commit b4ddd9d on the release-safety-net branch).
|
|
When that branch merges, import its guard instead of this copy.
|
|
"""
|
|
boolean_keys = ("CONVERGED", "AUTO_BUMPED", "DRIFT")
|
|
lowercase_literal = re.compile(r"""['"](true|false)['"]""")
|
|
json_bool_access = re.compile(r"""json\.load\(|\[\s*['"](?:converged|merged)['"]\s*\]""")
|
|
for key in boolean_keys:
|
|
for line in payload.splitlines():
|
|
capture = re.search(rf"^\s*{key}=\$\((.*)\)\s*$", line)
|
|
if not capture:
|
|
continue
|
|
expr = capture.group(1)
|
|
if ".lower()" in expr or lowercase_literal.search(expr):
|
|
continue
|
|
if re.search(r"\b(True|False)\b", expr) or json_bool_access.search(expr):
|
|
raise AssertionError(
|
|
f"env-case-mismatch recurrence: boolean env export {key} does not force lowercase "
|
|
f"(str(...).lower()) — shell conditionals compare env.{key} == 'false'-style strings: {line.strip()}"
|
|
)
|
|
|
|
|
|
CLASS_CHECKERS: dict[str, object] = {
|
|
"env-case-mismatch": _check_env_case_mismatch,
|
|
}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Generated regression test file
|
|
# ---------------------------------------------------------------------------
|
|
|
|
GENERATED_TEST_NAME = "test_production_fixtures.py"
|
|
|
|
GENERATED_TEST_TEMPLATE = '''"""Generated regression tests over production failure fixtures (#157).
|
|
|
|
WRITTEN BY scripts/capture_failure.py (template embedded there) — do not
|
|
hand-edit the assertions; change the template so generator and generated stay
|
|
in lockstep. Parametrizes over tests/fixtures/production/manifest.json: one
|
|
regression test per fixture asserting the system's CURRENT handling of the
|
|
historical payload (plan 2026-09-07-157):
|
|
|
|
* sha256 tamper-evidence — a payload edited without re-running capture FAILS;
|
|
* sanitiser idempotence — the stored payload is already sanitised and the
|
|
current redactor must leave it unchanged;
|
|
* metadata shape — class/source/date/redactions recorded;
|
|
* per-class checker — the payload must be CAUGHT by the checker for its
|
|
failure class (see scripts/capture_failure.py::CLASS_CHECKERS), so a
|
|
recurrence of the recorded failure fails CI.
|
|
|
|
Payloads are snapshots, not live replays: we assert current-code handling of
|
|
the historical payload, never that old behaviour is preserved.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import importlib.util
|
|
import json
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
ROOT = Path(__file__).resolve().parents[3]
|
|
SCRIPTS_DIR = Path("__GENERATED_SCRIPTS_DIR__")
|
|
FIXTURES_DIR = Path(__file__).resolve().parent
|
|
MANIFEST = FIXTURES_DIR / "manifest.json"
|
|
|
|
|
|
def _load_capture_module():
|
|
# __GENERATED_SCRIPTS_DIR__ is substituted at generation time with the
|
|
# capturing repo's scripts/ directory (absolute). The walk-up below is a
|
|
# secondary fallback for relocated checkouts.
|
|
candidates = [SCRIPTS_DIR]
|
|
candidates.extend(parent / "scripts" for parent in FIXTURES_DIR.parents)
|
|
scripts_dir = next((c for c in candidates if (c / "capture_failure.py").is_file()), SCRIPTS_DIR)
|
|
spec = importlib.util.spec_from_file_location("capture_failure", scripts_dir / "capture_failure.py")
|
|
assert spec is not None and spec.loader is not None
|
|
module = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
def _manifest_entries() -> list[dict]:
|
|
if not MANIFEST.exists():
|
|
return []
|
|
manifest = json.loads(MANIFEST.read_text(encoding="utf-8"))
|
|
return list(manifest.get("fixtures", []))
|
|
|
|
|
|
def assert_fixture_ok(entry: dict, fixtures_dir: Path = FIXTURES_DIR) -> None:
|
|
"""The full per-fixture assertion set (shared by the parametrized tests
|
|
and by the pipeline's own tamper-evidence tests)."""
|
|
fixture_file = fixtures_dir / entry["class"] / entry["id"] / "payload.json"
|
|
assert fixture_file.exists(), f"manifest entry {entry['id']} has no payload fixture at {fixture_file}"
|
|
stored = json.loads(fixture_file.read_text(encoding="utf-8"))
|
|
payload = stored.get("payload")
|
|
assert isinstance(payload, str) and payload.strip(), f"{entry['id']}: payload must be a non-empty string"
|
|
|
|
# 1. Tamper-evidence: the payload's sha256 must match the manifest.
|
|
actual_sha = hashlib.sha256(payload.encode("utf-8")).hexdigest()
|
|
assert actual_sha == entry["sha256"], (
|
|
f"{entry['id']}: payload sha256 mismatch — the fixture was edited without re-running "
|
|
f"scripts/capture_failure.py (expected {entry['sha256']}, found {actual_sha})"
|
|
)
|
|
|
|
# 2. Metadata shape: the manifest entry and the stored metadata agree.
|
|
assert stored.get("failure_class") == entry["class"], f"{entry['id']}: class disagrees with manifest"
|
|
assert stored.get("source") in ("issue", "ci", "manual"), f"{entry['id']}: source must be issue|ci|manual"
|
|
assert stored.get("date"), f"{entry['id']}: capture date missing"
|
|
assert isinstance(stored.get("redactions_applied"), int), f"{entry['id']}: redactions_applied missing"
|
|
|
|
# 3. Sanitiser idempotence: the current redactor must be a no-op on the
|
|
# stored payload (it was sanitised at capture time; re-redacting must
|
|
# not change it — otherwise the redaction surface has drifted).
|
|
cf = _load_capture_module()
|
|
assert cf.redact(payload) == payload, f"{entry['id']}: payload is not idempotent under the current redactor"
|
|
assert not cf.residual_secrets(payload), f"{entry['id']}: residual credential-shaped text in stored payload"
|
|
|
|
# 4. Per-class checker, per the recorded expectation: a "flags" fixture
|
|
# exhibits the defect — the current checker MUST still raise (the
|
|
# raise IS the recurrence alarm); a "passes" fixture must stay clean.
|
|
cf = _load_capture_module()
|
|
expectation = entry.get("checker_expectation", "passes")
|
|
if expectation == "flags":
|
|
try:
|
|
cf.check_payload(entry["class"], payload)
|
|
except AssertionError as exc:
|
|
pass # expected: the checker still catches the recorded defect
|
|
else:
|
|
raise AssertionError(
|
|
f"{entry['id']}: recorded defect is NO LONGER caught by the {entry['class']} checker "
|
|
f"— the class's regression coverage has rotted (checker_expectation=flags)"
|
|
)
|
|
else:
|
|
cf.check_payload(entry["class"], payload)
|
|
|
|
|
|
@pytest.mark.parametrize("entry", _manifest_entries(), ids=lambda e: e["id"])
|
|
def test_production_fixture_handling(entry: dict) -> None:
|
|
assert_fixture_ok(entry)
|
|
|
|
|
|
def test_manifest_registry_is_present_and_nonempty() -> None:
|
|
"""Guards against vacuous parametrization: once the pipeline is seeded,
|
|
the manifest must exist and carry at least one fixture."""
|
|
assert MANIFEST.exists(), (
|
|
"tests/fixtures/production/manifest.json missing — run scripts/capture_failure.py to seed the registry"
|
|
)
|
|
assert _manifest_entries(), "manifest exists but carries no fixtures — run scripts/capture_failure.py"
|
|
'''
|
|
|
|
|
|
def _ensure_generated_test(fixtures_root: Path) -> None:
|
|
"""Write the generated regression test file if absent (idempotent, and
|
|
never clobbers: the file is manifest-driven, so regeneration is churn).
|
|
|
|
The capturing repo's scripts/ directory (this file's location) is baked
|
|
in as an absolute path so the generated assertions can always import the
|
|
checkers, wherever the fixtures root lives."""
|
|
generated = fixtures_root / GENERATED_TEST_NAME
|
|
if generated.exists():
|
|
return
|
|
body = GENERATED_TEST_TEMPLATE.replace("__GENERATED_SCRIPTS_DIR__", str(Path(__file__).resolve().parent))
|
|
generated.write_text(body, encoding="utf-8")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# CLI
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def main(argv: list[str] | None = None) -> int:
|
|
parser = argparse.ArgumentParser(
|
|
prog="capture_failure.py",
|
|
description="Capture a production failure as a sanitised regression fixture (#157).",
|
|
)
|
|
parser.add_argument("payload", help="path to the raw payload file (error text or JSON blob)")
|
|
parser.add_argument(
|
|
"--class", dest="failure_class", required=True, help="failure class slug, e.g. env-case-mismatch"
|
|
)
|
|
parser.add_argument("--source", required=True, choices=VALID_SOURCES, help="where the failure came from")
|
|
parser.add_argument(
|
|
"--id", dest="fixture_id", default=None, help="fixture slug (default: derived from the payload)"
|
|
)
|
|
parser.add_argument("--fixtures-root", default=DEFAULT_FIXTURES_ROOT, help="fixture root (default: %(default)s)")
|
|
args = parser.parse_args(argv)
|
|
|
|
payload_file = Path(args.payload)
|
|
if not payload_file.is_file():
|
|
print(f"capture_failure: payload file not found: {payload_file}", file=sys.stderr)
|
|
return EXIT_USAGE
|
|
try:
|
|
entry = capture(
|
|
payload_file,
|
|
failure_class=args.failure_class,
|
|
source=args.source,
|
|
fixture_id=args.fixture_id,
|
|
fixtures_root=args.fixtures_root,
|
|
)
|
|
except ResidualSecretError as exc:
|
|
print(f"capture_failure: REFUSED — {exc}", file=sys.stderr)
|
|
return EXIT_RESIDUAL_SECRET
|
|
except ValueError as exc:
|
|
print(f"capture_failure: {exc}", file=sys.stderr)
|
|
return EXIT_USAGE
|
|
|
|
print(f"captured {entry['id']} -> {Path(args.fixtures_root) / entry['class'] / entry['id'] / 'payload.json'}")
|
|
return EXIT_OK
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|