DataOps-platform/scripts/validate_domain_replication.py

#!/usr/bin/env python3
"""Validate a self-contained domain replication implementation package."""

from __future__ import annotations

import argparse
import csv
import hashlib
import json
import re
import sys
from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]
if str(ROOT) not in sys.path:
    sys.path.insert(0, str(ROOT))

from app.core.governance.domain_replication import (  # noqa: E402
    DomainReplicationError,
    evaluate_replication_package,
)

_SHA256 = re.compile(r"^[0-9a-f]{64}$")


def _sha256(path: Path) -> str:
    return hashlib.sha256(path.read_bytes()).hexdigest()


def _json(path: Path) -> dict:
    value = json.loads(path.read_text(encoding="utf-8"))
    if not isinstance(value, dict):
        raise DomainReplicationError(f"JSON object required: {path}")
    return value


def _resolve(package_dir: Path, relative: str) -> Path:
    path = (package_dir / relative).resolve()
    if package_dir not in path.parents:
        raise DomainReplicationError(f"package path escapes package directory: {relative}")
    if not path.is_file():
        raise DomainReplicationError(f"package file does not exist: {relative}")
    return path


def _rows(path: Path) -> list[dict[str, str]]:
    with path.open(encoding="utf-8", newline="") as handle:
        return list(csv.DictReader(handle))


def validate_package(package_dir: Path) -> dict:
    package_dir = package_dir.resolve()
    manifest = _json(package_dir / "manifest.json")
    files = manifest.get("files")
    if not isinstance(files, dict):
        raise DomainReplicationError("manifest.files must be an object")
    resolved: dict[str, Path] = {}
    required_files = ("template", "evidence", "snapshot", "delta")
    optional_files = ("acceptance_report",)
    for key in (*required_files, *optional_files):
        binding = files.get(key)
        if key in optional_files and binding is None:
            continue
        if not isinstance(binding, dict):
            raise DomainReplicationError(f"manifest.files.{key} is required")
        path = _resolve(package_dir, str(binding.get("path", "")))
        expected = str(binding.get("sha256", ""))
        if not _SHA256.fullmatch(expected):
            raise DomainReplicationError(f"manifest.files.{key}.sha256 is invalid")
        actual = _sha256(path)
        if actual != expected:
            raise DomainReplicationError(
                f"package file digest mismatch for {key}: expected={expected}, actual={actual}"
            )
        resolved[key] = path
    report = evaluate_replication_package(
        manifest,
        _json(resolved["evidence"]),
        template=_json(resolved["template"]),
        snapshot_rows=_rows(resolved["snapshot"]),
        delta_rows=_rows(resolved["delta"]),
    )
    if "acceptance_report" in resolved:
        rendered = json.dumps(report, ensure_ascii=False, sort_keys=True, indent=2) + "\n"
        if resolved["acceptance_report"].read_text(encoding="utf-8") != rendered:
            raise DomainReplicationError(
                "acceptance report does not match the deterministic package result"
            )
    return report


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--package-dir", type=Path, required=True)
    parser.add_argument("--output", default="-")
    args = parser.parse_args(argv)
    report = validate_package(args.package_dir)
    rendered = json.dumps(report, ensure_ascii=False, sort_keys=True, indent=2) + "\n"
    if args.output == "-":
        sys.stdout.write(rendered)
    else:
        output = Path(args.output)
        output.parent.mkdir(parents=True, exist_ok=True)
        output.write_text(rendered, encoding="utf-8")
    return 0


if __name__ == "__main__":
    try:
        raise SystemExit(main())
    except (DomainReplicationError, OSError, ValueError, json.JSONDecodeError) as exc:
        print(f"ERROR: {exc}", file=sys.stderr)
        raise SystemExit(2) from exc