diff --git a/plugins/1c/connector/adapter_1c_server.py b/plugins/1c/connector/adapter_1c_server.py index 9c7e75b..4978fef 100644 --- a/plugins/1c/connector/adapter_1c_server.py +++ b/plugins/1c/connector/adapter_1c_server.py @@ -54673,16 +54673,92 @@ _FORM_MODULE_FILE_RE = re.compile(r"^(?:(?P[0-9a-f]{8}-[0-9a-f]{4}-[0 def resolve_write_gate_context(payload: dict[str, Any]) -> dict[str, Any]: - """Derive exact form ownership for a physical module reference before gates. + """Derive the exact development layer and form ownership before gates. This is read-only and deliberately does *not* select a lock session. The caller must still present the lock_session_id returned by confirmation. """ result = dict(payload) + target = dict(payload.get("target") or {}) if isinstance(payload.get("target"), dict) else {} + origin = target.get("origin") if isinstance(target.get("origin"), dict) else {} + origin_extension = origin.get("extension") if isinstance(origin.get("extension"), dict) else {} + explicit_extension_guid = str( + payload.get("extension_guid") + or target.get("extension_guid") + or origin_extension.get("guid") + or "" + ).strip().lower() + requested_extension_value = ( + target.get("extension") + or payload.get("extension") + or target.get("preferred_extension") + or payload.get("preferred_extension") + or "" + ) + if isinstance(requested_extension_value, dict): + explicit_extension_guid = str( + explicit_extension_guid or requested_extension_value.get("guid") or "" + ).strip().lower() + requested_extension = str( + requested_extension_value.get("name") or requested_extension_value.get("guid") or "" + ).strip() + else: + requested_extension = str(requested_extension_value or "").strip() + base_id = str(payload.get("base_id") or "").strip() + resolved_extension_guid = explicit_extension_guid + extension_error: dict[str, Any] | None = None + if not resolved_extension_guid and requested_extension and base_id: + resolved_extension_guid, extension_error = extension_filter_to_guid( + base_id, + requested_extension, + method="write.gate_context", + ) + if resolved_extension_guid: + resolved_extension_guid = resolved_extension_guid.lower() + result["extension_guid"] = resolved_extension_guid + target["extension_guid"] = resolved_extension_guid + target["origin"] = { + **origin, + "source": "extension", + "status": "ok", + "extension": { + **origin_extension, + "guid": resolved_extension_guid, + **({"name": requested_extension} if requested_extension and not is_guid_text(requested_extension) else {}), + }, + } + result["development_layer_resolution"] = { + "status": "resolved", + "source": "explicit_extension_guid" if explicit_extension_guid else "extension_name", + "extension": { + "guid": resolved_extension_guid, + **({"name": requested_extension} if requested_extension and not is_guid_text(requested_extension) else {}), + }, + "layer_id": f"extension:{resolved_extension_guid}", + } + elif requested_extension: + # Never let an unresolved public extension selector inherit the base + # repository/support policy. The synthetic layer cannot match a + # configured GUID layer and therefore keeps every write gate closed. + result["layer_id"] = "extension:unresolved" + target["layer_id"] = "extension:unresolved" + result["development_layer_resolution"] = { + "status": "not_resolved", + "source": "extension_name", + "extension": {"name": requested_extension}, + "layer_id": "extension:unresolved", + "diagnostics": extension_error + or { + "status": "not_resolved", + "error": "extension_resolution_requires_base_id", + "base_id": base_id or None, + }, + } + if target: + result["target"] = target existing = payload.get("owner_resolution") if isinstance(payload.get("owner_resolution"), dict) else {} if existing.get("status") == "resolved" and str(existing.get("repository_object") or "").strip(): return result - target = dict(payload.get("target") or {}) if isinstance(payload.get("target"), dict) else {} module_ref = str( target.get("module_ref") or target.get("module_id") or payload.get("module_ref") or payload.get("module_id") or "" ).strip() @@ -54690,7 +54766,6 @@ def resolve_write_gate_context(payload: dict[str, Any]) -> dict[str, Any]: match = _FORM_MODULE_FILE_RE.fullmatch(file_name or "") if table and file_name else None if not match: return result - base_id = str(payload.get("base_id") or "").strip() if not base_id: return result form_guid = str(match.group("form") or "").lower() @@ -54712,13 +54787,35 @@ def resolve_write_gate_context(payload: dict[str, Any]) -> dict[str, Any]: result["support_object_guid"] = form_guid if resolved_extension_guid: result["extension_guid"] = resolved_extension_guid + result.pop("layer_id", None) + target["extension_guid"] = resolved_extension_guid + target.pop("layer_id", None) origin = target.get("origin") if isinstance(target.get("origin"), dict) else {} - target["origin"] = {**origin, "extension": {"guid": resolved_extension_guid}} + previous_extension = origin.get("extension") if isinstance(origin.get("extension"), dict) else {} + target["origin"] = { + **origin, + "source": "extension", + "status": "ok", + "extension": { + "guid": resolved_extension_guid, + **( + {"name": previous_extension.get("name")} + if str(previous_extension.get("guid") or "").strip().lower() == resolved_extension_guid + and str(previous_extension.get("name") or "").strip() + else {} + ), + }, + } else: result.pop("extension_guid", None) + result.pop("layer_id", None) + target.pop("extension_guid", None) + target.pop("layer_id", None) if isinstance(target.get("origin"), dict): origin = dict(target["origin"]) origin.pop("extension", None) + origin["source"] = "configuration" + origin["status"] = "ok" target["origin"] = origin if target: result["target"] = target @@ -54728,6 +54825,12 @@ def resolve_write_gate_context(payload: dict[str, Any]) -> dict[str, Any]: "repository_object": canonical_object, "owner": resolved.get("owner"), "layer_id": f"extension:{resolved_extension_guid}" if resolved_extension_guid else "base", } + result["development_layer_resolution"] = { + "status": "resolved", + "source": "live_sql_form_guid", + "extension": {"guid": resolved_extension_guid} if resolved_extension_guid else None, + "layer_id": f"extension:{resolved_extension_guid}" if resolved_extension_guid else "base", + } return result diff --git a/tests/1c/test_payload_codec.py b/tests/1c/test_payload_codec.py index 3afe423..f3f5b28 100644 --- a/tests/1c/test_payload_codec.py +++ b/tests/1c/test_payload_codec.py @@ -21787,6 +21787,147 @@ def test_metadata_write_plan_reports_unresolved_public_form_selector(monkeypatch assert any(problem["code"] == "public_form_target_not_resolved" for problem in result["problems"]) +def test_write_gate_context_resolves_public_extension_name_to_guid(monkeypatch: pytest.MonkeyPatch) -> None: + extension_guid = "fb26cf42-7609-11f1-828f-005056b0d483" + monkeypatch.setattr( + adapter_server, + "extension_filter_to_guid", + lambda base_id, extension, method: ( + extension_guid, + None, + ) + if (base_id, extension, method) == ("upo_test", "test2", "write.gate_context") + else (None, {"status": "not_found"}), + ) + + result = adapter_server.resolve_write_gate_context( + { + "base_id": "upo_test", + "target": { + "kind": "form", + "extension": "test2", + "ref": "Catalog.test2", + "form": "t_Форма", + "command": "ЗаменаДомена", + }, + } + ) + + assert result["extension_guid"] == extension_guid + assert result["target"]["extension"] == "test2" + assert result["target"]["extension_guid"] == extension_guid + assert result["target"]["origin"]["extension"] == { + "guid": extension_guid, + "name": "test2", + } + assert result["target"]["origin"]["source"] == "extension" + assert result["target"]["origin"]["status"] == "ok" + assert result["development_layer_resolution"]["source"] == "extension_name" + assert adapter_server.repository_control.development_layer_id(result) == f"extension:{extension_guid}" + + +def test_write_gate_context_does_not_fall_back_to_base_for_unknown_extension( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setattr( + adapter_server, + "extension_filter_to_guid", + lambda *_args, **_kwargs: ( + None, + {"status": "not_found", "error": "extension_not_found"}, + ), + ) + + result = adapter_server.resolve_write_gate_context( + { + "base_id": "upo_test", + "target": { + "kind": "form", + "extension": "missing-extension", + "ref": "Catalog.test2", + "form": "t_Форма", + }, + } + ) + + assert "extension_guid" not in result + assert result["development_layer_resolution"]["status"] == "not_resolved" + assert result["development_layer_resolution"]["diagnostics"]["error"] == "extension_not_found" + assert adapter_server.repository_control.development_layer_id(result) == "extension:unresolved" + + +def test_metadata_write_preflight_routes_public_extension_form_gates_to_extension_layer( + monkeypatch: pytest.MonkeyPatch, +) -> None: + extension_guid = "fb26cf42-7609-11f1-828f-005056b0d483" + captured_layers: list[str] = [] + monkeypatch.setattr( + adapter_server, + "extension_filter_to_guid", + lambda base_id, extension, method: (extension_guid, None), + ) + monkeypatch.setattr( + adapter_server, + "metadata_form_write_target_resolve", + lambda payload: { + "schema": "onec_form_write_target_resolution.v1", + "status": "ok", + "source": { + "kind": "live_sql", + "table": "ConfigCASSave", + "file_name": f"{extension_guid}__form-guid.0", + }, + "counts": {"matches": 1}, + }, + ) + monkeypatch.setattr( + adapter_server, + "metadata_saved_state_diff", + lambda payload: { + "status": "changed", + "target": { + "table": payload["table"], + "file_name": payload["file_name"], + }, + "freshness": { + "source": "live_sql", + "status": "live_sql_verified", + "verified_against_sql": True, + }, + }, + ) + + def fake_gate(payload: dict[str, Any]) -> dict[str, Any]: + layer_id = adapter_server.repository_control.development_layer_id(payload) + captured_layers.append(layer_id) + return {"status": "ok", "allowed": True, "layer_id": layer_id} + + monkeypatch.setattr(adapter_server.repository_control, "write_gate", fake_gate) + monkeypatch.setattr(adapter_server, "live_sql_support_gate", fake_gate) + + result = adapter_server.metadata_write_preflight( + { + "base_id": "upo_test", + "target": { + "kind": "form", + "extension": "test2", + "ref": "Catalog.test2", + "form": "t_Форма", + "command": "ЗаменаДомена", + }, + "edits": [{"property": "Заголовок", "value": "Новое имя"}], + "resolve_origin": False, + } + ) + + expected_layer = f"extension:{extension_guid}" + assert result["status"] == "ready" + assert result["allowed"] is True + assert result["repository"]["layer_id"] == expected_layer + assert result["support"]["layer_id"] == expected_layer + assert captured_layers == [expected_layer, expected_layer] + + def test_metadata_write_plan_help_describes_name_first_form_and_module_targets() -> None: methods = { row["name"]: row