#!/usr/bin/env python3 from __future__ import annotations import argparse import json import subprocess import sys from pathlib import Path from typing import Any ROOT = Path(__file__).resolve().parents[1] DEFAULT_REST_ADAPTER_URL = "http://docker.cin.su:8011" DEFAULT_MCP_URL = "http://docker.cin.su:8021" SAVED_STATE_TABLES = ("ConfigSave", "ConfigCASSave") def duplicate_values(values: list[str]) -> list[str]: seen: set[str] = set() duplicates: set[str] = set() for value in values: if value in seen: duplicates.add(value) seen.add(value) return sorted(duplicates) def trim_output(value: str, max_chars: int) -> tuple[str, bool]: if max_chars <= 0 or len(value) <= max_chars: return value, False return value[-max_chars:], True def parse_json_output(value: str) -> dict[str, Any] | None: text = value.strip() if not text or not text.startswith("{"): return None try: parsed = json.loads(text) except json.JSONDecodeError: return None return parsed if isinstance(parsed, dict) else None def parsed_summary(parsed: dict[str, Any] | None) -> dict[str, Any] | None: if not isinstance(parsed, dict): return None result: dict[str, Any] = {"schema": parsed.get("schema")} for key in ("passed", "status", "failures", "issues"): if key in parsed: value = parsed.get(key) if isinstance(value, list): result[f"{key}_count"] = len(value) else: result[key] = value checks = parsed.get("checks") if isinstance(checks, dict): result["checks_count"] = len(checks) if all(isinstance(value, bool) for value in checks.values()): result["checks_failed"] = sorted(key for key, value in checks.items() if value is not True) counts = parsed.get("counts") if isinstance(counts, dict): result["counts"] = counts bases = parsed.get("bases") if isinstance(bases, dict): result["bases_count"] = len(bases) result["bases"] = { str(base_id): compact_base_summary(base) for base_id, base in bases.items() if isinstance(base, dict) } strict_codes = parsed.get("strict_skip_failure_codes") if isinstance(strict_codes, list): result["strict_skip_failure_codes"] = strict_codes coverage_codes = parsed.get("coverage_failure_codes") if isinstance(coverage_codes, list): result["coverage_failure_codes"] = coverage_codes consistency_codes = parsed.get("consistency_failure_codes") if isinstance(consistency_codes, list): result["consistency_failure_codes"] = consistency_codes safety_codes = parsed.get("safety_failure_codes") if isinstance(safety_codes, list): result["safety_failure_codes"] = safety_codes rollback_safety_codes = parsed.get("rollback_safety_failure_codes") if isinstance(rollback_safety_codes, list): result["rollback_safety_failure_codes"] = rollback_safety_codes saved_state_diff_codes = parsed.get("saved_state_diff_failure_codes") if isinstance(saved_state_diff_codes, list): result["saved_state_diff_failure_codes"] = saved_state_diff_codes schema_codes = parsed.get("schema_failure_codes") if isinstance(schema_codes, list): result["schema_failure_codes"] = schema_codes identity_codes = parsed.get("identity_failure_codes") if isinstance(identity_codes, list): result["identity_failure_codes"] = identity_codes endpoint_codes = parsed.get("endpoint_failure_codes") if isinstance(endpoint_codes, list): result["endpoint_failure_codes"] = endpoint_codes duplicate_codes = parsed.get("duplicate_failure_codes") if isinstance(duplicate_codes, list): result["duplicate_failure_codes"] = duplicate_codes saved_state_codes = parsed.get("saved_state_failure_codes") if isinstance(saved_state_codes, list): result["saved_state_failure_codes"] = saved_state_codes saved_state_strict_readiness_codes = parsed.get("saved_state_strict_readiness_failure_codes") if isinstance(saved_state_strict_readiness_codes, list): result["saved_state_strict_readiness_failure_codes"] = saved_state_strict_readiness_codes saved_state_copy_plan_codes = parsed.get("saved_state_copy_plan_failure_codes") if isinstance(saved_state_copy_plan_codes, list): result["saved_state_copy_plan_failure_codes"] = saved_state_copy_plan_codes saved_state_table_codes = parsed.get("saved_state_table_failure_codes") if isinstance(saved_state_table_codes, list): result["saved_state_table_failure_codes"] = saved_state_table_codes staleness_codes = parsed.get("staleness_failure_codes") if isinstance(staleness_codes, list): result["staleness_failure_codes"] = staleness_codes return result def compact_base_summary(base: dict[str, Any]) -> dict[str, Any]: reports = base.get("reports") if isinstance(base.get("reports"), dict) else {} summary: dict[str, Any] = {"passed": base.get("passed")} selector_chain: dict[str, Any] = {} write_plan_safety: dict[str, Any] = {} write_rollback_safety: dict[str, Any] = {} saved_state_diff: dict[str, Any] = {} saved_state: dict[str, Any] = {} for name, report in reports.items(): if not isinstance(report, dict): continue if name.startswith("selector_chain_"): transport = name.removeprefix("selector_chain_") selector_chain[transport] = { "passed": report.get("passed"), "base_id": report.get("base_id"), "transport": report.get("transport"), "endpoint_url": report.get("endpoint_url"), "resolve_overrides_status": report.get("resolve_overrides_status"), "write_plan_evidence": report.get("write_plan_evidence"), "next_method": report.get("next_method"), "saved_state_status": report.get("saved_state_status"), "saved_state_modules": report.get("saved_state_modules"), "write_plan_target": report.get("write_plan_target"), "composition_status": report.get("composition_status"), "composed": report.get("composed"), } elif name.startswith("write_plan_safety_"): transport = name.removeprefix("write_plan_safety_") write_plan_safety[transport] = { "status": report.get("status"), "checks": report.get("checks"), "base_id": report.get("base_id"), "transport": report.get("transport"), "endpoint_url": report.get("endpoint_url"), } elif name.startswith("write_rollback_safety_"): transport = name.removeprefix("write_rollback_safety_") write_rollback_safety[transport] = { "status": report.get("status"), "checks": report.get("checks"), "base_id": report.get("base_id"), "transport": report.get("transport"), "endpoint_url": report.get("endpoint_url"), } elif name.startswith("saved_state_diff_"): transport = name.removeprefix("saved_state_diff_") saved_state_diff[transport] = { "status": report.get("status"), "checks": report.get("checks"), "base_id": report.get("base_id"), "transport": report.get("transport"), "endpoint_url": report.get("endpoint_url"), "saved_state_table": report.get("saved_state_table"), "diff_status": report.get("diff_status"), "needs_prepare": report.get("needs_prepare"), } elif name == "saved_state_form_write": saved_state["form"] = { "passed": report.get("passed"), "status": report.get("status"), "base_id": report.get("base_id"), "table": report.get("table"), "routes": report.get("routes"), "preflight_status": report.get("preflight_status"), "preflight_counts": report.get("preflight_counts"), } elif name == "saved_state_module_write": saved_state["module"] = { "status": report.get("status"), "base_id": report.get("base_id"), "table": report.get("table"), "module_ref": report.get("module_ref"), "write_plan_allowed": report.get("write_plan_allowed"), "preflight_status": report.get("preflight_status"), "preflight_counts": report.get("preflight_counts"), } elif name == "saved_state_strict_readiness": saved_state["strict_readiness"] = { "status": report.get("status"), "ready": report.get("ready"), "base_id": report.get("base_id"), "table": report.get("table"), "tables": report.get("tables"), "saved_state_rows": report.get("saved_state_rows"), "forms": report.get("forms"), "modules": report.get("modules"), } elif name == "saved_state_copy_plan": source_family = report.get("source_family") if isinstance(report.get("source_family"), dict) else {} saved_state["copy_plan"] = { "status": report.get("status"), "ready_to_copy": report.get("ready_to_copy"), "base_id": report.get("base_id"), "target_table": report.get("target_table"), "source_family": { "expected_source_table": source_family.get("expected_source_table"), "source_tables": source_family.get("source_tables"), "valid": source_family.get("valid"), }, "source_rows": report.get("source_rows"), "found_source_storage_rows": report.get("found_source_storage_rows"), "target_collision_status": report.get("target_collision_status"), "target_collision_rows": report.get("target_collision_rows"), } elif name in {"saved_state_prepare_sql", "saved_state_cleanup_sql"}: key = "prepare_sql" if name == "saved_state_prepare_sql" else "cleanup_sql" saved_state[key] = { "status": report.get("status"), "base_id": report.get("base_id"), "table": report.get("table"), "source_table": report.get("source_table"), "read_only": report.get("read_only"), "sql_write_performed": report.get("sql_write_performed"), } if selector_chain: summary["selector_chain"] = selector_chain if write_plan_safety: summary["write_plan_safety"] = write_plan_safety if write_rollback_safety: summary["write_rollback_safety"] = write_rollback_safety if saved_state_diff: summary["saved_state_diff"] = saved_state_diff if saved_state: summary["saved_state"] = saved_state return summary def run(command: list[str], *, stream: bool, max_output_chars: int) -> dict[str, Any]: label = " ".join(command) if stream: print(f"\n== {label}", flush=True) result = subprocess.run(command, cwd=ROOT, text=True, check=False) return {"label": label, "command": command, "returncode": result.returncode} result = subprocess.run(command, cwd=ROOT, text=True, capture_output=True, check=False) parsed = parse_json_output(result.stdout) stdout, stdout_truncated = trim_output(result.stdout, max_output_chars) stderr, stderr_truncated = trim_output(result.stderr, max_output_chars) report = { "label": label, "command": command, "returncode": result.returncode, "stdout": stdout, "stderr": stderr, "stdout_truncated": stdout_truncated, "stderr_truncated": stderr_truncated, } summary = parsed_summary(parsed) if summary is not None: report["parsed"] = summary return report def main() -> int: parser = argparse.ArgumentParser(description="Run offline/static checks for the 1C adapter verification stack.") parser.add_argument("--base-id", nargs="+", default=["upo_test"], help="Base id(s) whose persisted verify reports should be checked.") parser.add_argument("--skip-persisted-reports", action="store_true", help="Skip checking reports/1c-sql/ artifacts.") parser.add_argument("--require-saved-state-write-smoke", action="store_true", help="Require persisted saved-state smoke reports to contain real write checks.") parser.add_argument("--require-selector-chain-write-plan-composition", action="store_true", help="Require persisted selector-chain reports to contain composed metadata.write.plan coverage.") parser.add_argument("--rest-adapter-url", default=DEFAULT_REST_ADAPTER_URL, help="Expected REST adapter endpoint_url in persisted reports.") parser.add_argument("--mcp-url", default=DEFAULT_MCP_URL, help="Expected MCP proxy endpoint_url in persisted reports.") parser.add_argument("--saved-state-table", choices=SAVED_STATE_TABLES, default="ConfigSave", help="Expected saved-state target table in persisted copy-plan/form/module reports.") parser.add_argument("--max-report-age-seconds", type=int, help="Fail if any checked persisted report file is older than this many seconds.") parser.add_argument("--report", type=Path, help="Optional JSON report path.") parser.add_argument("--json", action="store_true", help="Print JSON summary without streaming child command output.") parser.add_argument("--max-output-chars", type=int, default=4000, help="Maximum stdout/stderr characters retained per child command in JSON mode.") args = parser.parse_args() duplicate_base_ids = duplicate_values(args.base_id) if duplicate_base_ids: report = { "schema": "onec_adapter_verification_stack_check.v1", "passed": False, "base_id": args.base_id[0] if len(args.base_id) == 1 else None, "base_ids": args.base_id, "rest_adapter_url": args.rest_adapter_url, "mcp_url": args.mcp_url, "saved_state_table": args.saved_state_table, "checks": [], "failures": [{"code": "duplicate_base_id", "base_ids": duplicate_base_ids}], } if args.report: report_path = args.report if args.report.is_absolute() else ROOT / args.report report_path.parent.mkdir(parents=True, exist_ok=True) report_path.write_text(json.dumps(report, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") if args.json: print(json.dumps(report, ensure_ascii=False, indent=2)) else: print("Duplicate --base-id value(s): " + ", ".join(duplicate_base_ids), file=sys.stderr) return 1 commands = [ [ sys.executable, "-m", "py_compile", "scripts/check_1c_adapter_verification_stack.py", "scripts/check_1c_verify_reports.py", "scripts/check_powershell_scripts.py", "scripts/smoke_1c_mcp_selector_chain.py", "scripts/smoke_1c_write_plan_safety.py", "scripts/smoke_1c_write_preflight.py", "scripts/smoke_1c_write_rollback_safety.py", "scripts/smoke_1c_saved_state_diff.py", "scripts/smoke_1c_saved_state_changes.py", "scripts/smoke_1c_saved_state_write_routes.py", "scripts/smoke_1c_saved_state_module_write.py", "scripts/check_1c_saved_state_strict_readiness.py", "scripts/plan_1c_saved_state_copy.py", "scripts/prepare_1c_saved_state_copy_sql.py", "scripts/prepare_1c_saved_state_cleanup_sql.py", "scripts/verify_1c_saved_state_copy.py", ], [sys.executable, "scripts/check_powershell_scripts.py"], [sys.executable, "scripts/check_1c_mcp_adapter_contract.py", "--json"], [sys.executable, "scripts/check_1c_extension_action_contract.py", "--print"], [sys.executable, "scripts/check_1c_write_plan_contract.py", "--print"], [sys.executable, "scripts/smoke_1c_mcp_selector_chain.py", "--json", "--no-report"], [sys.executable, "scripts/check_1c_verify_reports.py", "--self-test", "--json"], ] if not args.skip_persisted_reports: report_command = [ sys.executable, "scripts/check_1c_verify_reports.py", "--base-id", *args.base_id, "--rest-adapter-url", args.rest_adapter_url, "--mcp-url", args.mcp_url, "--saved-state-table", args.saved_state_table, "--json", ] if args.require_saved_state_write_smoke: report_command.append("--require-saved-state-write-smoke") if args.require_selector_chain_write_plan_composition: report_command.append("--require-selector-chain-write-plan-composition") if args.max_report_age_seconds is not None: report_command.extend(["--max-report-age-seconds", str(args.max_report_age_seconds)]) commands.append(report_command) results: list[dict[str, Any]] = [] failures: list[str] = [] for command in commands: result = run(command, stream=not args.json, max_output_chars=args.max_output_chars) results.append(result) if result["returncode"] != 0: failures.append(str(result["label"])) report = { "schema": "onec_adapter_verification_stack_check.v1", "passed": not failures, "base_id": args.base_id[0] if len(args.base_id) == 1 else None, "base_ids": args.base_id, "rest_adapter_url": args.rest_adapter_url, "mcp_url": args.mcp_url, "saved_state_table": args.saved_state_table, "checks": results, "failures": failures, } if args.report: report_path = args.report if args.report.is_absolute() else ROOT / args.report report_path.parent.mkdir(parents=True, exist_ok=True) report_path.write_text(json.dumps(report, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") if args.json: print(json.dumps(report, ensure_ascii=False, indent=2)) return 0 if report["passed"] else 1 if failures: print("\n1C adapter verification stack check failed:", file=sys.stderr) for failure in failures: print(f"- {failure}", file=sys.stderr) return 1 print("\n1C adapter verification stack checks passed.") return 0 if __name__ == "__main__": raise SystemExit(main())