248 lines
11 KiB
Python
248 lines
11 KiB
Python
#!/usr/bin/env python3
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import sys
|
|
import urllib.request
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
|
|
DEFAULT_BASE_URL = "http://docker.cin.su:8011"
|
|
SOURCE_BY_TARGET = {"ConfigSave": "Config", "ConfigCASSave": "ConfigCAS"}
|
|
|
|
|
|
def rpc(base_url: str, method: str, payload: dict[str, Any], timeout: float) -> dict[str, Any]:
|
|
body = json.dumps({"method": method, "payload": payload}, ensure_ascii=False).encode("utf-8")
|
|
request = urllib.request.Request(
|
|
base_url.rstrip("/") + "/rpc",
|
|
data=body,
|
|
method="POST",
|
|
headers={"Content-Type": "application/json; charset=utf-8"},
|
|
)
|
|
with urllib.request.urlopen(request, timeout=timeout) as response:
|
|
data = json.loads(response.read().decode("utf-8"))
|
|
if not isinstance(data, dict):
|
|
raise RuntimeError(f"{method} returned non-object response: {data!r}")
|
|
return data
|
|
|
|
|
|
def sql_literal(value: str) -> str:
|
|
return "'" + value.replace("'", "''") + "'"
|
|
|
|
|
|
def read_json(path: Path) -> dict[str, Any]:
|
|
data = json.loads(path.read_text(encoding="utf-8"))
|
|
if not isinstance(data, dict):
|
|
raise ValueError(f"{path} must contain a JSON object")
|
|
return data
|
|
|
|
|
|
def expected_rows_from_plan(plan: dict[str, Any]) -> tuple[list[dict[str, Any]], dict[str, Any]]:
|
|
target = plan.get("target") if isinstance(plan.get("target"), dict) else {}
|
|
source_family = plan.get("source_family") if isinstance(plan.get("source_family"), dict) else {}
|
|
target_table = str(target.get("table") or "")
|
|
source_table = str(source_family.get("expected_source_table") or SOURCE_BY_TARGET.get(target_table) or "")
|
|
source_details = plan.get("source_row_details") if isinstance(plan.get("source_row_details"), dict) else {}
|
|
table_details = source_details.get(source_table) if isinstance(source_details.get(source_table), dict) else {}
|
|
expected: list[dict[str, Any]] = []
|
|
for row in table_details.get("rows") or []:
|
|
if not isinstance(row, dict):
|
|
continue
|
|
file_name = str(row.get("FileName") or "")
|
|
part_no = row.get("PartNo")
|
|
if file_name and isinstance(part_no, int):
|
|
expected.append({
|
|
"FileName": file_name,
|
|
"PartNo": part_no,
|
|
"DataSize": row.get("DataSize"),
|
|
"BinaryBytes": row.get("BinaryBytes"),
|
|
"BinarySHA1": str(row.get("BinarySHA1") or "").upper(),
|
|
})
|
|
metadata = {
|
|
"base_id": plan.get("base_id"),
|
|
"target_table": target_table,
|
|
"source_table": source_table,
|
|
"source_family_valid": source_family.get("valid"),
|
|
}
|
|
return expected, metadata
|
|
|
|
|
|
def validate_plan(plan: dict[str, Any], *, expected_base_id: str | None, expected_target_table: str | None) -> list[str]:
|
|
failures: list[str] = []
|
|
target = plan.get("target") if isinstance(plan.get("target"), dict) else {}
|
|
source_family = plan.get("source_family") if isinstance(plan.get("source_family"), dict) else {}
|
|
target_table = str(target.get("table") or "")
|
|
expected_source_table = SOURCE_BY_TARGET.get(target_table)
|
|
if plan.get("schema") != "onec_saved_state_copy_plan.v1":
|
|
failures.append("schema must be onec_saved_state_copy_plan.v1")
|
|
if plan.get("status") != "plan_ready" or plan.get("ready_to_copy") is not True:
|
|
failures.append("copy plan must be plan_ready and ready_to_copy=true")
|
|
if expected_base_id is not None and plan.get("base_id") != expected_base_id:
|
|
failures.append(f"base_id must be {expected_base_id}")
|
|
if expected_target_table is not None and target_table != expected_target_table:
|
|
failures.append(f"target.table must be {expected_target_table}")
|
|
if target_table not in SOURCE_BY_TARGET:
|
|
failures.append("target.table must be ConfigSave or ConfigCASSave")
|
|
if source_family.get("valid") is not True:
|
|
failures.append("source_family.valid must be true")
|
|
if expected_source_table and source_family.get("expected_source_table") != expected_source_table:
|
|
failures.append(f"source_family.expected_source_table must be {expected_source_table}")
|
|
target_collisions = plan.get("target_collisions") if isinstance(plan.get("target_collisions"), dict) else {}
|
|
if target_collisions.get("status") != "clear":
|
|
failures.append("copy plan must have been generated before preparation with target_collisions.status=clear")
|
|
return failures
|
|
|
|
|
|
def fetch_target_rows(base_url: str, base_id: str, target_table: str, file_names: list[str], timeout: float) -> dict[str, Any]:
|
|
query = (
|
|
"SELECT FileName, PartNo, DataSize, DATALENGTH(BinaryData) AS BinaryBytes, "
|
|
"CONVERT(varchar(40), HASHBYTES('SHA1', BinaryData), 2) AS BinarySHA1 "
|
|
f"FROM {target_table} WHERE FileName IN ({', '.join(sql_literal(name) for name in file_names)}) "
|
|
"ORDER BY FileName, PartNo"
|
|
)
|
|
return rpc(
|
|
base_url,
|
|
"query.run",
|
|
{"base_id": base_id, "diagnostic": True, "query": query, "timeout_seconds": int(timeout)},
|
|
timeout,
|
|
)
|
|
|
|
|
|
def row_key(row: dict[str, Any]) -> tuple[str, int]:
|
|
return str(row.get("FileName") or ""), int(row.get("PartNo"))
|
|
|
|
|
|
def compare_rows(expected: list[dict[str, Any]], actual: list[dict[str, Any]]) -> tuple[list[dict[str, Any]], list[dict[str, Any]], list[dict[str, Any]]]:
|
|
expected_by_key = {row_key(row): row for row in expected}
|
|
actual_by_key = {row_key(row): row for row in actual if row.get("FileName") and isinstance(row.get("PartNo"), int)}
|
|
missing = [row for key, row in sorted(expected_by_key.items()) if key not in actual_by_key]
|
|
unexpected = [row for key, row in sorted(actual_by_key.items()) if key not in expected_by_key]
|
|
mismatched: list[dict[str, Any]] = []
|
|
for key, expected_row in sorted(expected_by_key.items()):
|
|
actual_row = actual_by_key.get(key)
|
|
if actual_row is None:
|
|
continue
|
|
diffs = {}
|
|
for field in ("DataSize", "BinaryBytes", "BinarySHA1"):
|
|
expected_value = expected_row.get(field)
|
|
actual_value = actual_row.get(field)
|
|
if field == "BinarySHA1":
|
|
actual_value = str(actual_value or "").upper()
|
|
if actual_value != expected_value:
|
|
diffs[field] = {"expected": expected_value, "actual": actual_value}
|
|
if diffs:
|
|
mismatched.append({"FileName": key[0], "PartNo": key[1], "diffs": diffs})
|
|
return missing, unexpected, mismatched
|
|
|
|
|
|
def build_report(args: argparse.Namespace) -> dict[str, Any]:
|
|
plan = read_json(args.plan)
|
|
plan_failures = validate_plan(plan, expected_base_id=args.expected_base_id, expected_target_table=args.expected_target_table)
|
|
expected, metadata = expected_rows_from_plan(plan)
|
|
report: dict[str, Any] = {
|
|
"schema": "onec_saved_state_copy_verify.v1",
|
|
"base_url": args.base_url,
|
|
"base_id": metadata.get("base_id"),
|
|
"target_table": metadata.get("target_table"),
|
|
"source_table": metadata.get("source_table"),
|
|
"plan_path": str(args.plan),
|
|
"read_only": True,
|
|
"sql_write_performed": False,
|
|
"ready": False,
|
|
"status": "error",
|
|
"failures": [],
|
|
}
|
|
if plan_failures:
|
|
report["status"] = "blocked_invalid_plan"
|
|
report["failures"] = plan_failures
|
|
return report
|
|
if not expected:
|
|
report["status"] = "blocked_no_expected_rows"
|
|
report["failures"] = ["copy plan source_row_details did not contain expected rows"]
|
|
return report
|
|
actual_result = fetch_target_rows(
|
|
args.base_url,
|
|
str(metadata["base_id"]),
|
|
str(metadata["target_table"]),
|
|
sorted({str(row["FileName"]) for row in expected}),
|
|
args.timeout,
|
|
)
|
|
actual = [row for row in actual_result.get("rows") or [] if isinstance(row, dict)]
|
|
missing, unexpected, mismatched = compare_rows(expected, actual)
|
|
report["checks"] = {
|
|
"target_rows": {
|
|
"status": actual_result.get("status"),
|
|
"validation": actual_result.get("validation"),
|
|
"counts": actual_result.get("counts"),
|
|
},
|
|
"comparison": {
|
|
"expected_rows": len(expected),
|
|
"actual_rows": len(actual),
|
|
"missing_rows": len(missing),
|
|
"unexpected_rows": len(unexpected),
|
|
"mismatched_rows": len(mismatched),
|
|
},
|
|
}
|
|
report["details"] = {
|
|
"missing": missing,
|
|
"unexpected": unexpected,
|
|
"mismatched": mismatched,
|
|
}
|
|
if missing and not actual:
|
|
report["status"] = "blocked_missing_target_rows"
|
|
report["failures"] = ["planned rows are not present in the selected save layer"]
|
|
elif missing:
|
|
report["status"] = "blocked_incomplete_target_rows"
|
|
report["failures"] = ["some planned rows are missing from the selected save layer"]
|
|
elif unexpected:
|
|
report["status"] = "blocked_unexpected_target_rows"
|
|
report["failures"] = ["selected save layer returned rows outside the reviewed plan"]
|
|
elif mismatched:
|
|
report["status"] = "blocked_target_row_mismatch"
|
|
report["failures"] = ["one or more target rows differ from the reviewed active source rows"]
|
|
else:
|
|
report["status"] = "ready"
|
|
report["ready"] = True
|
|
return report
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description="Read-only verification that a planned 1C saved-state copy is present and byte-identical in the save layer.")
|
|
parser.add_argument("--base-url", default=DEFAULT_BASE_URL)
|
|
parser.add_argument("--plan", type=Path, default=Path("reports/1c-sql/upo_test/saved-state-copy-plan.json"))
|
|
parser.add_argument("--expected-base-id")
|
|
parser.add_argument("--expected-target-table", choices=sorted(SOURCE_BY_TARGET))
|
|
parser.add_argument("--timeout", type=float, default=60.0)
|
|
parser.add_argument("--report", type=Path)
|
|
parser.add_argument("--require-ready", action="store_true")
|
|
parser.add_argument("--json", action="store_true")
|
|
args = parser.parse_args()
|
|
|
|
try:
|
|
report = build_report(args)
|
|
except Exception as exc:
|
|
report = {
|
|
"schema": "onec_saved_state_copy_verify.v1",
|
|
"base_url": args.base_url,
|
|
"plan_path": str(args.plan),
|
|
"read_only": True,
|
|
"sql_write_performed": False,
|
|
"ready": False,
|
|
"status": "error",
|
|
"failures": [str(exc)],
|
|
}
|
|
if args.report:
|
|
args.report.parent.mkdir(parents=True, exist_ok=True)
|
|
args.report.write_text(json.dumps(report, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
|
if args.json or args.require_ready or not report.get("ready"):
|
|
print(json.dumps(report, ensure_ascii=False, indent=2), file=sys.stderr if args.require_ready and not report.get("ready") else sys.stdout)
|
|
else:
|
|
print(f"OK: saved-state copy verification passed for {report.get('base_id')} -> {report.get('target_table')}.")
|
|
return 0 if report.get("ready") or not args.require_ready else 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|