mirror of
https://github.com/ansible-collections/community.general.git
synced 2026-03-22 13:19:13 +00:00
5e0fd1201c
* ModuleHelper: ensure compatibility with `ModuleTestCase`. This change allows to configure the `module_fails_on_exception` decorator by passing a tuple of exception types that should not be handled by the decorator itself. In the context of `ModuleTestCase`, use `(AnsibleExitJson, AnsibleFailJson)` to let them pass through the decorator without modification. Signed-off-by: Fiehe Christoph <c.fiehe@eurodata.de> * Another approach allowing user-defined exception types to pass through the decorator. When the decorator should have no arguments at all, we must hard code the name of the attribute that is looked up on self. Signed-off-by: Fiehe Christoph <c.fiehe@eurodata.de> * Approach that removes decorator parametrization and relies on an object/class variable named `unhandled_exceptions`. Signed-off-by: Fiehe Christoph <c.fiehe@eurodata.de> * context manager implemented that allows to pass through some exception types Signed-off-by: Fiehe Christoph <c.fiehe@eurodata.de> * Update changelogs/fragments/11488-mh-ensure-compatibiliy-with-module-tests.yml Co-authored-by: Felix Fontein <felix@fontein.de> * Exception placeholder added Signed-off-by: Fiehe Christoph <c.fiehe@eurodata.de> --------- Signed-off-by: Fiehe Christoph <c.fiehe@eurodata.de> Co-authored-by: Fiehe Christoph <c.fiehe@eurodata.de> Co-authored-by: Felix Fontein <felix@fontein.de>
119 lines
3.5 KiB
Python
119 lines
3.5 KiB
Python
# (c) 2020, Alexei Znamensky <russoz@gmail.com>
|
|
# Copyright (c) 2020, Ansible Project
|
|
# Simplified BSD License (see LICENSES/BSD-2-Clause.txt or https://opensource.org/licenses/BSD-2-Clause)
|
|
# SPDX-License-Identifier: BSD-2-Clause
|
|
|
|
from __future__ import annotations
|
|
|
|
import traceback
|
|
from contextlib import contextmanager
|
|
from functools import wraps
|
|
|
|
from ansible_collections.community.general.plugins.module_utils.mh.exceptions import (
|
|
ModuleHelperException,
|
|
_UnhandledSentinel,
|
|
)
|
|
|
|
_unhandled_exceptions: tuple[type[Exception], ...] = (_UnhandledSentinel,)
|
|
|
|
|
|
def cause_changes(when=None):
|
|
def deco(func):
|
|
@wraps(func)
|
|
def wrapper(self, *args, **kwargs):
|
|
try:
|
|
func(self, *args, **kwargs)
|
|
if when == "success":
|
|
self.changed = True
|
|
except Exception:
|
|
if when == "failure":
|
|
self.changed = True
|
|
raise
|
|
finally:
|
|
if when == "always":
|
|
self.changed = True
|
|
|
|
return wrapper
|
|
|
|
return deco
|
|
|
|
|
|
@contextmanager
|
|
def no_handle_exceptions(*exceptions: type[Exception]):
|
|
global _unhandled_exceptions
|
|
current = _unhandled_exceptions
|
|
_unhandled_exceptions = tuple(exceptions)
|
|
try:
|
|
yield
|
|
finally:
|
|
_unhandled_exceptions = current
|
|
|
|
|
|
def module_fails_on_exception(func):
|
|
conflict_list = ("msg", "exception", "output", "vars", "changed")
|
|
|
|
@wraps(func)
|
|
def wrapper(self, *args, **kwargs):
|
|
def fix_key(k):
|
|
return k if k not in conflict_list else f"_{k}"
|
|
|
|
def fix_var_conflicts(output):
|
|
result = {fix_key(k): v for k, v in output.items()}
|
|
return result
|
|
|
|
try:
|
|
func(self, *args, **kwargs)
|
|
except _unhandled_exceptions:
|
|
# re-raise exception without further processing
|
|
raise
|
|
except ModuleHelperException as e:
|
|
if e.update_output:
|
|
self.update_output(e.update_output)
|
|
# patchy solution to resolve conflict with output variables
|
|
output = fix_var_conflicts(self.output)
|
|
self.module.fail_json(
|
|
msg=e.msg, exception=traceback.format_exc(), output=self.output, vars=self.vars.output(), **output
|
|
)
|
|
except Exception as e:
|
|
# patchy solution to resolve conflict with output variables
|
|
output = fix_var_conflicts(self.output)
|
|
msg = f"Module failed with exception: {str(e).strip()}"
|
|
self.module.fail_json(
|
|
msg=msg, exception=traceback.format_exc(), output=self.output, vars=self.vars.output(), **output
|
|
)
|
|
|
|
return wrapper
|
|
|
|
|
|
def check_mode_skip(func):
|
|
@wraps(func)
|
|
def wrapper(self, *args, **kwargs):
|
|
if not self.module.check_mode:
|
|
return func(self, *args, **kwargs)
|
|
|
|
return wrapper
|
|
|
|
|
|
def check_mode_skip_returns(callable=None, value=None):
|
|
def deco(func):
|
|
if callable is not None:
|
|
|
|
@wraps(func)
|
|
def wrapper_callable(self, *args, **kwargs):
|
|
if self.module.check_mode:
|
|
return callable(self, *args, **kwargs)
|
|
return func(self, *args, **kwargs)
|
|
|
|
return wrapper_callable
|
|
|
|
else:
|
|
|
|
@wraps(func)
|
|
def wrapper_value(self, *args, **kwargs):
|
|
if self.module.check_mode:
|
|
return value
|
|
return func(self, *args, **kwargs)
|
|
|
|
return wrapper_value
|
|
|
|
return deco
|