Files
llm/scripts/smoke_1c_write_preflight.py
T

280 lines
13 KiB
Python

#!/usr/bin/env python3
from __future__ import annotations
import argparse
import json
import urllib.request
from pathlib import Path
from typing import Any
from urllib.error import HTTPError, URLError
DEFAULT_BASE_URL = "http://docker-gpu.cin.su:8011"
DEFAULT_MCP_URL = "http://docker.cin.su:8021"
EXPECTED_CONTRACT_VERSION = "onec-selector-contract.v1"
PREFLIGHT_CLASSIFICATION_STATUSES = {
"ready",
"needs_prepare",
"needs_resolution",
"needs_repository_lock",
"blocked",
"blocked_by_support",
"blocked_by_support_live_sql",
"blocked_by_support_rule",
"blocked_support_unknown",
}
def post_json(url: str, payload: dict[str, Any], *, timeout: float, headers: dict[str, str] | None = None) -> tuple[dict[str, str], dict[str, Any]]:
request = urllib.request.Request(
url,
data=json.dumps(payload, ensure_ascii=False).encode("utf-8"),
method="POST",
headers={"Content-Type": "application/json", **(headers or {})},
)
with urllib.request.urlopen(request, timeout=timeout) as response:
raw = response.read().decode("utf-8")
data = json.loads(raw) if raw else {}
return dict(response.headers), data if isinstance(data, dict) else {"status": "error", "error": "response_not_object", "response": data}
def rest_rpc(base_url: str, method: str, payload: dict[str, Any], timeout: float) -> dict[str, Any]:
_, data = post_json(
base_url.rstrip("/") + "/rpc",
{"method": method, "payload": payload},
timeout=timeout,
headers={"Content-Type": "application/json; charset=utf-8"},
)
return data
def mcp_initialize(mcp_url: str, timeout: float) -> tuple[str | None, dict[str, Any]]:
body = {
"jsonrpc": "2.0",
"id": "write-preflight-initialize",
"method": "initialize",
"params": {
"protocolVersion": "2025-06-18",
"capabilities": {},
"clientInfo": {"name": "smoke_1c_write_preflight", "version": "1"},
},
}
try:
headers, data = post_json(
mcp_url.rstrip("/") + "/mcp",
body,
timeout=timeout,
headers={"Accept": "application/json, text/event-stream"},
)
except HTTPError as exc:
return None, {"status": "error", "error": "mcp_initialize_http_error", "http_status": exc.code, "response": exc.read().decode("utf-8", errors="replace")}
except URLError as exc:
return None, {"status": "error", "error": "mcp_initialize_url_error", "message": str(exc)}
if "error" in data:
return None, {"status": "error", "error": "mcp_initialize_jsonrpc_error", "details": data.get("error")}
return headers.get("Mcp-Session-Id"), data
def mcp_rpc(mcp_url: str, method: str, payload: dict[str, Any], timeout: float, session_id: str | None) -> dict[str, Any]:
headers = {"Accept": "application/json, text/event-stream"}
if session_id:
headers["Mcp-Session-Id"] = session_id
body = {
"jsonrpc": "2.0",
"id": f"write-preflight-{method}",
"method": "tools/call",
"params": {"name": "onec_request", "arguments": {"method": method, "payload": payload}},
}
try:
_, data = post_json(mcp_url.rstrip("/") + "/mcp", body, timeout=timeout, headers=headers)
except HTTPError as exc:
return {"status": "error", "error": "mcp_http_error", "http_status": exc.code, "response": exc.read().decode("utf-8", errors="replace")}
except URLError as exc:
return {"status": "error", "error": "mcp_url_error", "message": str(exc)}
if "error" in data:
return {"status": "error", "error": "mcp_jsonrpc_error", "details": data.get("error")}
result = data.get("result") if isinstance(data.get("result"), dict) else {}
content = result.get("content") if isinstance(result.get("content"), list) else []
first = content[0] if content and isinstance(content[0], dict) else {}
text = first.get("text")
if not isinstance(text, str):
return {"status": "error", "error": "mcp_tool_text_missing", "result": result}
try:
decoded = json.loads(text)
except json.JSONDecodeError as exc:
return {"status": "error", "error": "mcp_tool_text_not_json", "message": str(exc), "text": text[:500]}
return decoded if isinstance(decoded, dict) else {"status": "error", "error": "mcp_tool_payload_not_object", "payload": decoded}
def rpc_call(endpoint_url: str, method: str, payload: dict[str, Any], timeout: float, *, transport: str, session_id: str | None) -> dict[str, Any]:
if transport == "mcp":
return mcp_rpc(endpoint_url, method, payload, timeout, session_id)
return rest_rpc(endpoint_url, method, payload, timeout)
def require(condition: bool, message: str, failures: list[str]) -> None:
if not condition:
failures.append(message)
def classified_preflight_status(value: Any) -> bool:
return str(value or "") in PREFLIGHT_CLASSIFICATION_STATUSES
def method_names(help_result: dict[str, Any]) -> set[str]:
return {str(item.get("name") or "") for item in help_result.get("methods") or [] if isinstance(item, dict)}
def first_saved_module_target(endpoint_url: str, base_id: str, timeout: float, *, transport: str, session_id: str | None) -> dict[str, Any] | None:
result = rpc_call(
endpoint_url,
"metadata.saved_state.modules.search",
{"base_id": base_id, "tables": ["ConfigCASSave", "ConfigSave"], "limit": 5, "scan_limit": 500, "timeout_seconds": int(timeout)},
timeout,
transport=transport,
session_id=session_id,
)
modules = result.get("modules") if isinstance(result.get("modules"), list) else []
targets: list[dict[str, Any]] = []
for module in modules:
if not isinstance(module, dict):
continue
for stream in module.get("streams") or []:
if isinstance(stream, dict) and isinstance(stream.get("write_plan_target"), dict):
targets.append(stream["write_plan_target"])
# A concrete stream target can be freshness-checked without decoding and
# structurally diffing a whole saved form descriptor. Prefer it so this
# deployment smoke remains bounded on large ConfigCASSave forms.
return next(
(
target
for target in targets
if target.get("stream_index") is not None or "#stream:" in str(target.get("module_ref") or "")
),
targets[0] if targets else None,
)
def run_smoke(endpoint_url: str, base_id: str, timeout: float, *, transport: str) -> dict[str, Any]:
failures: list[str] = []
checks: dict[str, Any] = {}
session_id: str | None = None
initialize_result: dict[str, Any] | None = None
if transport == "mcp":
session_id, initialize_result = mcp_initialize(endpoint_url, timeout)
contract_version = (((initialize_result.get("result") or {}).get("serverInfo") or {}).get("contract_version") if isinstance(initialize_result.get("result"), dict) else None)
checks["mcp.initialize"] = {
"status": "ok" if session_id and contract_version == EXPECTED_CONTRACT_VERSION else "error",
"session": bool(session_id),
"contract_version": contract_version,
}
require(bool(session_id), "MCP initialize must return Mcp-Session-Id", failures)
require(contract_version == EXPECTED_CONTRACT_VERSION, "MCP initialize must report expected contract version", failures)
help_result = rpc_call(endpoint_url, "help.methods", {}, timeout, transport=transport, session_id=session_id)
names = method_names(help_result)
checks["method_exposed"] = {
"status": "ok" if "metadata.write.preflight" in names else "missing",
"method_count": len(names),
}
require("metadata.write.preflight" in names, "metadata.write.preflight must be exposed by help.methods", failures)
blocked = rpc_call(
endpoint_url,
"metadata.write.preflight",
{
"base_id": base_id,
"target": {"kind": "form", "canonical_path": "Справочник.Контрагенты.Наименование"},
"intent": {"operation": "property_change", "property": "Заголовок", "value": "SMOKE"},
"timeout_seconds": int(timeout),
},
timeout,
transport=transport,
session_id=session_id,
)
checks["effective_path_preflight"] = {
"schema": blocked.get("schema"),
"status": blocked.get("status"),
"allowed": blocked.get("allowed"),
"plan_allowed": (blocked.get("plan") or {}).get("allowed") if isinstance(blocked.get("plan"), dict) else None,
}
require(blocked.get("schema") == "onec_metadata_write_preflight.v1", "preflight must return expected schema", failures)
require(classified_preflight_status(blocked.get("status")), "effective path preflight must classify readiness or a safety gate", failures)
require(blocked.get("allowed") is False, "effective path preflight must not be allowed", failures)
target = first_saved_module_target(endpoint_url, base_id, timeout, transport=transport, session_id=session_id)
if target:
concrete = rpc_call(
endpoint_url,
"metadata.write.preflight",
{
"base_id": base_id,
"target": {"kind": "module", **target},
"intent": {"operation": "replace_with_control", "control_fragment": "codex-smoke-nonmatching-fragment", "new": "codex-smoke"},
"timeout_seconds": int(timeout),
},
timeout,
transport=transport,
session_id=session_id,
)
freshness = (concrete.get("saved_state") or {}).get("freshness") if isinstance(concrete.get("saved_state"), dict) else {}
checks["concrete_saved_state_preflight"] = {
"schema": concrete.get("schema"),
"status": concrete.get("status"),
"allowed": concrete.get("allowed"),
"saved_state_status": (concrete.get("saved_state") or {}).get("status") if isinstance(concrete.get("saved_state"), dict) else None,
"freshness": freshness.get("status") if isinstance(freshness, dict) else None,
"writer": (concrete.get("route") or {}).get("writer") if isinstance(concrete.get("route"), dict) else None,
}
require(concrete.get("schema") == "onec_metadata_write_preflight.v1", "concrete preflight must return expected schema", failures)
require(classified_preflight_status(concrete.get("status")), "concrete preflight must classify readiness or a safety gate", failures)
if concrete.get("status") == "ready":
require(freshness.get("status") == "live_sql_verified", "ready concrete preflight must be live SQL verified", failures)
else:
checks["concrete_saved_state_preflight"] = {"status": "skipped_no_saved_module_target"}
return {
"schema": "onec_write_preflight_smoke.v1",
"status": "ok" if not failures else "failed",
"endpoint_url": endpoint_url,
"transport": transport,
"base_id": base_id,
"checks": checks,
"failures": failures,
**({"mcp_initialize": initialize_result} if initialize_result and failures else {}),
}
def main() -> int:
parser = argparse.ArgumentParser(description="Smoke deployed metadata.write.preflight behavior without SQL writes.")
parser.add_argument("--base-url", default=DEFAULT_BASE_URL)
parser.add_argument("--mcp-url", default=DEFAULT_MCP_URL)
parser.add_argument("--transport", choices=("rest", "mcp"), default="rest")
parser.add_argument("--base-id", default="upo_test")
parser.add_argument("--timeout", type=float, default=30.0)
parser.add_argument("--report", type=Path)
args = parser.parse_args()
endpoint_url = args.mcp_url if args.transport == "mcp" else args.base_url
try:
report = run_smoke(endpoint_url, args.base_id, args.timeout, transport=args.transport)
except (TimeoutError, URLError, OSError) as exc:
report = {
"schema": "onec_write_preflight_smoke.v1",
"status": "failed",
"endpoint_url": endpoint_url,
"transport": args.transport,
"base_id": args.base_id,
"checks": {},
"failures": [f"transport timeout/error: {type(exc).__name__}: {str(exc)[:300]}"],
}
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")
print(json.dumps(report, ensure_ascii=False, indent=2))
return 0 if report["status"] == "ok" else 1
if __name__ == "__main__":
raise SystemExit(main())