#!/usr/bin/env python3 """Validate a 1C saved-state report delta contract.""" from __future__ import annotations import argparse import json from pathlib import Path from typing import Any EXPECTED_SCHEMA = "onec_saved_state_object_report_delta.v1" KNOWN_PAYLOAD_ROLES = { "bsl_module_text", "form_descriptor", "form_body", "primary_payload", "metadata_payload", } def load_json(path: Path) -> dict[str, Any]: return json.loads(path.read_text(encoding="utf-8-sig")) def write_json(path: Path, data: dict[str, Any]) -> None: path.parent.mkdir(parents=True, exist_ok=True) path.write_text(json.dumps(data, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") def issue(severity: str, code: str, message: str, *, path: Path | str | None = None, detail: dict[str, Any] | None = None) -> dict[str, Any]: result: dict[str, Any] = {"severity": severity, "code": code, "message": message} if path is not None: result["path"] = str(path) if detail: result["detail"] = detail return result def linked_path(delta_path: Path, value: Any) -> Path | None: if not value: return None path = Path(str(value)) if path.is_absolute(): return path return (delta_path.parent / path).resolve() def check_payload_roles(parts: list[Any], findings: list[dict[str, Any]], *, context: str) -> None: for part in parts: if not isinstance(part, dict): findings.append(issue("error", "invalid_part", "Payload part must be an object.", detail={"context": context})) continue role = part.get("payload_role") if role not in KNOWN_PAYLOAD_ROLES: findings.append(issue("error", "unknown_payload_role", "Unknown payload_role.", detail={"context": context, "role": role, "file_name": part.get("file_name")})) def check_compact_object(item: dict[str, Any], findings: list[dict[str, Any]], *, context: str) -> None: if not item.get("full_name"): findings.append(issue("error", "missing_full_name", "Object delta item is missing full_name.", detail={"context": context})) for key in ("added_terms", "removed_terms", "parts"): if not isinstance(item.get(key), list): findings.append(issue("error", f"invalid_{key}", f"Object field {key} must be an array.", detail={"context": context, "full_name": item.get("full_name")})) parts = item.get("parts") if isinstance(item.get("parts"), list) else [] if item.get("parts_count") != len(parts): findings.append(issue("error", "parts_count_mismatch", "parts_count must match parts length.", detail={"context": context, "full_name": item.get("full_name"), "parts_count": item.get("parts_count"), "actual": len(parts)})) check_payload_roles(parts, findings, context=context) def check_changed_object(item: dict[str, Any], findings: list[dict[str, Any]]) -> None: full_name = item.get("full_name") before = item.get("before") after = item.get("after") if not isinstance(before, dict) or not isinstance(after, dict): findings.append(issue("error", "invalid_changed_object_shape", "Changed object must include before and after objects.", detail={"full_name": full_name})) return check_compact_object(before, findings, context=f"changed.before:{full_name}") check_compact_object(after, findings, context=f"changed.after:{full_name}") for key in ("before_fingerprint", "after_fingerprint"): if not item.get(key): findings.append(issue("error", f"missing_{key}", f"Changed object is missing {key}.", detail={"full_name": full_name})) if item.get("before_fingerprint") == item.get("after_fingerprint"): findings.append(issue("error", "unchanged_fingerprint_in_changed", "Changed object has equal before and after fingerprints.", detail={"full_name": full_name})) term_delta = item.get("term_delta") if not isinstance(term_delta, dict): findings.append(issue("error", "invalid_term_delta", "Changed object term_delta must be an object.", detail={"full_name": full_name})) else: for list_name in ("added_terms", "removed_terms"): delta = term_delta.get(list_name) if not isinstance(delta, dict) or not isinstance(delta.get("added"), list) or not isinstance(delta.get("removed"), list): findings.append(issue("error", "invalid_term_delta_list", "Term delta must contain added and removed arrays.", detail={"full_name": full_name, "list": list_name})) part_delta = item.get("part_delta") if not isinstance(part_delta, dict): findings.append(issue("error", "invalid_part_delta", "Changed object part_delta must be an object.", detail={"full_name": full_name})) else: for key in ("added", "removed", "changed"): if not isinstance(part_delta.get(key), list): findings.append(issue("error", f"invalid_part_delta_{key}", f"part_delta.{key} must be an array.", detail={"full_name": full_name})) check_payload_roles(part_delta.get("added") or [], findings, context=f"part_delta.added:{full_name}") check_payload_roles(part_delta.get("removed") or [], findings, context=f"part_delta.removed:{full_name}") for part in part_delta.get("changed") or []: if not isinstance(part, dict): findings.append(issue("error", "invalid_changed_part", "part_delta.changed item must be an object.", detail={"full_name": full_name})) continue before_part = part.get("before") after_part = part.get("after") if not isinstance(before_part, dict) or not isinstance(after_part, dict): findings.append(issue("error", "invalid_changed_part_shape", "Changed part must include before and after.", detail={"full_name": full_name, "file_name": part.get("file_name")})) continue check_payload_roles([before_part, after_part], findings, context=f"part_delta.changed:{full_name}") def check_delta(delta_path: Path) -> dict[str, Any]: findings: list[dict[str, Any]] = [] if not delta_path.exists(): findings.append(issue("error", "missing_delta", "Delta report file is missing.", path=delta_path)) return build_result(delta_path, findings) data = load_json(delta_path) if data.get("schema") != EXPECTED_SCHEMA: findings.append(issue("error", "invalid_delta_schema", "Delta schema is invalid.", path=delta_path, detail={"schema": data.get("schema")})) safety = data.get("safety") or {} if safety.get("read_only") is not True: findings.append(issue("error", "delta_not_read_only", "Delta safety.read_only must be true.")) if safety.get("sql_write_performed") is not False: findings.append(issue("error", "delta_sql_write_flag", "Delta safety.sql_write_performed must be false.")) if safety.get("public_terms_are_1c_objects") is not True: findings.append(issue("error", "delta_public_terms_flag", "Delta must expose public terms as 1C objects.")) for key in ("before_report", "after_report"): path = linked_path(delta_path, data.get(key)) if path is None or not path.exists(): findings.append(issue("warning", f"missing_{key}", f"Linked {key} is missing.", path=path or "")) markdown_path = linked_path(delta_path, data.get("markdown")) if data.get("markdown") is not None and (markdown_path is None or not markdown_path.exists()): findings.append(issue("error", "missing_markdown", "Linked Markdown delta report is missing.", path=markdown_path or "")) objects = data.get("objects") if not isinstance(objects, dict): findings.append(issue("error", "invalid_objects", "Delta objects must be an object.")) objects = {} for key in ("added", "removed", "changed", "unchanged"): if not isinstance(objects.get(key), list): findings.append(issue("error", f"invalid_objects_{key}", f"objects.{key} must be an array.")) for item in objects.get("added") or []: if isinstance(item, dict): check_compact_object(item, findings, context="objects.added") for item in objects.get("removed") or []: if isinstance(item, dict): check_compact_object(item, findings, context="objects.removed") for item in objects.get("changed") or []: if isinstance(item, dict): check_changed_object(item, findings) system = data.get("system_changes") if not isinstance(system, dict): findings.append(issue("error", "invalid_system_changes", "Delta system_changes must be an object.")) system = {} for key in ("added", "removed", "changed"): if not isinstance(system.get(key), list): findings.append(issue("error", f"invalid_system_changes_{key}", f"system_changes.{key} must be an array.")) counts = data.get("counts") or {} expected_counts = { "objects_added": len(objects.get("added") or []), "objects_removed": len(objects.get("removed") or []), "objects_changed": len(objects.get("changed") or []), "objects_unchanged": len(objects.get("unchanged") or []), "system_added": len(system.get("added") or []), "system_removed": len(system.get("removed") or []), "system_changed": len(system.get("changed") or []), } for key, expected in expected_counts.items(): if counts.get(key) != expected: findings.append(issue("error", "delta_count_mismatch", "Delta count does not match payload.", detail={"key": key, "reported": counts.get(key), "expected": expected})) return build_result(delta_path, findings) def build_result(delta_path: Path, findings: list[dict[str, Any]]) -> dict[str, Any]: errors = [row for row in findings if row.get("severity") == "error"] warnings = [row for row in findings if row.get("severity") == "warning"] return { "schema": "onec_saved_state_object_report_delta_check.v1", "delta": str(delta_path), "passed": not errors, "findings": findings, "counts": { "errors": len(errors), "warnings": len(warnings), }, "safety": { "read_only": True, "sql_write_performed": False, }, } def main() -> int: parser = argparse.ArgumentParser(description="Validate a 1C saved-state report delta.") parser.add_argument("--delta", type=Path, required=True) parser.add_argument("--output", type=Path) args = parser.parse_args() result = check_delta(args.delta) if args.output: write_json(args.output, result) print(json.dumps({"output": str(args.output) if args.output else None, "passed": result["passed"], "counts": result["counts"]}, ensure_ascii=False)) return 0 if result["passed"] else 2 if __name__ == "__main__": raise SystemExit(main())