239 lines
12 KiB
Python
239 lines
12 KiB
Python
#!/usr/bin/env python3
|
|
"""Validate a 1C saved-state object report contract."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
|
|
EXPECTED_SCHEMA = "onec_saved_state_object_report.v1"
|
|
EXPECTED_COMPARISON_SCHEMA = "onec_saved_state_object_comparison.v1"
|
|
EXPECTED_DETAIL_SCHEMA = "onec_saved_state_object_detail.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(report_path: Path, value: Any) -> Path | None:
|
|
if not value:
|
|
return None
|
|
path = Path(str(value))
|
|
if path.is_absolute():
|
|
return path
|
|
return (report_path.parent / path).resolve()
|
|
|
|
|
|
def is_list(value: Any) -> bool:
|
|
return isinstance(value, list)
|
|
|
|
|
|
def check_agent_summary(report: dict[str, Any], findings: list[dict[str, Any]]) -> None:
|
|
summary = report.get("agent_summary")
|
|
if not isinstance(summary, dict):
|
|
findings.append(issue("error", "missing_agent_summary", "Report must include agent_summary."))
|
|
return
|
|
|
|
object_changes = summary.get("object_changes")
|
|
if not isinstance(object_changes, list):
|
|
findings.append(issue("error", "invalid_agent_summary_object_changes", "agent_summary.object_changes must be an array."))
|
|
object_changes = []
|
|
object_names = summary.get("object_names")
|
|
if not isinstance(object_names, list):
|
|
findings.append(issue("error", "invalid_agent_summary_object_names", "agent_summary.object_names must be an array."))
|
|
object_names = []
|
|
|
|
names_from_objects = [item.get("full_name") for item in object_changes if isinstance(item, dict)]
|
|
if names_from_objects != object_names:
|
|
findings.append(issue(
|
|
"error",
|
|
"agent_summary_names_mismatch",
|
|
"agent_summary.object_names must match object_changes full_name order.",
|
|
detail={"object_names": object_names, "from_objects": names_from_objects},
|
|
))
|
|
|
|
for index, item in enumerate(object_changes):
|
|
if not isinstance(item, dict):
|
|
findings.append(issue("error", "invalid_agent_summary_object", "agent_summary object item must be an object.", detail={"index": index}))
|
|
continue
|
|
full_name = item.get("full_name")
|
|
if not full_name:
|
|
findings.append(issue("error", "missing_agent_summary_full_name", "agent_summary object item is missing full_name.", detail={"index": index}))
|
|
for key in ("added_terms", "removed_terms", "parts"):
|
|
if not is_list(item.get(key)):
|
|
findings.append(issue("error", f"invalid_{key}", f"agent_summary object field {key} must be an array.", detail={"object": full_name, "type": type(item.get(key)).__name__}))
|
|
parts = item.get("parts") if isinstance(item.get("parts"), list) else []
|
|
if item.get("parts_count") != len(parts):
|
|
findings.append(issue("error", "agent_summary_parts_count_mismatch", "parts_count must match parts length.", detail={"object": full_name, "parts_count": item.get("parts_count"), "actual": len(parts)}))
|
|
text_diff_parts = 0
|
|
active_missing_parts = 0
|
|
for part in parts:
|
|
if not isinstance(part, dict):
|
|
continue
|
|
role = part.get("payload_role")
|
|
if role not in KNOWN_PAYLOAD_ROLES:
|
|
findings.append(issue("error", "unknown_payload_role", "Unknown payload_role in agent_summary part.", detail={"object": full_name, "role": role, "file_name": part.get("file_name")}))
|
|
if part.get("summary") in {"Text payload differs.", "Text payload matches."}:
|
|
text_diff_parts += 1
|
|
if part.get("active_exists") is False:
|
|
active_missing_parts += 1
|
|
if item.get("text_diff_parts") != text_diff_parts:
|
|
findings.append(issue("warning", "agent_summary_text_diff_count_mismatch", "text_diff_parts differs from counted comparable parts.", detail={"object": full_name, "reported": item.get("text_diff_parts"), "counted": text_diff_parts}))
|
|
if item.get("active_missing_parts") != active_missing_parts:
|
|
findings.append(issue("error", "agent_summary_active_missing_count_mismatch", "active_missing_parts must match parts with active_exists=false.", detail={"object": full_name, "reported": item.get("active_missing_parts"), "counted": active_missing_parts}))
|
|
|
|
system_changes = summary.get("system_changes")
|
|
if not isinstance(system_changes, list):
|
|
findings.append(issue("error", "invalid_agent_summary_system_changes", "agent_summary.system_changes must be an array."))
|
|
system_changes = []
|
|
system_names = summary.get("system_change_names")
|
|
if not isinstance(system_names, list):
|
|
findings.append(issue("error", "invalid_agent_summary_system_names", "agent_summary.system_change_names must be an array."))
|
|
system_names = []
|
|
names_from_system = [item.get("name") for item in system_changes if isinstance(item, dict)]
|
|
if names_from_system != system_names:
|
|
findings.append(issue("error", "agent_summary_system_names_mismatch", "system_change_names must match system_changes name order.", detail={"system_change_names": system_names, "from_system": names_from_system}))
|
|
|
|
|
|
def check_detail(detail: dict[str, Any], findings: list[dict[str, Any]]) -> None:
|
|
if detail.get("schema") != EXPECTED_DETAIL_SCHEMA:
|
|
findings.append(issue("error", "invalid_detail_schema", "Detail schema is invalid.", detail={"schema": detail.get("schema")}))
|
|
for obj in detail.get("object_details") or []:
|
|
if not isinstance(obj, dict):
|
|
continue
|
|
full_name = obj.get("full_name")
|
|
for part in obj.get("details") or []:
|
|
if not isinstance(part, dict):
|
|
continue
|
|
role = part.get("payload_role")
|
|
if role not in KNOWN_PAYLOAD_ROLES:
|
|
findings.append(issue("error", "detail_unknown_payload_role", "Unknown payload_role in detail part.", detail={"object": full_name, "role": role, "file_name": part.get("file_name")}))
|
|
payload = part.get("payload") or {}
|
|
hints = payload.get("semantic_hints")
|
|
if hints is not None:
|
|
for key in ("added_terms", "removed_terms"):
|
|
if not isinstance(hints.get(key), list):
|
|
findings.append(issue("error", "invalid_semantic_hints", f"semantic_hints.{key} must be an array.", detail={"object": full_name, "file_name": part.get("file_name")}))
|
|
|
|
|
|
def check_report(report_path: Path) -> dict[str, Any]:
|
|
findings: list[dict[str, Any]] = []
|
|
if not report_path.exists():
|
|
findings.append(issue("error", "missing_report", "Saved-state report file is missing.", path=report_path))
|
|
return build_result(report_path, findings, None, None)
|
|
|
|
report = load_json(report_path)
|
|
if report.get("schema") != EXPECTED_SCHEMA:
|
|
findings.append(issue("error", "invalid_report_schema", "Report schema is invalid.", path=report_path, detail={"schema": report.get("schema")}))
|
|
|
|
safety = report.get("safety") or {}
|
|
if safety.get("read_only") is not True:
|
|
findings.append(issue("error", "report_not_read_only", "Report safety.read_only must be true."))
|
|
if safety.get("sql_write_performed") is not False:
|
|
findings.append(issue("error", "report_sql_write_flag", "Report safety.sql_write_performed must be false."))
|
|
if safety.get("public_terms_are_1c_objects") is not True:
|
|
findings.append(issue("error", "report_public_terms_flag", "Report must expose public terms as 1C objects."))
|
|
if safety.get("secrets_in_report") is not False:
|
|
findings.append(issue("error", "report_secrets_flag", "Report safety.secrets_in_report must be false."))
|
|
|
|
comparison_path = linked_path(report_path, report.get("comparison"))
|
|
detail_path = linked_path(report_path, report.get("detail"))
|
|
markdown_path = linked_path(report_path, report.get("markdown"))
|
|
comparison: dict[str, Any] | None = None
|
|
detail: dict[str, Any] | None = None
|
|
|
|
if comparison_path is None or not comparison_path.exists():
|
|
findings.append(issue("error", "missing_comparison", "Linked comparison JSON is missing.", path=comparison_path or "<null>"))
|
|
else:
|
|
comparison = load_json(comparison_path)
|
|
if comparison.get("schema") != EXPECTED_COMPARISON_SCHEMA:
|
|
findings.append(issue("error", "invalid_comparison_schema", "Comparison schema is invalid.", path=comparison_path, detail={"schema": comparison.get("schema")}))
|
|
|
|
if detail_path is None or not detail_path.exists():
|
|
findings.append(issue("error", "missing_detail", "Linked detail JSON is missing.", path=detail_path or "<null>"))
|
|
else:
|
|
detail = load_json(detail_path)
|
|
check_detail(detail, findings)
|
|
|
|
if report.get("markdown") is not None and (markdown_path is None or not markdown_path.exists()):
|
|
findings.append(issue("error", "missing_markdown", "Linked Markdown report is missing.", path=markdown_path or "<null>"))
|
|
|
|
counts = report.get("counts") or {}
|
|
if comparison:
|
|
comparison_counts = comparison.get("counts") or {}
|
|
if counts.get("object_changes") != comparison_counts.get("object_changes"):
|
|
findings.append(issue("error", "object_change_count_mismatch", "Report object_changes count differs from comparison."))
|
|
if counts.get("system_changes") != comparison_counts.get("system_changes"):
|
|
findings.append(issue("error", "system_change_count_mismatch", "Report system_changes count differs from comparison."))
|
|
if detail:
|
|
detail_counts = detail.get("counts") or {}
|
|
if counts.get("detail_objects") != detail_counts.get("objects"):
|
|
findings.append(issue("error", "detail_object_count_mismatch", "Report detail_objects count differs from detail."))
|
|
if counts.get("detail_parts") != detail_counts.get("details"):
|
|
findings.append(issue("error", "detail_part_count_mismatch", "Report detail_parts count differs from detail."))
|
|
|
|
check_agent_summary(report, findings)
|
|
return build_result(report_path, findings, comparison, detail)
|
|
|
|
|
|
def build_result(report_path: Path, findings: list[dict[str, Any]], comparison: dict[str, Any] | None, detail: dict[str, Any] | None) -> 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_check.v1",
|
|
"report": str(report_path),
|
|
"comparison_schema": (comparison or {}).get("schema"),
|
|
"detail_schema": (detail or {}).get("schema"),
|
|
"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 object report.")
|
|
parser.add_argument("--report", type=Path, required=True)
|
|
parser.add_argument("--output", type=Path)
|
|
args = parser.parse_args()
|
|
|
|
result = check_report(args.report)
|
|
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())
|