pytest-ansible
pytest-ansible copied to clipboard
ansible
pytest-ansible's stdout_callback fails on Ansible 2.19
After updating from Ansible 2.18 to 2.19, I started getting an error during the setup that pytest-ansible does for a test:
self = <pytest_ansible.module_dispatcher.v213.ModuleDispatcherV213 object at 0x10576b590>, module_args = (), complex_args = {'name': 'ANSIBLE.PGU7.T8520199.CDM7L5KT', 'state': 'absent'}
hosts = [******], extra_hosts = [], no_hosts = False, args = ['pytest-ansible', 'all', '--connection=ssh', '--user=omvsadm', '--become-method=sudo', '--become-user=root', ...]
verbosity = None, verbosity_syntax = '-vvvvv', argument = 'module-path', arg_value = ['/Users/andre/Desktop/work_2025/ibm_zos_core/venv/venv-2.19/ansible_collections/ibm/ibm_zos_core/plugins/modules']
def _run(self, *module_args, **complex_args): # type: ignore[no-untyped-def] # noqa: ANN002, ANN003, ANN202, C901, PLR0912, PLR0914, PLR0915
"""Execute an ansible adhoc command returning the result in a AdhocResult object.
Raises:
ansible.errors.AnsibleError: If the host is unreachable.
AnsibleConnectionFailure: If the host is unreachable.
""" # noqa: DOC201
# Assemble module argument string
if module_args:
complex_args.update({"_raw_params": " ".join(module_args)})
# Assert hosts matching the provided pattern exist
hosts = self.options["inventory_manager"].list_hosts()
if self.options.get("extra_inventory_manager", None):
extra_hosts = self.options["extra_inventory_manager"].list_hosts()
else:
extra_hosts = []
no_hosts = False
if len(hosts + extra_hosts) == 0:
no_hosts = True
warnings.warn("provided hosts list is empty, only localhost is available") # noqa: B028
self.options["inventory_manager"].subset(self.options.get("subset"))
hosts = self.options["inventory_manager"].list_hosts(
self.options["host_pattern"],
)
if self.options.get("extra_inventory_manager", None):
self.options["extra_inventory_manager"].subset(self.options.get("subset"))
extra_hosts = self.options["extra_inventory_manager"].list_hosts()
else:
extra_hosts = []
if len(hosts + extra_hosts) == 0 and not no_hosts:
msg = "Specified hosts and/or --limit does not match any hosts."
raise ansible.errors.AnsibleError(
msg,
)
# Pass along cli options
args = ["pytest-ansible"]
verbosity = None
for verbosity_syntax in ("-v", "-vv", "-vvv", "-vvvv", "-vvvvv"):
if verbosity_syntax in sys.argv:
verbosity = verbosity_syntax
break
if verbosity is not None:
args.append(verbosity_syntax)
args.extend([self.options["host_pattern"]])
for argument in (
"connection",
"user",
"become",
"become_method",
"become_user",
"module_path",
):
arg_value = self.options.get(argument)
argument = argument.replace("_", "-") # noqa: PLW2901
if isinstance(arg_value, typing.Hashable) and arg_value in {None, False}:
continue
if arg_value is True:
args.append(f"--{argument}")
else:
args.append(f"--{argument}={arg_value}")
# Use Ansible's own adhoc cli to parse the fake command line we created and then save it
# into Ansible's global context
adhoc = AdHocCLI(args)
adhoc.parse()
# And now we'll never speak of this again
del adhoc
# Initialize callbacks to capture module JSON responses
callback = ResultAccumulator()
kwargs = {
"inventory": self.options["inventory_manager"],
"variable_manager": self.options["variable_manager"],
"loader": self.options["loader"],
"stdout_callback": callback,
"passwords": {"conn_pass": None, "become_pass": None},
}
kwargs_extra = {}
# If we have an extra inventory, do the same that we did for the inventory
if self.options.get("extra_inventory_manager", None):
callback_extra = ResultAccumulator()
kwargs_extra = {
"inventory": self.options["extra_inventory_manager"],
"variable_manager": self.options["extra_variable_manager"],
"loader": self.options["extra_loader"],
"stdout_callback": callback_extra,
"passwords": {"conn_pass": None, "become_pass": None},
}
# create a pseudo-play to execute the specified module via a single task
play_ds = {
"name": "pytest-ansible",
"hosts": self.options["host_pattern"],
"become": self.options.get("become"),
"become_user": self.options.get("become_user"),
"gather_facts": "no",
"tasks": [
{
"action": {
"module": self.options["module_name"],
"args": complex_args,
},
},
],
}
play = Play().load(
play_ds,
variable_manager=self.options["variable_manager"],
loader=self.options["loader"],
)
play_extra = None
if self.options.get("extra_inventory_manager", None):
play_extra = Play().load(
play_ds,
variable_manager=self.options["extra_variable_manager"],
loader=self.options["extra_loader"],
)
if HAS_CUSTOM_LOADER_SUPPORT:
# Load the collection finder, unsupported, may change in future
init_plugin_loader(COLLECTIONS_PATHS)
# now create a task queue manager to execute the play
tqm = None
try:
> tqm = TaskQueueManager(**kwargs)
^^^^^^^^^^^^^^^^^^^^^^^^^^
E TypeError: TaskQueueManager.__init__() got an unexpected keyword argument 'stdout_callback'
venv/venv-2.19/lib/python3.12/site-packages/pytest_ansible/module_dispatcher/v213.py:251: TypeError
And after looking around a bit, I found that this commit from Ansible introduced some changes to the way callbacks are handled by the engine. This is happening on the most recent version of pytest-ansible, 25.6.3.
I only spent a couple of minutes looking this over so by no means am I suggesting this is the solution but a start would be to update pytest_ansible/module_dispatcher/v213.py with the updated keyword used by ansible now which is stdout_callback_name.
Line(s) of interest in pytest_ansible/module_dispatcher/v213.py The PR for the change from stdout_callback to stdout_callback_name.
After changing the name stdout_callback to stdout_callback_name it failed with the following error:
../../venv/AC2.19PY3.13/lib/python3.13/site-packages/pytest_ansible/module_dispatcher/v213.py:255: in _run
tqm.run(play)
../../venv/AC2.19PY3.13/lib/python3.13/site-packages/ansible/executor/task_queue_manager.py:284: in run
self.load_callbacks()
../../venv/AC2.19PY3.13/lib/python3.13/site-packages/ansible/executor/task_queue_manager.py:206: in load_callbacks
stdout_callback = callback_loader.get(self._stdout_callback_name)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
../../venv/AC2.19PY3.13/lib/python3.13/site-packages/ansible/plugins/loader.py:978: in get
ctx = self.get_with_context(name, *args, **kwargs)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
../../venv/AC2.19PY3.13/lib/python3.13/site-packages/ansible/plugins/loader.py:1008: in get_with_context
plugin_load_context = self.find_plugin_with_context(name, collection_list=collection_list)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
../../venv/AC2.19PY3.13/lib/python3.13/site-packages/ansible/plugins/loader.py:711: in find_plugin_with_context
result = self._resolve_plugin_step(name, mod_type, ignore_deprecated, check_aliases, collection_list, plugin_load_context=plugin_load_context)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
../../venv/AC2.19PY3.13/lib/python3.13/site-packages/ansible/plugins/loader.py:762: in _resolve_plugin_step
if (AnsibleCollectionRef.is_valid_fqcr(name) or collection_list) and not name.startswith('Ansible'):
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
../../venv/AC2.19PY3.13/lib/python3.13/site-packages/ansible/utils/collection_loader/_collection_finder.py:1052: in is_valid_fqcr
ref = _to_text(ref)
^^^^^^^^^^^^^
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
value = ResultAccumulator(plugin_type='resultaccumulator', ansible_name=None, load_name=None), strict = False
def _to_text(value: str | bytes | _EncryptedStringProtocol | None, strict: bool = False) -> str | None:
"""Internal implementation to keep collection loader standalone."""
# FUTURE: remove this method when _to_bytes is removed
if value is None:
return None
if isinstance(value, str):
return value
if isinstance(value, bytes):
return value.decode(errors='strict' if strict else 'surrogateescape')
if isinstance(value, _EncryptedStringProtocol):
return value._decrypt()
> raise TypeError(f'unsupported type {type(value)}')
E TypeError: unsupported type <class 'pytest_ansible.module_dispatcher.v213.ResultAccumulator'>
../../venv/AC2.19PY3.13/lib/python3.13/site-packages/ansible/utils/collection_loader/__init__.py:37: TypeError
Got a similar error log.
Env
Target directory is not .git, molecule scenarios detection will be skipped.
Test session starts (platform: linux, Python 3.12.11, pytest 8.4.2, pytest-sugar 1.1.1)
cachedir: .pytest_cache
ansible: 2.19.2
rootdir: /home/zhuxu/automation/tests
configfile: pytest.ini
plugins: ansible-playbook-runner-0.5.4, xdist-3.8.0, ansible-25.8.0, plus-0.8.1, rerunfailures-16.0.1, sugar-1.1.1
Error Log
(automation) zhuxu@US-ZHUXU-LNX ~/automation $ pytest tests/test_inventory.py --ansible-inventory=ansible-test/inventories/dev.ini --ansible-host-pattern=dev_group -v
Target directory is not .git, molecule scenarios detection will be skipped.
Test session starts (platform: linux, Python 3.12.11, pytest 8.4.2, pytest-sugar 1.1.1)
cachedir: .pytest_cache
ansible: 2.19.2
rootdir: /home/zhuxu/automation/tests
configfile: pytest.ini
plugins: ansible-playbook-runner-0.5.4, xdist-3.8.0, ansible-25.8.0, plus-0.8.1, rerunfailures-16.0.1, sugar-1.1.1
collected 1 item
―――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――― test_inventory ―――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――
tests/test_inventory.py:7: in test_inventory
result = ansible_module.ping()
^^^^^^^^^^^^^^^^^^^^^
.venv/lib/python3.12/site-packages/pytest_ansible/module_dispatcher/v213.py:254: in _run
tqm = TaskQueueManager(**kwargs) # pylint: disable=unexpected-keyword-arg,useless-suppression
^^^^^^^^^^^^^^^^^^^^^^^^^^
E TypeError: TaskQueueManager.__init__() got an unexpected keyword argument 'stdout_callback'
----------------------------------------------------------------------------------------------- Captured stderr call ------------------------------------------------------------------------------------------------
[WARNING]: Deprecation warnings can be disabled by setting `deprecation_warnings=False` in ansible.cfg.
[DEPRECATION WARNING]: Using a mapping for `action` is deprecated. This feature will be removed from ansible-core version 2.23.
Origin: <unknown>
{'module': 'ansible.builtin.ping', 'args': {}}
Use a string value for `action`.
test_inventory.py::test_inventory ⨯ 100% ██████████
============================================================================================== short test summary info ==============================================================================================
FAILED tests/test_inventory.py::test_inventory - TypeError: TaskQueueManager.__init__() got an unexpected keyword argument 'stdout_callback'
Results (0.15s):
1 failed
- test_inventory.py:6 test_inventory
Test script
(automation) zhuxu@US-ZHUXU-LNX ~/automation $ cat tests/test_inventory.py
import pytest
def test_inventory(ansible_module):
result = ansible_module.ping()
assert result.is_successful
Hi folks, we’re going to look at and triage this issue. Thanks for raising!
@alisonlhart I opened issue https://github.com/ansible/ansible/issues/86137 about the incompatible change of the init parameter name, and that got closed again within a couple of minutes saying that TaskQueueManager is not a public API.
I guess it is on you to deal with that change.
PS: After adjusting the parameter names, I could reproduce the secondary issue @fernandofloresg reported above.
Hi @alisonlhart Do we have any updates on this issue? We are unable to test our collection in 2.19