132 lines
5.4 KiB
Python
132 lines
5.4 KiB
Python
#!/usr/bin/env python3
|
|
"""Capture and summarize the two remaining 1C form command-binding learning cases."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import time
|
|
import urllib.request
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
|
|
DEFAULT_BASE_URL = "http://docker.cin.su:8011"
|
|
DEFAULT_BASE_ID = "upo_test"
|
|
DEFAULT_TABLE = "ConfigCASSave"
|
|
DEFAULT_FILE_NAME = "f96a0c45-3eff-11f1-8287-005056b0d483__25c39fbf-35a4-4b43-8e3f-cd1f91082c88.0"
|
|
DEFAULT_REPORT = Path("reports/1c-sql/upo_test/form-command-binding-learning-run.json")
|
|
|
|
CASES = [
|
|
{
|
|
"learning_id": "form-command-binding-local-apply-command",
|
|
"element": "ФормаКомандаОбновить",
|
|
"property": "ИмяКоманды",
|
|
"expected_xml": "Form.Command.КомандаПрименить",
|
|
},
|
|
{
|
|
"learning_id": "form-command-binding-standard-customize-form",
|
|
"element": "ТЗИзменитьФорму",
|
|
"property": "ИмяКоманды",
|
|
"expected_xml": "Form.StandardCommand.CustomizeForm",
|
|
},
|
|
]
|
|
|
|
|
|
def rpc(base_url: str, method: str, payload: dict[str, Any], timeout: int) -> dict[str, Any]:
|
|
req = urllib.request.Request(
|
|
f"{base_url.rstrip('/')}/rpc",
|
|
data=json.dumps({"method": method, "payload": payload}, ensure_ascii=False).encode("utf-8"),
|
|
headers={"Content-Type": "application/json"},
|
|
)
|
|
with urllib.request.urlopen(req, timeout=timeout) as response:
|
|
return json.loads(response.read().decode("utf-8"))
|
|
|
|
|
|
def storage_sha(base_url: str, base_id: str, table: str, file_name: str, timeout: int) -> str | None:
|
|
result = rpc(
|
|
base_url,
|
|
"metadata.write_learning.capture_after",
|
|
{
|
|
"base_id": base_id,
|
|
"learning_id": "_command_binding_probe",
|
|
"table": table,
|
|
"file_name": file_name,
|
|
"timeout_seconds": timeout,
|
|
"max_items": 5000,
|
|
},
|
|
timeout,
|
|
)
|
|
storage = result.get("storage") if isinstance(result.get("storage"), dict) else {}
|
|
return storage.get("sha1")
|
|
|
|
|
|
def run_case(base_url: str, base_id: str, table: str, file_name: str, case: dict[str, str], timeout: int) -> dict[str, Any]:
|
|
capture_payload = {
|
|
"base_id": base_id,
|
|
"learning_id": case["learning_id"],
|
|
"table": table,
|
|
"file_name": file_name,
|
|
"form": "ТестНастройки",
|
|
"element": case["element"],
|
|
"property": case["property"],
|
|
"timeout_seconds": timeout,
|
|
"max_items": 5000,
|
|
}
|
|
after = rpc(base_url, "metadata.write_learning.capture_after", capture_payload, timeout)
|
|
diff = rpc(base_url, "metadata.write_learning.diff", {"learning_id": case["learning_id"]}, timeout)
|
|
infer = rpc(base_url, "metadata.write_learning.infer_rule", {"learning_id": case["learning_id"]}, timeout)
|
|
return {
|
|
"learning_id": case["learning_id"],
|
|
"element": case["element"],
|
|
"property": case["property"],
|
|
"expected_xml": case["expected_xml"],
|
|
"after": after,
|
|
"diff": diff,
|
|
"infer_rule": infer,
|
|
}
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--base-url", default=DEFAULT_BASE_URL)
|
|
parser.add_argument("--base-id", default=DEFAULT_BASE_ID)
|
|
parser.add_argument("--table", default=DEFAULT_TABLE)
|
|
parser.add_argument("--file-name", default=DEFAULT_FILE_NAME)
|
|
parser.add_argument("--report", type=Path, default=DEFAULT_REPORT)
|
|
parser.add_argument("--timeout", type=int, default=60)
|
|
parser.add_argument("--wait-for-sha-change", default="", help="Poll until storage sha1 differs from this value.")
|
|
parser.add_argument("--poll-seconds", type=int, default=10)
|
|
parser.add_argument("--max-wait-seconds", type=int, default=0)
|
|
args = parser.parse_args()
|
|
|
|
waited = 0
|
|
observed_sha = None
|
|
if args.wait_for_sha_change:
|
|
while True:
|
|
observed_sha = storage_sha(args.base_url, args.base_id, args.table, args.file_name, args.timeout)
|
|
if observed_sha and observed_sha != args.wait_for_sha_change:
|
|
break
|
|
if args.max_wait_seconds and waited >= args.max_wait_seconds:
|
|
break
|
|
time.sleep(max(1, args.poll_seconds))
|
|
waited += max(1, args.poll_seconds)
|
|
|
|
cases = [run_case(args.base_url, args.base_id, args.table, args.file_name, case, args.timeout) for case in CASES]
|
|
report = {
|
|
"schema": "onec_form_command_binding_learning_run.v1",
|
|
"status": "changed" if any((case.get("diff") or {}).get("status") == "changed" for case in cases) else "no_changes",
|
|
"base_id": args.base_id,
|
|
"source": {"table": args.table, "file_name": args.file_name},
|
|
"wait": {"expected_old_sha1": args.wait_for_sha_change or None, "observed_sha1": observed_sha, "waited_seconds": waited},
|
|
"cases": cases,
|
|
}
|
|
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")
|
|
print(json.dumps({"status": report["status"], "report": str(args.report), "cases": [{ "learning_id": c["learning_id"], "diff_status": (c.get("diff") or {}).get("status"), "counts": (c.get("diff") or {}).get("counts")} for c in cases]}, ensure_ascii=False, indent=2))
|
|
return 0 if report["status"] == "changed" else 2
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|