X-OP-02 Run record check

Field Value
Purpose Check a run record JSON file’s own required core, its breakdowns[] scale values, and whether a declared category is missing an explicit zero row.
Usage python3 scripts/op/run_record_check.py –help In a shell: python3 scripts/op/run_record_check.py RUN.json
Dependencies stdlib
Writes files no
License CC0-1.0
Inputs A run record JSON file: a top-level JSON object with schema_version, run and steps. A steps entry is a JSON object; its own fields are not otherwise checked here. An optional top-level breakdowns list holds objects with an id, a scale, an optional categories list, and an optional counts object mapping a category name to a count.
Outputs One line per finding, then a summary line, all printed to standard output.
Used in Run logging and dashboards
Source scripts/op/run_record_check.py

Source code

"""
ID: X-OP-02
Title: Run record check
Stage: OP
Purpose: Check a run record JSON file's own required core, its
    breakdowns[] scale values, and whether a declared category is
    missing an explicit zero row.
Usage: python3 scripts/op/run_record_check.py --help
    In a shell: python3 scripts/op/run_record_check.py RUN.json
Dependencies: stdlib
Writes files: no
License: CC0-1.0
Inputs: A run record JSON file: a top-level JSON object with
    schema_version, run and steps. A steps entry is a JSON object; its
    own fields are not otherwise checked here. An optional top-level
    breakdowns list holds objects with an id, a scale, an optional
    categories list, and an optional counts object mapping a category
    name to a count.
Outputs: One line per finding, then a summary line, all printed to
    standard output.

This script checks the record's own required core and its
breakdowns[] entries only. It does not check steps[] content, gate
decisions, metrics[], checks[], or budget; a run record can pass every
check here and still be wrong in a field this script does not look at.

Errors (exit 1): the record has no schema_version, no run, or no
steps; run is present but is not a JSON object; steps is present but
is not a JSON list; breakdowns is present but is not a JSON list; a
breakdowns[] entry is not a JSON object; a breakdowns[] entry's scale
is missing or is not one of nominal, ordinal or status.

Warnings (exit 0, never change the exit code): a breakdowns[] entry
declares a category name in its own categories list that has no
matching row in its own counts object. A category is "declared" by
appearing in that same breakdown's categories list; it "appears in its
own breakdown" by being a key in that same breakdown's counts object,
with any numeric value, including zero. A category present in counts
but absent from categories is not a finding here: categories is read
as the full list a breakdown promises to always show, not a strict
list of every key counts may hold.
"""

import argparse
import json
import sys
from pathlib import Path
from typing import Any

sys.dont_write_bytecode = True
if sys.version_info < (3, 10):
    print(
        "run_record_check.py: this script needs Python 3.10 or newer, but "
        f"this is {sys.version_info.major}.{sys.version_info.minor}. "
        "Run it with a newer python3.",
        file=sys.stderr,
    )
    sys.exit(2)

MAX_BYTES = 5_000_000
REQUIRED_CORE = ("schema_version", "run", "steps")
ALLOWED_SCALES = ("nominal", "ordinal", "status")


def load_record(path: Path) -> Any:
    """Read and parse the run record file; never follows a symlink."""
    if path.is_symlink():
        raise OSError(f"refusing to read a symlink: {path}")
    size = path.stat().st_size
    if size > MAX_BYTES:
        raise ValueError(f"{path} is over {MAX_BYTES} bytes; skipping")
    text = path.read_text(encoding="utf-8", errors="replace")
    return json.loads(text)


def check_record(data: Any) -> tuple[list[str], list[str], int]:
    """Return (errors, warnings, step_count)."""
    if not isinstance(data, dict):
        raise ValueError("the run record is not a JSON object")

    errors: list[str] = []
    warnings: list[str] = []

    for field_name in REQUIRED_CORE:
        if field_name not in data:
            errors.append(f"missing-core: the run record has no '{field_name}'")

    run = data.get("run")
    if "run" in data and not isinstance(run, dict):
        errors.append("bad-run: 'run' is present but is not a JSON object")

    steps = data.get("steps")
    if "steps" in data and not isinstance(steps, list):
        errors.append("bad-steps: 'steps' is present but is not a JSON list")
    step_count = len(steps) if isinstance(steps, list) else 0

    breakdowns = data.get("breakdowns")
    if breakdowns is not None and not isinstance(breakdowns, list):
        errors.append("bad-breakdowns: 'breakdowns' is present but is not a JSON list")
        breakdowns = []
    if breakdowns is None:
        breakdowns = []

    for position, entry in enumerate(breakdowns, start=1):
        label = f"breakdown #{position}"
        if not isinstance(entry, dict):
            errors.append(f"bad-breakdown: {label} is not a JSON object")
            continue
        entry_id = entry.get("id")
        if isinstance(entry_id, str) and entry_id:
            label = f"breakdown '{entry_id}'"

        scale = entry.get("scale")
        if scale not in ALLOWED_SCALES:
            shown = ", ".join(ALLOWED_SCALES)
            errors.append(
                f"bad-scale: {label} has scale {scale!r}, must be one of {shown}"
            )

        categories = entry.get("categories")
        counts = entry.get("counts")
        if isinstance(categories, list) and isinstance(counts, dict):
            for category in categories:
                if isinstance(category, str) and category not in counts:
                    warnings.append(
                        f"omitted-zero: {label} declares category {category!r} "
                        "but 'counts' has no row for it; write an explicit zero "
                        "row instead of leaving it out"
                    )

    return errors, warnings, step_count


def build_parser() -> argparse.ArgumentParser:
    """Build the argument parser."""
    parser = argparse.ArgumentParser(
        prog="run_record_check.py",
        description=(
            "Check a run record's required core, its breakdowns[] scale "
            "values, and whether a declared category is missing an "
            "explicit zero row. Writes no files."
        ),
    )
    parser.add_argument("run_record", metavar="RUN", help="path to the run record JSON")
    return parser


def main(argv: list[str] | None = None) -> int:
    """Command line entry point; prints the report, returns an exit code."""
    parser = build_parser()
    args = parser.parse_args(argv)
    try:
        data = load_record(Path(args.run_record))
        errors, warnings, step_count = check_record(data)
    except (
        OSError,
        UnicodeError,
        ValueError,
        json.JSONDecodeError,
        RecursionError,
    ) as exc:
        print(f"run_record_check.py: error: {exc}", file=sys.stderr)
        return 2
    for message in errors:
        print(f"error {message}")
    for message in warnings:
        print(f"warning {message}")
    print(f"steps={step_count} errors={len(errors)} warnings={len(warnings)}")
    return 1 if errors else 0


if __name__ == "__main__":
    sys.exit(main())

To the extent possible under law, copyright and related rights in this work are waived under CC0 1.0 Universal.

This site uses Just the Docs, a documentation theme for Jekyll.