fix script failed_when bypass (#85326)

* fix script failed_when bypass

* misuse of AnsibleActionFail for a retryable/alterable error case
* add tests

Co-authored-by: Matt Clay <matt@mystile.com>

* type annotation fixes

---------

Co-authored-by: Matt Clay <matt@mystile.com>
pull/85327/head
Matt Davis 11 months ago committed by GitHub
parent f7e0612a85
commit ed90a424a9
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194

@ -103,7 +103,7 @@ class ActionBase(ABC, _AnsiblePluginInfoMixin):
self._display = display
@abstractmethod
def run(self, tmp=None, task_vars=None):
def run(self, tmp: str | None = None, task_vars: dict[str, t.Any] | None = None) -> dict[str, t.Any]:
""" Action Plugins should implement this method to perform their
tasks. Everything else in this base class is a helper method for the
action plugin to do that.
@ -120,7 +120,7 @@ class ActionBase(ABC, _AnsiblePluginInfoMixin):
* Module parameters. These are stored in self._task.args
"""
# does not default to {'changed': False, 'failed': False}, as it used to break async
result = {}
result: dict[str, t.Any] = {}
if tmp is not None:
display.warning('ActionModule.run() no longer honors the tmp parameter. Action'

@ -20,6 +20,7 @@ import os
import pathlib
import re
import shlex
import typing as _t
from ansible.errors import AnsibleError, AnsibleActionFail, AnsibleActionSkip
from ansible.executor.powershell import module_manifest as ps_manifest
@ -35,7 +36,7 @@ class ActionModule(ActionBase):
# after chopping off a potential drive letter.
windows_absolute_path_detection = re.compile(r'^(?:[a-zA-Z]\:)?(\\|\/)')
def run(self, tmp=None, task_vars=None):
def run(self, tmp: str | None = None, task_vars: dict[str, _t.Any] | None = None) -> dict[str, _t.Any]:
""" handler for file transfer operations """
if task_vars is None:
task_vars = dict()
@ -130,7 +131,7 @@ class ActionModule(ActionBase):
self._fixup_perms2((self._connection._shell.tmpdir, tmp_src), execute=True)
# add preparation steps to one ssh roundtrip executing the script
env_dict = dict()
env_dict: dict[str, _t.Any] = {}
env_string = self._compute_environment_string(env_dict)
if executable:
@ -164,10 +165,10 @@ class ActionModule(ActionBase):
script_cmd = self._connection._shell.build_module_command(env_string='', shebang='#!powershell', cmd='')
# now we execute script, always assume changed.
result = dict(self._low_level_execute_command(cmd=script_cmd, in_data=exec_data, sudoable=True, chdir=chdir), changed=True)
result: dict[str, object] = dict(self._low_level_execute_command(cmd=script_cmd, in_data=exec_data, sudoable=True, chdir=chdir), changed=True)
if 'rc' in result and result['rc'] != 0:
raise AnsibleActionFail('non-zero return code', result=result)
result.update(msg='non-zero return code', failed=True)
return result
finally:

@ -0,0 +1,3 @@
#!/usr/bin/env bash
exit 1

@ -58,6 +58,29 @@
- "script_result0.rc == 0"
- "script_result0.stdout == 'win'"
- name: Basic non-zero RC
script: exit_1.sh
ignore_errors: true
register: non_zero_rc
- name: Ensure non-zero RC result
assert:
that:
- non_zero_rc is failed
- non_zero_rc.rc == 1
- name: Exercise failed_when on non-zero RC
script: exit_1.sh
register: non_zero_rc_failed_when
failed_when: false
- name: Ensure failed_when executed
assert:
that:
- non_zero_rc_failed_when is success
- non_zero_rc_failed_when.rc == 1
- non_zero_rc_failed_when.failed_when_result is false
- name: Execute a script with a space in the path
script: "'space path/test.sh'"
register: _space_path_test

Loading…
Cancel
Save