service_key: support editable-only API endpoints

plugins/action/base_action.py

@@ -359,6 +359,19 @@ def _pre_execute_hook(self, ansible_data: dict, write_only_data: dict, validated
             operation: The resolved operation string.
         """
 
+    def _handle_operation_error(self, error: Exception, operation: str, result: dict) -> bool:
+        """Handle a resource-specific operation failure.
+
+        Return ``True`` after updating *result* to convert an expected API
+        limitation into a successful module result.  The default leaves all
+        errors to the standard action-plugin failure path.
+        """
+        return False
+
+    def _skip_operation(self, operation: str, result: dict) -> bool:
+        """Optionally complete an operation without sending an API request."""
+        return False
+
     def _get_or_spawn_manager(self, task_vars: dict) -> Tuple[Union["DirectHTTPClient", "ManagerRPCClient"], Optional[Dict[str, Any]]]:
         """
         Dispatcher: Get connection client from the connection plugin.
@@ -1201,11 +1214,14 @@ def run(self, tmp: object = None, task_vars: Optional[dict] = None) -> dict:
                 else:
                     operation = "create"
 
-            # ---- check mode ------------------------------------------------
+            # ---- unsupported operation / check mode -----------------------
             ansible_data = self._build_ansible_data(resource, validated_params, operation)
             if operation == "update" and state == "enforced":
                 ansible_data["_platform_enforced"] = True
 
+            if self._skip_operation(operation, result):
+                return result
+
             if self._task.check_mode and operation in ("create", "update", "delete"):
                 if operation == "delete":
                     result.update(
@@ -1236,7 +1252,9 @@ def run(self, tmp: object = None, task_vars: Optional[dict] = None) -> dict:
                     module_name=self.MODULE_NAME,
                     ansible_data=ansible_data,
                 )
-            except ValueError as exc:
+            except Exception as exc:
+                if self._handle_operation_error(exc, operation, result):
+                    return result
                 if operation == "find" and ("not found" in str(exc).lower() or "resource with" in str(exc).lower()):
                     result.update(
                         {

plugins/action/service_key.py

@@ -5,6 +5,8 @@
 from __future__ import absolute_import, division, print_function
 
 __metaclass__ = type
+from dataclasses import asdict
+
 from ansible_collections.ansible.platform.plugins.action.base_action import BaseResourceActionPlugin
 from ansible_collections.ansible.platform.plugins.plugin_utils.ansible_models.service_key import AnsibleServiceKey
 
@@ -14,29 +16,91 @@ class ActionModule(BaseResourceActionPlugin):
     MODEL_CLASS = AnsibleServiceKey
     # mark_previous_inactive: operation-time directive; API never returns it.
     # secret: write-only; API returns null/hash, not the original value.
-    # Including either in _should_update() causes false positives.
     _WRITE_ONLY_FIELDS = frozenset({"mark_previous_inactive", "secret"})
+    _CREATE_ONLY_FIELDS = frozenset({"algorithm", "mark_previous_inactive", "secret", "secret_length", "service_cluster"})
+
+    def run(self, tmp=None, task_vars=None):
+        """Emit a deprecation warning when create-only fields are supplied."""
+        supplied_fields = sorted(field for field in self._CREATE_ONLY_FIELDS if self._task.args.get(field) is not None)
+        result = super().run(tmp, task_vars)
+        if supplied_fields:
+            result.setdefault("deprecations", []).append(
+                {
+                    "msg": "The service_key parameters %s are deprecated; newer AAP versions ignore them for existing service keys."
+                    % ", ".join(supplied_fields),
+                    "version": "4.0.0",
+                    "collection_name": "ansible.platform",
+                }
+            )
+        return result
 
     def _pre_execute_hook(self, ansible_data, write_only_data, validated_params, operation):
         """Re-inject write-only fields so they reach the API payload.
 
-        ``mark_previous_inactive`` and ``secret`` are excluded from the
-        AnsibleServiceKey dataclass (via _WRITE_ONLY_FIELDS) to prevent
-        false-positive idempotency checks — the API never echoes these
-        fields back in GET responses, so _should_update() would always
-        see None vs. a user-supplied value and report changed.
-
-        For create/update operations however, both fields must still reach
-        the transform and ultimately the API request body.  This hook puts
-        them back into ansible_data (from the write_only_data stash) so
-        the transform can include them when they are non-None.
-
-        Note: mark_previous_inactive=False is a valid explicit value and
-        must not be filtered out here — only skip genuinely absent (None)
-        values.
+        Older AAP versions accept all fields on update. Newer versions
+        advertise no POST action and only allow name/is_active changes.
         """
         if operation in ("create", "update"):
-            for field in ("mark_previous_inactive", "secret"):
+            for field in self._WRITE_ONLY_FIELDS:
                 val = write_only_data.get(field)
                 if val is not None:
                     ansible_data[field] = val
+
+        if operation == "update" and not self._supports_create():
+            for field in self._CREATE_ONLY_FIELDS:
+                ansible_data.pop(field, None)
+
+    def _supports_create(self):
+        """Use OPTIONS to distinguish legacy and editable-only APIs."""
+        return self._client.endpoint_supports_method("/api/gateway/v1/service_keys/", "POST")
+
+    def _should_update(self, desired_data, current_data):
+        """Compare all fields on legacy AAP, editable fields on newer AAP."""
+        if self._supports_create():
+            return super()._should_update(desired_data, current_data)
+
+        ignored_fields = [field for field in self._CREATE_ONLY_FIELDS if self._task.args.get(field) is not None]
+        if ignored_fields:
+            self._display.warning("The AAP instance will ignore: %s when making this request." % ", ".join(sorted(ignored_fields)))
+        editable_data = {key: value for key, value in desired_data.items() if key not in self._CREATE_ONLY_FIELDS}
+        return super()._should_update(editable_data, current_data)
+
+    def _skip_operation(self, operation, result):
+        """Avoid POST when OPTIONS reports that the endpoint is read-only."""
+        if operation == "create" and not self._supports_create():
+            return self._warn_create_unsupported(result)
+        return False
+
+    def _handle_operation_error(self, error, operation, result):
+        """Warn when newer AAP versions reject service-key creation."""
+        error_text = str(error).lower()
+        status_code = getattr(error, "status_code", None)
+        if status_code is None:
+            status_code = getattr(error, "details", {}).get("status_code")
+        if operation != "create" or (status_code != 405 and "405" not in error_text):
+            return False
+
+        return self._warn_create_unsupported(result)
+
+    def _warn_create_unsupported(self, result):
+        """Return a stable no-op result for an editable-only AAP instance."""
+
+        warning = "This version of AAP does not support creating service_keys through the API."
+        # Match the normal resource result shape without implying that AAP
+        # created the requested key.  ``state`` is an input directive, not a
+        # returned resource attribute.
+        service_key = asdict(AnsibleServiceKey(name=self._task.args.get("name")))
+        service_key.pop("state")
+        self._display.warning(warning)
+        result.update(
+            {
+                "changed": False,
+                "failed": False,
+                self.MODULE_NAME: service_key,
+                # Keep legacy flat result access stable for follow-up tasks.
+                **service_key,
+                "msg": warning,
+                "warnings": [warning],
+            }
+        )
+        return True

plugins/plugin_utils/manager/platform_manager.py

@@ -483,6 +483,23 @@ def _build_url(self, endpoint: str, query_params: Optional[Dict] = None) -> str:
 
         return url
 
+    def endpoint_supports_method(self, path: str, method: str) -> bool:
+        """Return whether an endpoint advertises an HTTP method via OPTIONS."""
+        method = method.upper()
+        cache_key = f"endpoint-method:{path}:{method}"
+        if cache_key in self.cache:
+            return self.cache[cache_key]
+
+        response = self.session.options(self._build_url(path), timeout=self.request_timeout, verify=self.requests_verify)
+        response.raise_for_status()
+        allowed = response.headers.get("Allow", "")
+        if allowed:
+            supported = method in {item.strip().upper() for item in allowed.split(",")}
+        else:
+            supported = method in response.json().get("actions", {})
+        self.cache[cache_key] = supported
+        return supported
+
     def execute(self, operation: str, module_name: str, ansible_data_dict: dict) -> dict:
         """
         Execute a generic operation on any resource.
@@ -660,7 +677,7 @@ def _update_resource(self, ansible_data: Any, mixin_class: type, context: dict)
                 lookup_field = mixin_class.get_lookup_field()
                 api_normalized_fields = {"slug"}
                 internal_fields = {"organization_id"}
-                skip_fields = read_only_fields | {"state", lookup_field} | api_normalized_fields | internal_fields
+                skip_fields = read_only_fields | {"state", "new_name", lookup_field} | api_normalized_fields | internal_fields
                 requested = asdict(ansible_data)
                 for k, v in requested.items():
                     if k in skip_fields or v is None:
@@ -703,7 +720,7 @@ def _update_resource(self, ansible_data: Any, mixin_class: type, context: dict)
             internal_fields = {"organization_id"}
             norm = self._normalize_for_compare
             lookup_field = mixin_class.get_lookup_field()
-            skip_fields = read_only_fields | {"state", lookup_field} | api_normalized_fields | internal_fields
+            skip_fields = read_only_fields | {"state", "new_name", lookup_field} | api_normalized_fields | internal_fields
             requested = asdict(ansible_data)
             changed = False
             for k, v in requested.items():

plugins/plugin_utils/manager/rpc_client.py

@@ -112,6 +112,10 @@ def lookup_resource_id(self, endpoint: str, lookup_field: str, lookup_value: str
         """
         return self.service_proxy.lookup_resource_id(endpoint, lookup_field, lookup_value)
 
+    def endpoint_supports_method(self, path: str, method: str) -> bool:
+        """Ask the manager whether an endpoint advertises an HTTP method."""
+        return self.service_proxy.endpoint_supports_method(path, method)
+
     def search_api(self, endpoint: str, query_params: Optional[dict] = None, return_all: bool = False, max_objects: int = 1000) -> dict:
         """
         Execute a raw GET via the manager subprocess and return the JSON response.

plugins/plugin_utils/platform/direct_client.py

@@ -1018,3 +1018,23 @@ def direct_request(self, method: str, path: str, data=None) -> dict:
             return json.loads(response_body) if response_body else {}
         except Exception:
             return {}
+
+    def endpoint_supports_method(self, path: str, method: str) -> bool:
+        """Return whether an endpoint advertises an HTTP method via OPTIONS."""
+        method = method.upper()
+        cache_key = f"endpoint-method:{path}:{method}"
+        if cache_key in self.cache:
+            return self.cache[cache_key]
+
+        response = self._make_request("OPTIONS", self._build_url(path), operation="options", resource=path)
+        allowed = getattr(response, "headers", {}).get("Allow", "")
+        if allowed:
+            supported = method in {item.strip().upper() for item in allowed.split(",")}
+        else:
+            try:
+                response_data = json.loads(response.read() or "{}")
+            except (TypeError, ValueError):
+                response_data = {}
+            supported = method in response_data.get("actions", {})
+        self.cache[cache_key] = supported
+        return supported