Compose name-first form write plans

This commit is contained in:
2026-07-26 17:30:04 +03:00
parent c1d5d06f6d
commit 39c8c4ab18
6 changed files with 603 additions and 10 deletions
+170 -4
View File
@@ -41169,14 +41169,29 @@ def metadata_saved_state_forms_search(payload: dict[str, Any]) -> dict[str, Any]
break
if len(rows) >= int(limit or 50):
break
public_rows = rows if include_storage else [
extension_names_by_guid: dict[str, str] = {}
if not include_storage and not extension_filter and any(row.get("table") == "ConfigCASSave" for row in rows):
extension_names_by_guid = {
guid: str(item.get("name") or guid)
for guid, item in extension_map_by_guid(base_id).items()
}
public_rows = rows if include_storage else []
if not include_storage:
for row in rows:
row_extension = extension_filter or None
if not row_extension and row.get("table") == "ConfigCASSave":
row_source = row.get("source") if isinstance(row.get("source"), dict) else {}
row_file_name = str(row.get("file_name") or row_source.get("file_name") or "")
extension_guid_from_file = row_file_name.split("__", 1)[0].strip().lower() if "__" in row_file_name else ""
if extension_guid_from_file:
row_extension = extension_names_by_guid.get(extension_guid_from_file)
public_rows.append(
public_saved_state_form_search_row(
row,
requested_owner_kind=owner_kind,
requested_extension=extension_filter or None,
requested_extension=row_extension,
)
)
for row in rows
]
return {
"schema": "onec_saved_state_form_search.v1",
"status": "ok",
@@ -46265,6 +46280,26 @@ def metadata_write_plan_apply_payload_hint(
hint["payload"]["attribute"] = path_resolution.get("attribute_name")
elif form_member_path:
hint["payload"]["element"] = form_member_path[-1]
for field in (
"extension",
"ref",
"object_ref",
"form",
"form_name",
"command",
"element",
"element_name",
"attribute",
"element_path",
"element_id",
"id",
):
value = target.get(field) if target.get(field) not in (None, "") else payload.get(field)
if value not in (None, ""):
hint["payload"][field] = value
edits = payload.get("edits") if isinstance(payload.get("edits"), list) else intent.get("edits")
if isinstance(edits, list) and edits:
hint["payload"]["edits"] = edits
for field in ("table", "property", "value"):
value = metadata_write_plan_first_value(payload, target, intent, field)
if value is not None:
@@ -46681,6 +46716,115 @@ def metadata_write_plan_resolve_public_module_target(
}
def metadata_write_plan_resolve_public_form_target(
payload: dict[str, Any],
target: dict[str, Any],
intent: dict[str, Any],
) -> tuple[dict[str, Any], dict[str, Any]]:
"""Resolve a public 1C form selector to an internal saved-state handle."""
selector_ref = str(target.get("ref") or target.get("object_ref") or payload.get("ref") or payload.get("object_ref") or "").strip()
selector_kind = str(
target.get("object_kind")
or target.get("object_type")
or payload.get("object_kind")
or payload.get("object_type")
or ""
).strip()
selector_name = str(target.get("object_name") or payload.get("object_name") or "").strip()
form_name = str(target.get("form") or target.get("form_name") or payload.get("form") or payload.get("form_name") or "").strip()
parsed_ref = parse_1c_object_path(selector_ref) if selector_ref else {}
if not selector_kind and parsed_ref.get("kind"):
selector_kind = str(parsed_ref.get("kind") or "")
if not selector_name and parsed_ref.get("name"):
selector_name = str(parsed_ref.get("name") or "")
if not form_name and canonical_kind(selector_kind) == "CommonForm":
form_name = selector_name
extension = str(target.get("extension") or payload.get("extension") or "").strip()
member_selector: dict[str, Any] = {}
for key in ("command", "element", "element_name", "attribute", "element_path", "element_id", "id"):
value = target.get(key) if target.get(key) not in (None, "") else payload.get(key)
if value not in (None, ""):
member_selector[key] = value
public_selector = {
**({"extension": extension} if extension else {}),
**({"ref": selector_ref} if selector_ref else {}),
**({"kind": canonical_kind(selector_kind)} if selector_kind and not selector_ref else {}),
**({"name": selector_name} if selector_name and not selector_ref else {}),
**({"form": form_name} if form_name else {}),
**member_selector,
}
if not (selector_ref or (selector_kind and selector_name)):
return target, {
"status": "not_attempted",
"error": "public_object_selector_missing",
"selector": public_selector,
}
if not form_name and canonical_kind(selector_kind) != "CommonForm":
return target, {
"status": "not_attempted",
"error": "public_form_selector_missing",
"selector": public_selector,
}
first_edit = next(
(item for item in payload.get("edits") or [] if isinstance(item, dict)),
{},
)
property_name = metadata_write_plan_first_value(payload, target, intent, "property")
if property_name in (None, ""):
property_name = first_edit.get("property")
value_present = any(
isinstance(source, dict) and "value" in source
for source in (intent, target, payload, first_edit)
)
property_value = metadata_write_plan_first_value(payload, target, intent, "value")
if "value" in first_edit and property_value is None:
property_value = first_edit.get("value")
resolve_payload: dict[str, Any] = {
"base_id": payload.get("base_id"),
"table": str(target.get("table") or payload.get("table") or ("ConfigCASSave" if extension else "ConfigSave")),
"include_storage": True,
"search_limit": int(payload.get("search_limit") or 10),
"scan_limit": int(payload.get("scan_limit") or 1000),
"timeout_seconds": int(payload.get("timeout_seconds") or 60),
**public_selector,
**({"property": property_name} if property_name not in (None, "") else {}),
**({"value": property_value} if value_present else {}),
}
resolved = metadata_form_write_target_resolve(resolve_payload)
if resolved.get("status") != "ok":
return target, {
"status": str(resolved.get("status") or "error"),
"error": "public_form_target_not_resolved",
"selector": public_selector,
"resolution_status": resolved.get("status"),
**({"counts": resolved.get("counts")} if isinstance(resolved.get("counts"), dict) else {}),
}
source = resolved.get("source") if isinstance(resolved.get("source"), dict) else {}
file_name = str(source.get("file_name") or "").strip()
table = str(source.get("table") or resolve_payload.get("table") or "").strip()
if not file_name or table not in FORM_ELEMENT_SAVED_STATE_TABLES:
return target, {
"status": "error",
"error": "public_form_target_storage_handle_missing",
"selector": public_selector,
}
resolved_target = {
**target,
"table": table,
"file_name": file_name,
}
return resolved_target, {
"status": "resolved",
"method": FORM_WRITE_TARGET_RESOLVE_METHOD,
"selector": public_selector,
"counts": {"matches": int((resolved.get("counts") or {}).get("matches") or 1)},
}
def metadata_write_plan(payload: dict[str, Any]) -> dict[str, Any]:
method = METADATA_WRITE_PLAN_METHOD
target = dict(payload.get("target")) if isinstance(payload.get("target"), dict) else {}
@@ -46695,6 +46839,7 @@ def metadata_write_plan(payload: dict[str, Any]) -> dict[str, Any]:
)
target_kind_hint = str(target_kind_raw or "").strip().casefold()
module_name_resolution: dict[str, Any] | None = None
form_name_resolution: dict[str, Any] | None = None
has_public_module_selector = bool(
target.get("ref")
or target.get("object_ref")
@@ -46706,6 +46851,17 @@ def metadata_write_plan(payload: dict[str, Any]) -> dict[str, Any]:
and has_public_module_selector
):
target, module_name_resolution = metadata_write_plan_resolve_public_module_target(payload, target)
has_public_form_selector = bool(
target.get("ref")
or target.get("object_ref")
or ((target.get("object_kind") or target.get("object_type")) and target.get("object_name"))
) and bool(target.get("form") or target.get("form_name") or target.get("command") or target.get("element") or target.get("attribute"))
if (
target_kind_hint in {"form", "форма"}
and not metadata_write_concrete_reference_info(payload, target)
and has_public_form_selector
):
target, form_name_resolution = metadata_write_plan_resolve_public_form_target(payload, target, intent)
canonical_path = str(
target.get("canonical_path")
@@ -46757,6 +46913,14 @@ def metadata_write_plan(payload: dict[str, Any]) -> dict[str, Any]:
"resolution_status": module_name_resolution.get("status"),
}
)
if form_name_resolution and form_name_resolution.get("status") != "resolved":
problems.append(
{
"code": str(form_name_resolution.get("error") or "public_form_target_not_resolved"),
"message": "The public 1C form selector did not resolve to exactly one saved-state form target.",
"resolution_status": form_name_resolution.get("status"),
}
)
if not path_resolution.get("is_full_path") and not concrete_reference:
problems.append(
{
@@ -46833,6 +46997,8 @@ def metadata_write_plan(payload: dict[str, Any]) -> dict[str, Any]:
}
if module_name_resolution:
route["name_resolution"] = module_name_resolution
elif form_name_resolution:
route["name_resolution"] = form_name_resolution
if extension_action:
route["extension_action"] = extension_action
if operation_was_inferred_from_extension_action:
+28 -1
View File
@@ -146,6 +146,7 @@ def validate_selector_chain(
resolve_overrides = coverage.get("resolve_overrides") if isinstance(coverage.get("resolve_overrides"), dict) else {}
saved_state_resolution = coverage.get("saved_state_resolution") if isinstance(coverage.get("saved_state_resolution"), dict) else {}
composition = coverage.get("write_plan_composition") if isinstance(coverage.get("write_plan_composition"), dict) else {}
form_composition = coverage.get("form_write_plan_composition") if isinstance(coverage.get("form_write_plan_composition"), dict) else {}
skips = coverage.get("skips") if isinstance(coverage.get("skips"), list) else None
summary = {
"path": str(path),
@@ -161,6 +162,9 @@ def validate_selector_chain(
"write_plan_target": saved_state_resolution.get("write_plan_target"),
"composition_status": composition.get("status"),
"composed": composition.get("composed"),
"form_composition_status": form_composition.get("status"),
"form_composed": form_composition.get("composed"),
"form_name_first": form_composition.get("name_first"),
}
if not report:
return summary
@@ -177,7 +181,7 @@ def validate_selector_chain(
)
if report.get("passed") is not True:
failures.append({"code": "selector_chain_not_passed", "label": label, "path": str(path), "issues": report.get("issues")})
for section in ("resolve_overrides", "saved_state_resolution", "write_plan_composition", "skips"):
for section in ("resolve_overrides", "saved_state_resolution", "write_plan_composition", "form_write_plan_composition", "skips"):
if section not in coverage:
failures.append({"code": "selector_chain_coverage_section_missing", "label": label, "section": section, "path": str(path)})
if resolve_overrides.get("attempted") is not True:
@@ -207,10 +211,23 @@ def validate_selector_chain(
})
if composition.get("composed") is True and composition.get("status") in {None, "skipped_no_saved_state_target"}:
failures.append({"code": "selector_chain_composed_status_unexpected", "label": label, "path": str(path), "composition": composition})
if not isinstance(form_composition.get("attempted"), bool):
failures.append({"code": "selector_chain_form_composition_attempted_not_boolean", "label": label, "path": str(path), "composition": form_composition})
if not isinstance(form_composition.get("composed"), bool):
failures.append({"code": "selector_chain_form_composed_not_boolean", "label": label, "path": str(path), "composition": form_composition})
if form_composition.get("candidate") is True and form_composition.get("composed") is not True:
failures.append({"code": "selector_chain_form_target_not_composed", "label": label, "path": str(path), "composition": form_composition})
if form_composition.get("composed") is True and form_composition.get("name_first") is not True:
failures.append({"code": "selector_chain_form_composition_not_name_first", "label": label, "path": str(path), "composition": form_composition})
if skips is None:
failures.append({"code": "selector_chain_skips_not_list", "label": label, "path": str(path), "skips": coverage.get("skips")})
if require_composition and composition.get("composed") is not True:
failures.append({"code": "selector_chain_composition_required", "label": label, "path": str(path), "composition": composition})
if require_composition and (
form_composition.get("composed") is not True
or form_composition.get("name_first") is not True
):
failures.append({"code": "selector_chain_form_composition_required", "label": label, "path": str(path), "composition": form_composition})
steps = report.get("steps") if isinstance(report.get("steps"), list) else []
for step in steps:
if not isinstance(step, dict):
@@ -1445,6 +1462,15 @@ def write_self_test_reports(report_dir: Path, *, base_id: str, composed: bool, s
"composed": composed,
"from_write_plan_target": composed,
},
"form_write_plan_composition": {
"attempted": composed,
"search_attempted": composed,
"search_status": "ok" if composed else None,
"candidate": composed,
"status": "planned" if composed else None,
"composed": composed,
"name_first": composed,
},
"skips": [] if composed else [{"step": "metadata.write.plan", "status": "skipped_no_saved_state_target"}],
},
"steps": [
@@ -1970,6 +1996,7 @@ def run_self_test() -> dict[str, Any]:
strict_codes = {str(failure.get("code")) for failure in strict_skip["failures"]}
for expected in (
"selector_chain_composition_required",
"selector_chain_form_composition_required",
"saved_state_form_write_required",
"saved_state_module_write_required",
):
+2
View File
@@ -169,6 +169,8 @@ def check_adapter_verify_wiring(scripts: list[Path], executable: str) -> list[st
failures.append("verify_1c_adapter_deployment.ps1 must use -SavedStateTable for copy plan and saved-state smoke commands.")
if verify_text.count("--require-write-plan-composition") < 2:
failures.append("verify_1c_adapter_deployment.ps1 must pass --require-write-plan-composition to both REST and MCP selector-chain smoke commands.")
if verify_text.count("--require-form-write-plan-composition") < 2:
failures.append("verify_1c_adapter_deployment.ps1 must pass --require-form-write-plan-composition to both REST and MCP selector-chain smoke commands.")
if "--allow-empty-saved-state" not in verify_text or "if (-not $RequireSavedStateWriteSmoke)" not in verify_text:
failures.append("verify_1c_adapter_deployment.ps1 must allow empty saved-state only when -RequireSavedStateWriteSmoke is not set.")
if '"skipped_no_saved_state_candidate"' not in verify_text:
+154
View File
@@ -126,6 +126,54 @@ def first_write_plan_target_from_saved_state(result: dict[str, Any]) -> dict[str
return None
def first_public_form_write_plan_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("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")):
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
old_value = str(prop.get("value") or "")
normalized_property = str(prop.get("property") or "").strip().casefold()
priority = 0 if normalized_property == "title" else (2 if normalized_property in {"id", "name"} else 1)
candidates.append(
{
"priority": priority,
"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"{old_value} [NAME-FIRST PLAN]",
},
}
)
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 live_coverage_from_steps(steps: list[dict[str, Any]]) -> dict[str, Any]:
def last_step(name: str) -> dict[str, Any]:
for step in reversed(steps):
@@ -136,6 +184,8 @@ def live_coverage_from_steps(steps: list[dict[str, Any]]) -> dict[str, Any]:
resolve = last_step("metadata.resolve_overrides")
saved_state = last_step("metadata.saved_state.modules.search")
write_plan = last_step("metadata.write.plan")
form_search = last_step("metadata.saved_state.forms.search")
form_write_plan = last_step("metadata.write.plan.form")
skips = [
{"step": str(step.get("name")), "status": step.get("status"), **({"reason": step.get("reason")} if step.get("reason") else {})}
for step in steps
@@ -160,6 +210,15 @@ def live_coverage_from_steps(steps: list[dict[str, Any]]) -> dict[str, Any]:
"composed": write_plan.get("allowed") is True,
"from_write_plan_target": bool(write_plan.get("from_write_plan_target")),
},
"form_write_plan_composition": {
"attempted": bool(form_write_plan),
"search_attempted": bool(form_search),
"search_status": form_search.get("status"),
"candidate": bool(form_search.get("write_plan_candidate")),
"status": form_write_plan.get("status"),
"composed": form_write_plan.get("allowed") is True,
"name_first": bool(form_write_plan.get("name_first")),
},
"skips": skips,
}
@@ -175,6 +234,18 @@ def live_write_plan_composition_required_issue(coverage: dict[str, Any]) -> dict
}
def live_form_write_plan_composition_required_issue(coverage: dict[str, Any]) -> dict[str, Any] | None:
composition = coverage.get("form_write_plan_composition") if isinstance(coverage.get("form_write_plan_composition"), dict) else {}
if composition.get("composed") is True and composition.get("name_first") is True:
return None
return {
"code": "live_form_write_plan_composition_required",
"status": composition.get("status"),
"search_status": composition.get("search_status"),
"candidate": composition.get("candidate"),
}
def live_object_summary(*sources: dict[str, Any] | None) -> dict[str, Any]:
summary: dict[str, Any] = {}
for source in sources:
@@ -504,6 +575,7 @@ def build_live_report(
transport: str = "rest",
object_selector: dict[str, Any] | None = None,
require_write_plan_composition: bool = False,
require_form_write_plan_composition: bool = False,
) -> dict[str, Any]:
steps: list[dict[str, Any]] = []
issues: list[dict[str, Any]] = []
@@ -856,11 +928,91 @@ def build_live_report(
if code_read.get("status") not in {"ok", "summary"}:
issues.append({"code": "live_code_read_failed", "status": code_read.get("status"), "diagnostics": code_read.get("diagnostics")})
if require_form_write_plan_composition:
form_search = rpc_call(
endpoint_url,
"metadata.saved_state.forms.search",
{
"base_id": base_id,
"limit": 20,
"scan_limit": 1000,
"timeout_seconds": int(timeout),
},
timeout=timeout,
transport=transport,
session_id=session_id,
)
form_candidate = first_public_form_write_plan_candidate(form_search)
steps.append(
{
"name": "metadata.saved_state.forms.search",
"status": form_search.get("status"),
"forms": int((form_search.get("counts") or {}).get("forms") or 0),
"write_plan_candidate": bool(form_candidate),
"include_storage": False,
}
)
if form_search.get("status") != "ok":
issues.append({"code": "live_saved_state_forms_search_failed", "status": form_search.get("status")})
elif not form_candidate:
steps.append(
{
"name": "metadata.write.plan.form",
"status": "skipped_no_public_form_target",
"allowed": False,
"name_first": False,
}
)
else:
form_plan_payload = {
"base_id": base_id,
"target": form_candidate["target"],
"edits": [form_candidate["edit"]],
"resolve_origin": False,
}
form_plan = rpc_call(
endpoint_url,
"metadata.write.plan",
form_plan_payload,
timeout=timeout,
transport=transport,
session_id=session_id,
)
form_route = form_plan.get("route") if isinstance(form_plan.get("route"), dict) else {}
form_hint = form_route.get("apply_payload_hint") if isinstance(form_route.get("apply_payload_hint"), dict) else {}
form_name_resolution = form_route.get("name_resolution") if isinstance(form_route.get("name_resolution"), dict) else {}
steps.append(
{
"name": "metadata.write.plan.form",
"status": form_plan.get("status"),
"allowed": form_plan.get("allowed"),
"apply_method": form_route.get("apply_method"),
"ready_for_apply_method": form_hint.get("ready_for_apply_method"),
"name_resolution": form_name_resolution.get("status"),
"name_first": not any(
key in form_plan_payload["target"]
for key in ("table", "file_name", "guid", "form_guid", "module_ref")
),
}
)
if form_plan.get("allowed") is not True:
issues.append({"code": "live_form_write_plan_not_allowed", "status": form_plan.get("status"), "problems": form_plan.get("problems")})
if form_route.get("apply_method") != "metadata.form.element.write_apply":
issues.append({"code": "live_form_write_plan_apply_method_mismatch", "actual": form_route.get("apply_method")})
if form_hint.get("ready_for_apply_method") is not True:
issues.append({"code": "live_form_write_plan_hint_not_ready", "hint": form_hint})
if form_name_resolution.get("status") != "resolved":
issues.append({"code": "live_form_write_plan_name_resolution_missing", "name_resolution": form_name_resolution})
coverage = live_coverage_from_steps(steps)
if require_write_plan_composition:
strict_issue = live_write_plan_composition_required_issue(coverage)
if strict_issue:
issues.append(strict_issue)
if require_form_write_plan_composition:
strict_form_issue = live_form_write_plan_composition_required_issue(coverage)
if strict_form_issue:
issues.append(strict_form_issue)
return {
"schema": "onec_mcp_selector_chain_live_smoke.v1",
@@ -1037,6 +1189,7 @@ def main() -> int:
parser.add_argument("--guid", help="Optional concrete metadata object GUID for deterministic --live checks.")
parser.add_argument("--timeout", type=float, default=15.0, help="HTTP timeout for --live.")
parser.add_argument("--require-write-plan-composition", action="store_true", help="Fail live smoke when saved-state lookup does not produce a concrete write_plan_target and composed metadata.write.plan.")
parser.add_argument("--require-form-write-plan-composition", action="store_true", help="Fail live smoke unless a public saved-state form selector composes metadata.write.plan without storage identifiers.")
args = parser.parse_args()
if args.live and not args.base_id:
@@ -1051,6 +1204,7 @@ def main() -> int:
transport=args.transport,
object_selector=object_selector or None,
require_write_plan_composition=args.require_write_plan_composition,
require_form_write_plan_composition=args.require_form_write_plan_composition,
)
else:
report = build_report()
+9 -1
View File
@@ -147,7 +147,7 @@ function Assert-SelectorChainReport {
if (-not $report.coverage) {
throw "$Label report is missing coverage: $Path"
}
foreach ($section in @("resolve_overrides", "saved_state_resolution", "write_plan_composition", "skips")) {
foreach ($section in @("resolve_overrides", "saved_state_resolution", "write_plan_composition", "form_write_plan_composition", "skips")) {
if ($report.coverage.PSObject.Properties.Name -notcontains $section) {
throw "$Label report coverage is missing '$section': $Path"
}
@@ -155,6 +155,12 @@ function Assert-SelectorChainReport {
if ($RequireComposition -and $report.coverage.write_plan_composition.composed -ne $true) {
throw "$Label report did not compose metadata.write.plan in strict mode: $Path"
}
if ($RequireComposition -and (
$report.coverage.form_write_plan_composition.composed -ne $true -or
$report.coverage.form_write_plan_composition.name_first -ne $true
)) {
throw "$Label report did not compose a name-first form metadata.write.plan in strict mode: $Path"
}
foreach ($step in @($report.steps)) {
if ($step.name -in @("metadata.resolve_overrides", "code.search") -and $step.status -notlike "skipped*") {
if ($step.working_state -ne "working") {
@@ -441,6 +447,7 @@ try {
)
if ($RequireSelectorChainWritePlanComposition) {
$restCommand += "--require-write-plan-composition"
$restCommand += "--require-form-write-plan-composition"
}
Invoke-CheckedCommand -Label "REST adapter selector-chain live smoke ($currentBaseId)" -Command (Add-ObjectSelectorArgs -Command $restCommand)
Assert-SelectorChainReport -Label "REST adapter selector-chain live smoke ($currentBaseId)" -Path $selectorChainReport -RequireComposition:$RequireSelectorChainWritePlanComposition
@@ -736,6 +743,7 @@ try {
)
if ($RequireSelectorChainWritePlanComposition) {
$mcpCommand += "--require-write-plan-composition"
$mcpCommand += "--require-form-write-plan-composition"
}
Invoke-CheckedCommand -Label "MCP proxy selector-chain live smoke ($currentBaseId)" -Command (Add-ObjectSelectorArgs -Command $mcpCommand)
Assert-SelectorChainReport -Label "MCP proxy selector-chain live smoke ($currentBaseId)" -Path $selectorChainReport -RequireComposition:$RequireSelectorChainWritePlanComposition
+236
View File
@@ -3129,6 +3129,67 @@ def test_saved_state_forms_search_default_is_name_first(monkeypatch: pytest.Monk
assert storage_marker not in str(result)
def test_saved_state_forms_search_infers_public_extension_name_without_filter(monkeypatch: pytest.MonkeyPatch) -> None:
extension_guid = "fb26cf42-7609-11f1-828f-005056b0d483"
file_name = f"{extension_guid}__form-guid.0"
monkeypatch.setattr(
adapter_server,
"extension_map_by_guid",
lambda base_id: {extension_guid: {"guid": extension_guid, "name": "test2"}},
)
monkeypatch.setattr(
adapter_server,
"storage_files_list",
lambda payload: {"status": "ok", "files": [{"FileName": file_name}]},
)
monkeypatch.setattr(
adapter_server,
"saved_state_form_search_row",
lambda **kwargs: {
"table": "ConfigCASSave",
"file_name": file_name,
"name": "t_Форма",
"owner": {"status": "resolved", "kind": "Catalog", "name": "test2"},
"form": {"name": "t_Форма"},
"source": {"table": "ConfigCASSave", "file_name": file_name},
"matches": [
{
"name": "ЗаменаДомена",
"section": "commands",
"writable_properties": [
{
"property": "title",
"presentation": "Заголовок",
"value": "Замена домена",
"value_type": "string",
}
],
}
],
},
)
result = adapter_server.metadata_saved_state_forms_search(
{
"base_id": "upo_test",
"tables": ["ConfigCASSave"],
"limit": 1,
}
)
form = result["forms"][0]
assert form["selector"]["extension"] == "test2"
assert form["matches"][0]["selector"] == {
"extension": "test2",
"ref": "Catalog.test2",
"form": "t_Форма",
"qualified_name": "test2.t_Форма",
"command": "ЗаменаДомена",
}
assert extension_guid not in str(result)
assert "ConfigCASSave" not in str(result["forms"])
def test_saved_state_form_search_resolves_owner_for_internal_name_first_write(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(
adapter_server,
@@ -14015,6 +14076,87 @@ def test_mcp_selector_chain_smoke_uses_generic_placeholders() -> None:
assert concrete_value not in examples_json
def test_mcp_selector_chain_selects_name_first_form_write_plan_candidate() -> None:
candidate = mcp_selector_chain_smoke.first_public_form_write_plan_candidate(
{
"forms": [
{
"selector": {
"extension": "test2",
"ref": "Catalog.test2",
"form": "t_Форма",
},
"matches": [
{
"selector": {
"extension": "test2",
"ref": "Catalog.test2",
"form": "t_Форма",
"command": "ЗаменаДомена",
},
"writable_properties": [
{
"property": "name",
"presentation": "Имя",
"value": "ЗаменаДомена",
"value_type": "string",
},
{
"property": "title",
"presentation": "Заголовок",
"value": "Замена домена",
"value_type": "string",
},
],
}
],
}
]
}
)
assert candidate == {
"target": {
"kind": "form",
"extension": "test2",
"ref": "Catalog.test2",
"form": "t_Форма",
"command": "ЗаменаДомена",
},
"edit": {
"property": "Заголовок",
"value": "Замена домена [NAME-FIRST PLAN]",
},
}
assert not any(key in candidate["target"] for key in ("table", "file_name", "guid", "form_guid"))
def test_mcp_selector_chain_form_composition_strict_check() -> None:
missing = mcp_selector_chain_smoke.live_form_write_plan_composition_required_issue(
{
"form_write_plan_composition": {
"attempted": True,
"status": "skipped_no_public_form_target",
"composed": False,
"name_first": False,
}
}
)
ready = mcp_selector_chain_smoke.live_form_write_plan_composition_required_issue(
{
"form_write_plan_composition": {
"attempted": True,
"status": "planned",
"composed": True,
"name_first": True,
}
}
)
assert missing["code"] == "live_form_write_plan_composition_required"
assert ready is None
def test_mcp_selector_chain_smoke_rejects_concrete_selector_values() -> None:
issues = mcp_selector_chain_smoke.validate_chain(
{
@@ -21551,6 +21693,100 @@ def test_metadata_write_plan_resolves_public_module_selector_internally(monkeypa
assert hint["payload"]["expected_sha1"] == "saved-payload-sha1"
def test_metadata_write_plan_resolves_public_form_selector_internally(monkeypatch: pytest.MonkeyPatch) -> None:
seen: dict[str, Any] = {}
def fake_resolve(payload: dict[str, Any]) -> dict[str, Any]:
seen.update(payload)
return {
"schema": "onec_form_write_target_resolution.v1",
"status": "ok",
"source": {
"kind": "live_sql",
"table": "ConfigCASSave",
"file_name": "extension-guid__form-guid.0",
},
"counts": {"matches": 1},
}
monkeypatch.setattr(adapter_server, "metadata_form_write_target_resolve", fake_resolve)
result = adapter_server.metadata_write_plan(
{
"base_id": "upo_test",
"target": {
"kind": "form",
"extension": "test2",
"ref": "Catalog.test2",
"form": "t_Форма",
"command": "ЗаменаДомена",
},
"edits": [{"property": "Заголовок", "value": "Новое имя"}],
"resolve_origin": False,
}
)
assert seen["table"] == "ConfigCASSave"
assert seen["ref"] == "Catalog.test2"
assert seen["form"] == "t_Форма"
assert seen["command"] == "ЗаменаДомена"
assert seen["property"] == "Заголовок"
assert seen["value"] == "Новое имя"
assert seen["include_storage"] is True
assert result["allowed"] is True
assert result["status"] == "planned"
assert result["route"]["name_resolution"] == {
"status": "resolved",
"method": "metadata.form.write_target.resolve",
"selector": {
"extension": "test2",
"ref": "Catalog.test2",
"form": "t_Форма",
"command": "ЗаменаДомена",
},
"counts": {"matches": 1},
}
hint = result["route"]["apply_payload_hint"]
assert hint["ready_for_apply_method"] is True
assert hint["payload"]["table"] == "ConfigCASSave"
assert hint["payload"]["file_name"] == "extension-guid__form-guid.0"
assert hint["payload"]["ref"] == "Catalog.test2"
assert hint["payload"]["form"] == "t_Форма"
assert hint["payload"]["command"] == "ЗаменаДомена"
assert hint["payload"]["edits"] == [{"property": "Заголовок", "value": "Новое имя"}]
def test_metadata_write_plan_reports_unresolved_public_form_selector(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(
adapter_server,
"metadata_form_write_target_resolve",
lambda payload: {
"schema": "onec_form_write_target_resolution.v1",
"status": "not_found",
"counts": {"matches": 0},
},
)
result = adapter_server.metadata_write_plan(
{
"base_id": "upo_test",
"target": {
"kind": "form",
"ref": "Catalog.Контрагенты",
"form": "ФормаЭлемента",
"element": "КнопкаЗаписать",
},
"edits": [{"property": "Заголовок", "value": "Записать"}],
"resolve_origin": False,
}
)
assert result["allowed"] is False
assert result["status"] == "needs_origin"
assert result["route"]["name_resolution"]["status"] == "not_found"
assert any(problem["code"] == "public_form_target_not_resolved" for problem in result["problems"])
def test_metadata_write_preflight_verifies_concrete_saved_state(monkeypatch: pytest.MonkeyPatch) -> None:
def fake_diff(payload: dict[str, Any]) -> dict[str, Any]:
assert payload["module_ref"] == "ConfigCASSave:object-guid__module-guid.0#stream:4"