m/llm
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Require name-first extension preflight smoke

Browse Source
This commit is contained in:
Mikhail
2026-07-26 17:48:04 +03:00
parent ef2a94a3d0
commit d99d57eaf9
5 changed files with 333 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/smoke_1c_write_preflight.py
+135 -2
View File
@@ -3,6 +3,7 @@ from __future__ import annotations
import argparse
import json
import re
import urllib.request
from pathlib import Path
from typing import Any
@@ -23,6 +24,10 @@ PREFLIGHT_CLASSIFICATION_STATUSES = {
"blocked_by_support_rule",
"blocked_support_unknown",
}
EXTENSION_GUID_LAYER_RE = re.compile(
r"^extension:[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$",
re.IGNORECASE,
)
def post_json(url: str, payload: dict[str, Any], *, timeout: float, headers: dict[str, str] | None = None) -> tuple[dict[str, str], dict[str, Any]]:
@@ -155,7 +160,61 @@ def first_saved_module_target(endpoint_url: str, base_id: str, timeout: float, *
)
def run_smoke(endpoint_url: str, base_id: str, timeout: float, *, transport: str) -> dict[str, Any]:
def first_public_extension_form_candidate(result: dict[str, Any]) -> dict[str, Any] | None:
candidates: list[dict[str, Any]] = []
for form in result.get("forms") or []:
if not isinstance(form, dict):
continue
for match in form.get("matches") or []:
if not isinstance(match, dict):
continue
selector = match.get("selector") if isinstance(match.get("selector"), dict) else {}
if not selector.get("extension") or not selector.get("ref") or not selector.get("form"):
continue
if not any(selector.get(key) for key in ("command", "element", "attribute")):
continue
if any(key in selector for key in ("table", "file_name", "guid", "form_guid", "module_ref")):
continue
for prop in match.get("writable_properties") or []:
if not isinstance(prop, dict) or prop.get("value_type") != "string":
continue
property_name = str(prop.get("presentation") or prop.get("property") or "").strip()
if not property_name:
continue
normalized_property = str(prop.get("property") or "").strip().casefold()
candidates.append(
{
"priority": 0 if normalized_property == "title" else (2 if normalized_property in {"id", "name"} else 1),
"target": {
"kind": "form",
**{
key: value
for key, value in selector.items()
if key in {"extension", "ref", "form", "command", "element", "attribute"}
and value not in (None, "")
},
},
"edit": {
"property": property_name,
"value": f"{str(prop.get('value') or '')} [NAME-FIRST PREFLIGHT]",
},
}
)
if not candidates:
return None
candidate = sorted(candidates, key=lambda item: int(item.get("priority") or 0))[0]
candidate.pop("priority", None)
return candidate
def run_smoke(
endpoint_url: str,
base_id: str,
timeout: float,
*,
transport: str,
require_name_first_extension_form: bool = False,
) -> dict[str, Any]:
failures: list[str] = []
checks: dict[str, Any] = {}
session_id: str | None = None
@@ -233,12 +292,79 @@ def run_smoke(endpoint_url: str, base_id: str, timeout: float, *, transport: str
else:
checks["concrete_saved_state_preflight"] = {"status": "skipped_no_saved_module_target"}
forms = rpc_call(
endpoint_url,
"metadata.saved_state.forms.search",
{
"base_id": base_id,
"limit": 20,
"scan_limit": 1000,
"timeout_seconds": int(timeout),
},
timeout,
transport=transport,
session_id=session_id,
)
form_candidate = first_public_extension_form_candidate(forms)
if not form_candidate:
checks["name_first_extension_form_preflight"] = {
"status": "skipped_no_public_extension_form_target",
"search_status": forms.get("status"),
"forms": int((forms.get("counts") or {}).get("forms") or 0),
}
if require_name_first_extension_form:
failures.append("name-first extension form preflight target is required but was not discovered")
else:
form_preflight = rpc_call(
endpoint_url,
"metadata.write.preflight",
{
"base_id": base_id,
"target": form_candidate["target"],
"edits": [form_candidate["edit"]],
"resolve_origin": False,
"timeout_seconds": int(timeout),
},
timeout,
transport=transport,
session_id=session_id,
)
plan = form_preflight.get("plan") if isinstance(form_preflight.get("plan"), dict) else {}
repository = form_preflight.get("repository") if isinstance(form_preflight.get("repository"), dict) else {}
support = form_preflight.get("support") if isinstance(form_preflight.get("support"), dict) else {}
repository_layer = str(repository.get("layer_id") or "")
support_layer = str(support.get("layer_id") or "")
name_first = not any(
key in form_candidate["target"]
for key in ("table", "file_name", "guid", "form_guid", "module_ref", "extension_guid")
)
checks["name_first_extension_form_preflight"] = {
"schema": form_preflight.get("schema"),
"status": form_preflight.get("status"),
"allowed": form_preflight.get("allowed"),
"plan_status": plan.get("status"),
"plan_allowed": plan.get("allowed"),
"repository_layer_id": repository_layer or None,
"support_layer_id": support_layer or None,
"name_first": name_first,
"extension": form_candidate["target"].get("extension"),
}
require(form_preflight.get("schema") == "onec_metadata_write_preflight.v1", "name-first extension form preflight must return expected schema", failures)
require(classified_preflight_status(form_preflight.get("status")), "name-first extension form preflight must classify readiness or a safety gate", failures)
require(plan.get("status") == "planned" and plan.get("allowed") is True, "name-first extension form preflight plan must be allowed and planned", failures)
require(name_first, "extension form preflight input must remain name-first", failures)
require(bool(EXTENSION_GUID_LAYER_RE.fullmatch(repository_layer)), "repository gate must use a resolved extension GUID layer", failures)
require(repository_layer == support_layer, "repository and support gates must use the same extension layer", failures)
return {
"schema": "onec_write_preflight_smoke.v1",
"status": "ok" if not failures else "failed",
"endpoint_url": endpoint_url,
"transport": transport,
"base_id": base_id,
"requirements": {
"name_first_extension_form": require_name_first_extension_form,
},
"checks": checks,
"failures": failures,
**({"mcp_initialize": initialize_result} if initialize_result and failures else {}),
@@ -253,11 +379,18 @@ def main() -> int:
parser.add_argument("--base-id", default="upo_test")
parser.add_argument("--timeout", type=float, default=30.0)
parser.add_argument("--report", type=Path)
parser.add_argument("--require-name-first-extension-form", action="store_true")
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)
report = run_smoke(
endpoint_url,
args.base_id,
args.timeout,
transport=args.transport,
require_name_first_extension_form=args.require_name_first_extension_form,
)
except (TimeoutError, URLError, OSError) as exc:
report = {
"schema": "onec_write_preflight_smoke.v1",