Preserve technical failures in capability checks

Author: marcofortina

src/automax/plugins/capabilities.py

@@ -110,24 +110,31 @@ def manual_commands(self, params: Dict[str, Any], context: ExecutionContext) ->
 
     def execute(self, params: Dict[str, Any], context: ExecutionContext) -> PluginResult:
         missing: list[str] = []
+        errors: list[dict[str, Any]] = []
         stdout_parts: list[str] = []
         stderr_parts: list[str] = []
-        for command in self.manual_commands(params, context):
+        checked = self.manual_commands(params, context)
+        for command in checked:
             rc, out, err = exec_remote(context, command)
             stdout_parts.append(out)
             stderr_parts.append(err)
-            if rc != 0:
+            if rc == 1:
                 missing.append(command)
+            elif rc != 0:
+                errors.append({"command": command, "rc": rc, "stderr": err})
         stdout = "\n".join(part for part in stdout_parts if part)
         stderr = "\n".join(part for part in stderr_parts if part)
-        checked = self.manual_commands(params, context)
-        return PluginResult.success(
-            changed=False,
-            rc=0,
-            stdout=stdout,
-            stderr=stderr,
-            data={"checked": checked, "failed": missing, "matches": not missing},
-        )
+        data = {"checked": checked, "failed": missing, "matches": not missing and not errors}
+        if errors:
+            data["errors"] = errors
+            return PluginResult.failure(
+                rc=errors[0]["rc"],
+                stdout=stdout,
+                stderr=stderr,
+                message="os.capability.check failed with a technical command error",
+                data=data,
+            )
+        return PluginResult.success(changed=False, rc=0, stdout=stdout, stderr=stderr, data=data)
 
 
 class PluginRequirementsPlugin(BasePlugin):

tests/test_next_engine.py

@@ -5707,6 +5707,19 @@ def test_os_check_plugins_return_predicates_on_condition_false():
         assert result.rc == 0
         assert result.data[key] is False
 
+def test_capability_check_preserves_technical_failures():
+    registry = AutomaxEngine().plugin_registry
+
+    result = registry.get("os.capability.check").execute(
+        {"commands": ["sh -c 'exit 2'"]},
+        _remote_context_for_result(2, stderr="syntax error"),
+    )
+
+    assert result.ok is False
+    assert result.rc == 2
+    assert result.data["matches"] is False
+    assert result.data["errors"][0]["rc"] == 2
+
 def test_data_archive_and_compression_checks_return_predicates_on_condition_false():
     registry = AutomaxEngine().plugin_registry