pytest-ansible's stdout_callback fails on Ansible 2.19 #489
Description
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: TypeErrorAnd 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.
Metadata
Metadata
Assignees
Type
Projects
Status