scripts/validate_config_layout.py

#!/usr/bin/env python3
"""Validate the purpose-based configuration layout and portable config paths."""

from __future__ import annotations

import argparse
import hashlib
import json
from collections import Counter
from datetime import datetime, timezone
from pathlib import Path
from typing import Any

import jsonschema


REPO_ROOT = Path(__file__).resolve().parents[1]
CONFIG_ROOT = REPO_ROOT / "configs"
EXPECTED_CATEGORY_COUNTS = {
    "evaluation": 14,
    "examples": 2,
    "fixtures": 1,
    "mlir": 24,
    "pipelines": 21,
}
def utc_now() -> str:
    return datetime.now(timezone.utc).isoformat()


def sha256(path: Path) -> str:
    digest = hashlib.sha256()
    with path.open("rb") as handle:
        for block in iter(lambda: handle.read(1024 * 1024), b""):
            digest.update(block)
    return digest.hexdigest()


def load_json(path: Path) -> Any:
    return json.loads(path.read_text(encoding="utf-8"))


def model_config_paths(root: Path) -> list[Path]:
    paths: list[Path] = []
    for relative in ("examples", "pipelines"):
        paths.extend((root / "configs" / relative).rglob("*.json"))
    return sorted(paths)


def audit_layout(root: Path = REPO_ROOT) -> dict[str, Any]:
    config_root = root / "configs"
    checks: list[dict[str, Any]] = []
    errors: list[dict[str, Any]] = []

    def check(name: str, passed: bool, **details: Any) -> None:
        record = {"check": name, "passed": bool(passed), **details}
        checks.append(record)
        if not passed:
            errors.append(record)

    json_paths = sorted(config_root.rglob("*.json"))
    categories = Counter(path.relative_to(config_root).parts[0] for path in json_paths)
    root_directories = {path.name for path in config_root.iterdir() if path.is_dir()}
    check(
        "canonical_root_categories",
        root_directories == set(EXPECTED_CATEGORY_COUNTS),
        actual=sorted(root_directories),
        expected=sorted(EXPECTED_CATEGORY_COUNTS),
    )
    check(
        "category_file_counts",
        dict(categories) == EXPECTED_CATEGORY_COUNTS,
        actual=dict(sorted(categories.items())),
        expected=EXPECTED_CATEGORY_COUNTS,
    )
    direct_json = sorted(path.name for path in config_root.glob("*.json"))
    check("no_root_level_json", not direct_json, found=direct_json)
    check(
        "model_schema_location",
        (root / "schemas" / "model_config.schema.json").is_file(),
        path="schemas/model_config.schema.json",
    )

    parsed: dict[Path, Any] = {}
    parse_errors: list[str] = []
    for path in json_paths:
        try:
            parsed[path] = load_json(path)
        except (OSError, json.JSONDecodeError) as error:
            parse_errors.append(f"{path.relative_to(root)}: {error}")
    check("all_config_json_parse", not parse_errors, errors=parse_errors)

    model_schema = load_json(root / "schemas" / "model_config.schema.json")
    model_errors: list[str] = []
    for path in model_config_paths(root):
        document = parsed.get(path)
        if document is None:
            continue
        for error in jsonschema.Draft202012Validator(model_schema).iter_errors(document):
            model_errors.append(f"{path.relative_to(root)}: {error.message}")
    check(
        "model_pipeline_schema",
        not model_errors,
        config_count=len(model_config_paths(root)),
        errors=model_errors,
    )

    mlir_schema = load_json(root / "schemas" / "mlir_batch_config.schema.json")
    mlir_paths = sorted((config_root / "mlir" / "batch").glob("*_mlir.json"))
    mlir_errors: list[str] = []
    absolute_mlir_paths: list[str] = []
    for path in mlir_paths:
        document = parsed.get(path)
        if document is None:
            continue
        for error in jsonschema.Draft202012Validator(mlir_schema).iter_errors(document):
            mlir_errors.append(f"{path.relative_to(root)}: {error.message}")
        candidate_values = [
            document["model_dir"],
            document["toolchain_lock"],
            *(variant["input_path"] for variant in document["variants"].values()),
        ]
        absolute_mlir_paths.extend(
            f"{path.relative_to(root)}: {value}"
            for value in candidate_values
            if Path(value).is_absolute()
        )
    check("mlir_batch_schema", not mlir_errors, config_count=len(mlir_paths), errors=mlir_errors)
    check(
        "mlir_paths_repository_relative",
        not absolute_mlir_paths,
        errors=absolute_mlir_paths,
    )

    inventory = [
        {
            "path": str(path.relative_to(root)),
            "category": path.relative_to(config_root).parts[0],
            "bytes": path.stat().st_size,
            "sha256": sha256(path),
        }
        for path in json_paths
    ]
    return {
        "schema_version": "1.0",
        "generated_at": utc_now(),
        "status": "PASS" if not errors else "FAIL",
        "repository_root": str(root),
        "config_count": len(json_paths),
        "category_counts": dict(sorted(categories.items())),
        "check_count": len(checks),
        "error_count": len(errors),
        "checks": checks,
        "errors": errors,
        "inventory": inventory,
    }


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument(
        "--output",
        type=Path,
        default=REPO_ROOT / "reports" / "config_layout_validation.json",
    )
    args = parser.parse_args()
    result = audit_layout()
    output = args.output.resolve()
    output.parent.mkdir(parents=True, exist_ok=True)
    output.write_text(
        json.dumps(result, indent=2, sort_keys=True, ensure_ascii=False) + "\n",
        encoding="utf-8",
    )
    print(
        json.dumps(
            {
                "status": result["status"],
                "configs": result["config_count"],
                "checks": result["check_count"],
                "errors": result["error_count"],
                "output": str(output),
            },
            sort_keys=True,
        )
    )
    return 0 if result["status"] == "PASS" else 1


if __name__ == "__main__":
    raise SystemExit(main())