404 lines
19 KiB
Python
404 lines
19 KiB
Python
#!/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/<base-id> 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())
|