mirror of
https://github.com/ansible-collections/community.general.git
synced 2024-09-14 20:13:21 +02:00
Command Runner (#4476)
* initial commit, passing unit tests
* passing one very silly integration test
* multiple changes:
- updated copyright year
- cmd_runner
- added fmt_optval
- created specific exceptions
- fixed bug in context class where values from module params were not
being used for resolving cmd arguments
- changed order of class declaration for readability purpose
- tests
- minor improvements in integration test code
- removed some extraneous code in msimple.yml
- minor improvements in unit tests
- added few missing cases to unit test
* multiple changes
cmd_runner.py
- renamed InvalidParameterName to MissingArgumentFormat
- improved exception parameters
- added repr and str to all exceptions
- added unpacking decorator for fmt functions
- CmdRunner
- improved parameter validation
- _CmdRunnerContext
- Context runs must now pass named arguments
- Simplified passing of additional arguments to module.run_command()
- Provided multiple context variables with info about the run
Integration tests
- rename msimple.py to cmd_echo.py for clarity
- added more test cases
* cmd_runner: env update can be passed to runner
* adding runner context info to output
* added comment on OrderedDict
* wrong variable
* refactored all fmt functions into static methods of a class
Imports should be simpler now, only one object fmt, with attr access to all callables
* added unit tests for CmdRunner
* fixed sanity checks
* fixed mock imports
* added more unit tests for CmdRunner
* terminology consistency
* multiple adjustments:
- remove extraneous imports
- renamed some variables
- added wrapper around arg formatters to handle individual arg ignore_none behaviour
* removed old code commented out in test
* multiple changes:
- ensure fmt functions return list of strings
- renamed fmt parameter from `option` to `args`
- renamed fmt.mapped to fmt.as_map
- simplified fmt.as_map
- added tests for fmt.as_fixed
* more improvements in formats
* fixed sanity
* args_order can be a string (to be split())
and improved integration test
* simplified integration test
* removed overkill str() on values - run_command does that for us
* as_list makes more sense than as_str in that context
* added changelog fragment
* Update plugins/module_utils/cmd_runner.py
Co-authored-by: Felix Fontein <felix@fontein.de>
* adjusted __repr__ output for the exceptions
* added superclass object to classes
* added additional comment on the testcase sample/example
* suggestion from PR
Co-authored-by: Felix Fontein <felix@fontein.de>
This commit is contained in:
Alexei Znamensky
GitHub
parent
dbad1e0f11
commit
f5b1b3c6f0
8 changed files with 782 additions and 0 deletions
2
changelogs/fragments/4476-cmd_runner.yml
Normal file
2
changelogs/fragments/4476-cmd_runner.yml
Normal file
|
|
@ -0,0 +1,2 @@
|
|||
minor_changes:
|
||||
- cmd_runner module util - reusable command runner with consistent argument formatting and sensible defaults (https://github.com/ansible-collections/community.general/pull/4476).
|
||||
291
plugins/module_utils/cmd_runner.py
Normal file
291
plugins/module_utils/cmd_runner.py
Normal file
|
|
@ -0,0 +1,291 @@
|
|||
# -*- coding: utf-8 -*-
|
||||
# (c) 2022, Alexei Znamensky <russoz@gmail.com>
|
||||
# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
|
||||
|
||||
from __future__ import absolute_import, division, print_function
|
||||
__metaclass__ = type
|
||||
|
||||
from functools import wraps
|
||||
|
||||
from ansible.module_utils.common.collections import is_sequence
|
||||
from ansible.module_utils.six import iteritems
|
||||
|
||||
|
||||
def _ensure_list(value):
|
||||
return list(value) if is_sequence(value) else [value]
|
||||
|
||||
|
||||
def _process_as_is(rc, out, err):
|
||||
return rc, out, err
|
||||
|
||||
|
||||
class CmdRunnerException(Exception):
|
||||
pass
|
||||
|
||||
|
||||
class MissingArgumentFormat(CmdRunnerException):
|
||||
def __init__(self, arg, args_order, args_formats):
|
||||
self.args_order = args_order
|
||||
self.arg = arg
|
||||
self.args_formats = args_formats
|
||||
|
||||
def __repr__(self):
|
||||
return "MissingArgumentFormat({0!r}, {1!r}, {2!r})".format(
|
||||
self.arg,
|
||||
self.args_order,
|
||||
self.args_formats,
|
||||
)
|
||||
|
||||
def __str__(self):
|
||||
return "Cannot find format for parameter {0} {1} in: {2}".format(
|
||||
self.arg,
|
||||
self.args_order,
|
||||
self.args_formats,
|
||||
)
|
||||
|
||||
|
||||
class MissingArgumentValue(CmdRunnerException):
|
||||
def __init__(self, args_order, arg):
|
||||
self.args_order = args_order
|
||||
self.arg = arg
|
||||
|
||||
def __repr__(self):
|
||||
return "MissingArgumentValue({0!r}, {1!r})".format(
|
||||
self.args_order,
|
||||
self.arg,
|
||||
)
|
||||
|
||||
def __str__(self):
|
||||
return "Cannot find value for parameter {0} in {1}".format(
|
||||
self.arg,
|
||||
self.args_order,
|
||||
)
|
||||
|
||||
|
||||
class FormatError(CmdRunnerException):
|
||||
def __init__(self, name, value, args_formats, exc):
|
||||
self.name = name
|
||||
self.value = value
|
||||
self.args_formats = args_formats
|
||||
self.exc = exc
|
||||
super(FormatError, self).__init__()
|
||||
|
||||
def __repr__(self):
|
||||
return "FormatError({0!r}, {1!r}, {2!r}, {3!r})".format(
|
||||
self.name,
|
||||
self.value,
|
||||
self.args_formats,
|
||||
self.exc,
|
||||
)
|
||||
|
||||
def __str__(self):
|
||||
return "Failed to format parameter {0} with value {1}: {2}".format(
|
||||
self.name,
|
||||
self.value,
|
||||
self.exc,
|
||||
)
|
||||
|
||||
|
||||
class _ArgFormat(object):
|
||||
def __init__(self, func, ignore_none=None):
|
||||
self.func = func
|
||||
self.ignore_none = ignore_none
|
||||
|
||||
def __call__(self, value, ctx_ignore_none):
|
||||
ignore_none = self.ignore_none if self.ignore_none is not None else ctx_ignore_none
|
||||
if value is None and ignore_none:
|
||||
return []
|
||||
f = self.func
|
||||
return [str(x) for x in f(value)]
|
||||
|
||||
|
||||
class _Format(object):
|
||||
@staticmethod
|
||||
def as_bool(args):
|
||||
return _ArgFormat(lambda value: _ensure_list(args) if value else [])
|
||||
|
||||
@staticmethod
|
||||
def as_bool_not(args):
|
||||
return _ArgFormat(lambda value: [] if value else _ensure_list(args), ignore_none=False)
|
||||
|
||||
@staticmethod
|
||||
def as_optval(arg, ignore_none=None):
|
||||
return _ArgFormat(lambda value: ["{0}{1}".format(arg, value)], ignore_none=ignore_none)
|
||||
|
||||
@staticmethod
|
||||
def as_opt_val(arg, ignore_none=None):
|
||||
return _ArgFormat(lambda value: [arg, value], ignore_none=ignore_none)
|
||||
|
||||
@staticmethod
|
||||
def as_opt_eq_val(arg, ignore_none=None):
|
||||
return _ArgFormat(lambda value: ["{0}={1}".format(arg, value)], ignore_none=ignore_none)
|
||||
|
||||
@staticmethod
|
||||
def as_list(ignore_none=None):
|
||||
return _ArgFormat(_ensure_list, ignore_none=ignore_none)
|
||||
|
||||
@staticmethod
|
||||
def as_fixed(args):
|
||||
return _ArgFormat(lambda value: _ensure_list(args), ignore_none=False)
|
||||
|
||||
@staticmethod
|
||||
def as_func(func, ignore_none=None):
|
||||
return _ArgFormat(func, ignore_none=ignore_none)
|
||||
|
||||
@staticmethod
|
||||
def as_map(_map, default=None, ignore_none=None):
|
||||
return _ArgFormat(lambda value: _ensure_list(_map.get(value, default)), ignore_none=ignore_none)
|
||||
|
||||
@staticmethod
|
||||
def as_default_type(_type, arg="", ignore_none=None):
|
||||
fmt = _Format
|
||||
if _type == "dict":
|
||||
return fmt.as_func(lambda d: ["--{0}={1}".format(*a) for a in iteritems(d)],
|
||||
ignore_none=ignore_none)
|
||||
if _type == "list":
|
||||
return fmt.as_func(lambda value: ["--{0}".format(x) for x in value], ignore_none=ignore_none)
|
||||
if _type == "bool":
|
||||
return fmt.as_bool("--{0}".format(arg))
|
||||
|
||||
return fmt.as_opt_val("--{0}".format(arg), ignore_none=ignore_none)
|
||||
|
||||
@staticmethod
|
||||
def unpack_args(func):
|
||||
@wraps(func)
|
||||
def wrapper(v):
|
||||
return func(*v)
|
||||
return wrapper
|
||||
|
||||
@staticmethod
|
||||
def unpack_kwargs(func):
|
||||
@wraps(func)
|
||||
def wrapper(v):
|
||||
return func(**v)
|
||||
return wrapper
|
||||
|
||||
|
||||
class CmdRunner(object):
|
||||
"""
|
||||
Wrapper for ``AnsibleModule.run_command()``.
|
||||
|
||||
It aims to provide a reusable runner with consistent argument formatting
|
||||
and sensible defaults.
|
||||
"""
|
||||
|
||||
@staticmethod
|
||||
def _prepare_args_order(order):
|
||||
return tuple(order) if is_sequence(order) else tuple(order.split())
|
||||
|
||||
def __init__(self, module, command, arg_formats=None, default_args_order=(),
|
||||
check_rc=False, force_lang="C", path_prefix=None, environ_update=None):
|
||||
self.module = module
|
||||
self.command = _ensure_list(command)
|
||||
self.default_args_order = self._prepare_args_order(default_args_order)
|
||||
if arg_formats is None:
|
||||
arg_formats = {}
|
||||
self.arg_formats = dict(arg_formats)
|
||||
self.check_rc = check_rc
|
||||
self.force_lang = force_lang
|
||||
self.path_prefix = path_prefix
|
||||
if environ_update is None:
|
||||
environ_update = {}
|
||||
self.environ_update = environ_update
|
||||
|
||||
self.command[0] = module.get_bin_path(command[0], opt_dirs=path_prefix, required=True)
|
||||
|
||||
for mod_param_name, spec in iteritems(module.argument_spec):
|
||||
if mod_param_name not in self.arg_formats:
|
||||
self.arg_formats[mod_param_name] = _Format.as_default_type(spec['type'], mod_param_name)
|
||||
|
||||
def context(self, args_order=None, output_process=None, ignore_value_none=True, **kwargs):
|
||||
if output_process is None:
|
||||
output_process = _process_as_is
|
||||
if args_order is None:
|
||||
args_order = self.default_args_order
|
||||
args_order = self._prepare_args_order(args_order)
|
||||
for p in args_order:
|
||||
if p not in self.arg_formats:
|
||||
raise MissingArgumentFormat(p, args_order, tuple(self.arg_formats.keys()))
|
||||
return _CmdRunnerContext(runner=self,
|
||||
args_order=args_order,
|
||||
output_process=output_process,
|
||||
ignore_value_none=ignore_value_none, **kwargs)
|
||||
|
||||
def has_arg_format(self, arg):
|
||||
return arg in self.arg_formats
|
||||
|
||||
|
||||
class _CmdRunnerContext(object):
|
||||
def __init__(self, runner, args_order, output_process, ignore_value_none, **kwargs):
|
||||
self.runner = runner
|
||||
self.args_order = tuple(args_order)
|
||||
self.output_process = output_process
|
||||
self.ignore_value_none = ignore_value_none
|
||||
self.run_command_args = dict(kwargs)
|
||||
|
||||
self.environ_update = runner.environ_update
|
||||
self.environ_update.update(self.run_command_args.get('environ_update', {}))
|
||||
if runner.force_lang:
|
||||
self.environ_update.update({
|
||||
'LANGUAGE': runner.force_lang,
|
||||
'LC_ALL': runner.force_lang,
|
||||
})
|
||||
self.run_command_args['environ_update'] = self.environ_update
|
||||
|
||||
if 'check_rc' not in self.run_command_args:
|
||||
self.run_command_args['check_rc'] = runner.check_rc
|
||||
self.check_rc = self.run_command_args['check_rc']
|
||||
|
||||
self.cmd = None
|
||||
self.results_rc = None
|
||||
self.results_out = None
|
||||
self.results_err = None
|
||||
self.results_processed = None
|
||||
|
||||
def run(self, **kwargs):
|
||||
runner = self.runner
|
||||
module = self.runner.module
|
||||
self.cmd = list(runner.command)
|
||||
self.context_run_args = dict(kwargs)
|
||||
|
||||
named_args = dict(module.params)
|
||||
named_args.update(kwargs)
|
||||
for arg_name in self.args_order:
|
||||
value = None
|
||||
try:
|
||||
value = named_args[arg_name]
|
||||
self.cmd.extend(runner.arg_formats[arg_name](value, ctx_ignore_none=self.ignore_value_none))
|
||||
except KeyError:
|
||||
raise MissingArgumentValue(self.args_order, arg_name)
|
||||
except Exception as e:
|
||||
raise FormatError(arg_name, value, runner.arg_formats[arg_name], e)
|
||||
|
||||
results = module.run_command(self.cmd, **self.run_command_args)
|
||||
self.results_rc, self.results_out, self.results_err = results
|
||||
self.results_processed = self.output_process(*results)
|
||||
return self.results_processed
|
||||
|
||||
@property
|
||||
def run_info(self):
|
||||
return dict(
|
||||
ignore_value_none=self.ignore_value_none,
|
||||
check_rc=self.check_rc,
|
||||
environ_update=self.environ_update,
|
||||
args_order=self.args_order,
|
||||
cmd=self.cmd,
|
||||
run_command_args=self.run_command_args,
|
||||
context_run_args=self.context_run_args,
|
||||
results_rc=self.results_rc,
|
||||
results_out=self.results_out,
|
||||
results_err=self.results_err,
|
||||
results_processed=self.results_processed,
|
||||
)
|
||||
|
||||
def __enter__(self):
|
||||
return self
|
||||
|
||||
def __exit__(self, exc_type, exc_val, exc_tb):
|
||||
return False
|
||||
|
||||
|
||||
fmt = _Format()
|
||||
1
tests/integration/targets/cmd_runner/aliases
Normal file
1
tests/integration/targets/cmd_runner/aliases
Normal file
|
|
@ -0,0 +1 @@
|
|||
shippable/posix/group2
|
||||
78
tests/integration/targets/cmd_runner/library/cmd_echo.py
Normal file
78
tests/integration/targets/cmd_runner/library/cmd_echo.py
Normal file
|
|
@ -0,0 +1,78 @@
|
|||
#!/usr/bin/python
|
||||
# -*- coding: utf-8 -*-
|
||||
# (c) 2022, Alexei Znamensky <russoz@gmail.com>
|
||||
# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
|
||||
|
||||
from __future__ import absolute_import, division, print_function
|
||||
__metaclass__ = type
|
||||
|
||||
import sys
|
||||
|
||||
DOCUMENTATION = '''
|
||||
module: cmd_echo
|
||||
author: "Alexei Znamensky (@russoz)"
|
||||
short_description: Simple module for testing
|
||||
description:
|
||||
- Simple module test description.
|
||||
options:
|
||||
command:
|
||||
description: aaa
|
||||
type: list
|
||||
elements: str
|
||||
required: true
|
||||
arg_formats:
|
||||
description: bbb
|
||||
type: dict
|
||||
required: true
|
||||
arg_order:
|
||||
description: ccc
|
||||
type: raw
|
||||
required: true
|
||||
arg_values:
|
||||
description: ddd
|
||||
type: list
|
||||
required: true
|
||||
aa:
|
||||
description: eee
|
||||
type: raw
|
||||
'''
|
||||
|
||||
EXAMPLES = ""
|
||||
|
||||
RETURN = ""
|
||||
|
||||
from ansible.module_utils.basic import AnsibleModule
|
||||
from ansible_collections.community.general.plugins.module_utils.cmd_runner import CmdRunner, fmt
|
||||
|
||||
|
||||
def main():
|
||||
module = AnsibleModule(
|
||||
argument_spec=dict(
|
||||
arg_formats=dict(type="dict", default={}),
|
||||
arg_order=dict(type="raw", required=True),
|
||||
arg_values=dict(type="dict", default={}),
|
||||
aa=dict(type="raw"),
|
||||
),
|
||||
)
|
||||
p = module.params
|
||||
|
||||
arg_formats = {}
|
||||
for arg, fmt_spec in p['arg_formats'].items():
|
||||
func = getattr(fmt, fmt_spec['func'])
|
||||
args = fmt_spec.get("args", [])
|
||||
|
||||
arg_formats[arg] = func(*args)
|
||||
|
||||
runner = CmdRunner(module, ['echo', '--'], arg_formats=arg_formats)
|
||||
|
||||
info = None
|
||||
with runner.context(p['arg_order']) as ctx:
|
||||
result = ctx.run(**p['arg_values'])
|
||||
info = ctx.run_info
|
||||
rc, out, err = result
|
||||
|
||||
module.exit_json(rc=rc, out=out, err=err, info=info)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
main()
|
||||
7
tests/integration/targets/cmd_runner/tasks/main.yml
Normal file
7
tests/integration/targets/cmd_runner/tasks/main.yml
Normal file
|
|
@ -0,0 +1,7 @@
|
|||
# (c) 2022, Alexei Znamensky
|
||||
# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
|
||||
|
||||
- name: parameterized test cmd_echo
|
||||
ansible.builtin.include_tasks:
|
||||
file: test_cmd_echo.yml
|
||||
loop: "{{ cmd_echo_tests }}"
|
||||
13
tests/integration/targets/cmd_runner/tasks/test_cmd_echo.yml
Normal file
13
tests/integration/targets/cmd_runner/tasks/test_cmd_echo.yml
Normal file
|
|
@ -0,0 +1,13 @@
|
|||
---
|
||||
- name: test cmd_echo [{{ item.name }}]
|
||||
cmd_echo:
|
||||
arg_formats: "{{ item.arg_formats|default(omit) }}"
|
||||
arg_order: "{{ item.arg_order }}"
|
||||
arg_values: "{{ item.arg_values|default(omit) }}"
|
||||
aa: "{{ item.aa|default(omit) }}"
|
||||
register: test_result
|
||||
ignore_errors: "{{ item.expect_error|default(omit) }}"
|
||||
|
||||
- name: check results [{{ item.name }}]
|
||||
assert:
|
||||
that: "{{ item.assertions }}"
|
||||
84
tests/integration/targets/cmd_runner/vars/main.yml
Normal file
84
tests/integration/targets/cmd_runner/vars/main.yml
Normal file
|
|
@ -0,0 +1,84 @@
|
|||
# -*- coding: utf-8 -*-
|
||||
# (c) 2022, Alexei Znamensky
|
||||
# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
|
||||
|
||||
cmd_echo_tests:
|
||||
- name: set aa and bb value
|
||||
arg_formats:
|
||||
aa:
|
||||
func: as_opt_eq_val
|
||||
args: [--answer]
|
||||
bb:
|
||||
func: as_bool
|
||||
args: [--bb-here]
|
||||
arg_order: 'aa bb'
|
||||
arg_values:
|
||||
bb: true
|
||||
aa: 11
|
||||
assertions:
|
||||
- test_result.rc == 0
|
||||
- test_result.out == "-- --answer=11 --bb-here\n"
|
||||
- test_result.err == ""
|
||||
|
||||
- name: default aa value
|
||||
arg_formats:
|
||||
aa:
|
||||
func: as_opt_eq_val
|
||||
args: [--answer]
|
||||
bb:
|
||||
func: as_bool
|
||||
args: [--bb-here]
|
||||
arg_order: ['aa', 'bb']
|
||||
arg_values:
|
||||
aa: 43
|
||||
bb: true
|
||||
assertions:
|
||||
- test_result.rc == 0
|
||||
- test_result.out == "-- --answer=43 --bb-here\n"
|
||||
- test_result.err == ""
|
||||
|
||||
- name: implicit aa format
|
||||
arg_formats:
|
||||
bb:
|
||||
func: as_bool
|
||||
args: [--bb-here]
|
||||
arg_order: ['aa', 'bb']
|
||||
arg_values:
|
||||
bb: true
|
||||
aa: 1984
|
||||
assertions:
|
||||
- test_result.rc == 0
|
||||
- test_result.out == "-- --aa 1984 --bb-here\n"
|
||||
- test_result.err == ""
|
||||
|
||||
- name: missing bb format
|
||||
arg_order: ['aa', 'bb']
|
||||
arg_values:
|
||||
bb: true
|
||||
aa: 1984
|
||||
expect_error: true
|
||||
assertions:
|
||||
- test_result is failed
|
||||
- test_result.rc == 1
|
||||
- '"out" not in test_result'
|
||||
- '"err" not in test_result'
|
||||
- >-
|
||||
"MissingArgumentFormat: Cannot find format for parameter bb"
|
||||
in test_result.module_stderr
|
||||
|
||||
- name: missing bb value
|
||||
arg_formats:
|
||||
bb:
|
||||
func: as_bool
|
||||
args: [--bb-here]
|
||||
arg_order: 'aa bb'
|
||||
aa: 1984
|
||||
expect_error: true
|
||||
assertions:
|
||||
- test_result is failed
|
||||
- test_result.rc == 1
|
||||
- '"out" not in test_result'
|
||||
- '"err" not in test_result'
|
||||
- >-
|
||||
"MissingArgumentValue: Cannot find value for parameter bb"
|
||||
in test_result.module_stderr
|
||||
306
tests/unit/plugins/module_utils/test_cmd_runner.py
Normal file
306
tests/unit/plugins/module_utils/test_cmd_runner.py
Normal file
|
|
@ -0,0 +1,306 @@
|
|||
# -*- coding: utf-8 -*-
|
||||
# (c) 2022, Alexei Znamensky <russoz@gmail.com>
|
||||
# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
|
||||
|
||||
from __future__ import (absolute_import, division, print_function)
|
||||
__metaclass__ = type
|
||||
|
||||
from sys import version_info
|
||||
|
||||
import pytest
|
||||
|
||||
from ansible_collections.community.general.tests.unit.compat.mock import MagicMock, PropertyMock
|
||||
from ansible_collections.community.general.plugins.module_utils.cmd_runner import CmdRunner, fmt
|
||||
|
||||
|
||||
TC_FORMATS = dict(
|
||||
simple_boolean__true=(fmt.as_bool, ("--superflag",), True, ["--superflag"]),
|
||||
simple_boolean__false=(fmt.as_bool, ("--superflag",), False, []),
|
||||
simple_boolean__none=(fmt.as_bool, ("--superflag",), None, []),
|
||||
simple_boolean_not__true=(fmt.as_bool_not, ("--superflag",), True, []),
|
||||
simple_boolean_not__false=(fmt.as_bool_not, ("--superflag",), False, ["--superflag"]),
|
||||
simple_boolean_not__none=(fmt.as_bool_not, ("--superflag",), None, ["--superflag"]),
|
||||
simple_optval__str=(fmt.as_optval, ("-t",), "potatoes", ["-tpotatoes"]),
|
||||
simple_optval__int=(fmt.as_optval, ("-t",), 42, ["-t42"]),
|
||||
simple_opt_val__str=(fmt.as_opt_val, ("-t",), "potatoes", ["-t", "potatoes"]),
|
||||
simple_opt_val__int=(fmt.as_opt_val, ("-t",), 42, ["-t", "42"]),
|
||||
simple_opt_eq_val__str=(fmt.as_opt_eq_val, ("--food",), "potatoes", ["--food=potatoes"]),
|
||||
simple_opt_eq_val__int=(fmt.as_opt_eq_val, ("--answer",), 42, ["--answer=42"]),
|
||||
simple_list_potato=(fmt.as_list, (), "literal_potato", ["literal_potato"]),
|
||||
simple_list_42=(fmt.as_list, (), 42, ["42"]),
|
||||
simple_map=(fmt.as_map, ({'a': 1, 'b': 2, 'c': 3},), 'b', ["2"]),
|
||||
simple_default_type__list=(fmt.as_default_type, ("list",), [1, 2, 3, 5, 8], ["--1", "--2", "--3", "--5", "--8"]),
|
||||
simple_default_type__bool_true=(fmt.as_default_type, ("bool", "what"), True, ["--what"]),
|
||||
simple_default_type__bool_false=(fmt.as_default_type, ("bool", "what"), False, []),
|
||||
simple_default_type__potato=(fmt.as_default_type, ("any-other-type", "potato"), "42", ["--potato", "42"]),
|
||||
simple_fixed_true=(fmt.as_fixed, [("--always-here", "--forever")], True, ["--always-here", "--forever"]),
|
||||
simple_fixed_false=(fmt.as_fixed, [("--always-here", "--forever")], False, ["--always-here", "--forever"]),
|
||||
simple_fixed_none=(fmt.as_fixed, [("--always-here", "--forever")], None, ["--always-here", "--forever"]),
|
||||
simple_fixed_str=(fmt.as_fixed, [("--always-here", "--forever")], "something", ["--always-here", "--forever"]),
|
||||
)
|
||||
if tuple(version_info) >= (3, 1):
|
||||
from collections import OrderedDict
|
||||
|
||||
# needs OrderedDict to provide a consistent key order
|
||||
TC_FORMATS["simple_default_type__dict"] = ( # type: ignore
|
||||
fmt.as_default_type,
|
||||
("dict",),
|
||||
OrderedDict((('a', 1), ('b', 2))),
|
||||
["--a=1", "--b=2"]
|
||||
)
|
||||
TC_FORMATS_IDS = sorted(TC_FORMATS.keys())
|
||||
|
||||
|
||||
@pytest.mark.parametrize('func, fmt_opt, value, expected',
|
||||
(TC_FORMATS[tc] for tc in TC_FORMATS_IDS),
|
||||
ids=TC_FORMATS_IDS)
|
||||
def test_arg_format(func, fmt_opt, value, expected):
|
||||
fmt_func = func(*fmt_opt)
|
||||
actual = fmt_func(value, ctx_ignore_none=True)
|
||||
print("formatted string = {0}".format(actual))
|
||||
assert actual == expected, "actual = {0}".format(actual)
|
||||
|
||||
|
||||
TC_RUNNER = dict(
|
||||
# SAMPLE: This shows all possible elements of a test case. It does not actually run.
|
||||
#
|
||||
# testcase_name=(
|
||||
# # input
|
||||
# dict(
|
||||
# args_bundle = dict(
|
||||
# param1=dict(
|
||||
# type="int",
|
||||
# value=11,
|
||||
# fmt_func=fmt.as_opt_eq_val,
|
||||
# fmt_arg="--answer",
|
||||
# ),
|
||||
# param2=dict(
|
||||
# fmt_func=fmt.as_bool,
|
||||
# fmt_arg="--bb-here",
|
||||
# )
|
||||
# ),
|
||||
# runner_init_args = dict(
|
||||
# command="testing",
|
||||
# default_args_order=(),
|
||||
# check_rc=False,
|
||||
# force_lang="C",
|
||||
# path_prefix=None,
|
||||
# environ_update=None,
|
||||
# ),
|
||||
# runner_ctx_args = dict(
|
||||
# args_order=['aa', 'bb'],
|
||||
# output_process=None,
|
||||
# ignore_value_none=True,
|
||||
# ),
|
||||
# ),
|
||||
# # command execution
|
||||
# dict(
|
||||
# runner_ctx_run_args = dict(bb=True),
|
||||
# rc = 0,
|
||||
# out = "",
|
||||
# err = "",
|
||||
# ),
|
||||
# # expected
|
||||
# dict(
|
||||
# results=(),
|
||||
# run_info=dict(
|
||||
# cmd=['/mock/bin/testing', '--answer=11', '--bb-here'],
|
||||
# environ_update={'LANGUAGE': 'C', 'LC_ALL': 'C'},
|
||||
# ),
|
||||
# exc=None,
|
||||
# ),
|
||||
# ),
|
||||
#
|
||||
aa_bb=(
|
||||
dict(
|
||||
args_bundle=dict(
|
||||
aa=dict(type="int", value=11, fmt_func=fmt.as_opt_eq_val, fmt_arg="--answer"),
|
||||
bb=dict(fmt_func=fmt.as_bool, fmt_arg="--bb-here"),
|
||||
),
|
||||
runner_init_args=dict(),
|
||||
runner_ctx_args=dict(args_order=['aa', 'bb']),
|
||||
),
|
||||
dict(runner_ctx_run_args=dict(bb=True), rc=0, out="", err=""),
|
||||
dict(
|
||||
run_info=dict(
|
||||
cmd=['/mock/bin/testing', '--answer=11', '--bb-here'],
|
||||
environ_update={'LANGUAGE': 'C', 'LC_ALL': 'C'},
|
||||
args_order=('aa', 'bb'),
|
||||
),
|
||||
),
|
||||
),
|
||||
aa_bb_default_order=(
|
||||
dict(
|
||||
args_bundle=dict(
|
||||
aa=dict(type="int", value=11, fmt_func=fmt.as_opt_eq_val, fmt_arg="--answer"),
|
||||
bb=dict(fmt_func=fmt.as_bool, fmt_arg="--bb-here"),
|
||||
),
|
||||
runner_init_args=dict(default_args_order=['bb', 'aa']),
|
||||
runner_ctx_args=dict(),
|
||||
),
|
||||
dict(runner_ctx_run_args=dict(bb=True), rc=0, out="", err=""),
|
||||
dict(
|
||||
run_info=dict(
|
||||
cmd=['/mock/bin/testing', '--bb-here', '--answer=11'],
|
||||
environ_update={'LANGUAGE': 'C', 'LC_ALL': 'C'},
|
||||
args_order=('bb', 'aa'),
|
||||
),
|
||||
),
|
||||
),
|
||||
aa_bb_default_order_args_order=(
|
||||
dict(
|
||||
args_bundle=dict(
|
||||
aa=dict(type="int", value=11, fmt_func=fmt.as_opt_eq_val, fmt_arg="--answer"),
|
||||
bb=dict(fmt_func=fmt.as_bool, fmt_arg="--bb-here"),
|
||||
),
|
||||
runner_init_args=dict(default_args_order=['bb', 'aa']),
|
||||
runner_ctx_args=dict(args_order=['aa', 'bb']),
|
||||
),
|
||||
dict(runner_ctx_run_args=dict(bb=True), rc=0, out="", err=""),
|
||||
dict(
|
||||
run_info=dict(
|
||||
cmd=['/mock/bin/testing', '--answer=11', '--bb-here'],
|
||||
environ_update={'LANGUAGE': 'C', 'LC_ALL': 'C'},
|
||||
args_order=('aa', 'bb'),
|
||||
),
|
||||
),
|
||||
),
|
||||
aa_bb_dup_in_args_order=(
|
||||
dict(
|
||||
args_bundle=dict(
|
||||
aa=dict(type="int", value=11, fmt_func=fmt.as_opt_eq_val, fmt_arg="--answer"),
|
||||
bb=dict(fmt_func=fmt.as_bool, fmt_arg="--bb-here"),
|
||||
),
|
||||
runner_init_args=dict(),
|
||||
runner_ctx_args=dict(args_order=['aa', 'bb', 'aa']),
|
||||
),
|
||||
dict(runner_ctx_run_args=dict(bb=True), rc=0, out="", err=""),
|
||||
dict(
|
||||
run_info=dict(
|
||||
cmd=['/mock/bin/testing', '--answer=11', '--bb-here', '--answer=11'],
|
||||
),
|
||||
),
|
||||
),
|
||||
aa_bb_process_output=(
|
||||
dict(
|
||||
args_bundle=dict(
|
||||
aa=dict(type="int", value=11, fmt_func=fmt.as_opt_eq_val, fmt_arg="--answer"),
|
||||
bb=dict(fmt_func=fmt.as_bool, fmt_arg="--bb-here"),
|
||||
),
|
||||
runner_init_args=dict(default_args_order=['bb', 'aa']),
|
||||
runner_ctx_args=dict(
|
||||
args_order=['aa', 'bb'],
|
||||
output_process=lambda rc, out, err: '-/-'.join([str(rc), out, err])
|
||||
),
|
||||
),
|
||||
dict(runner_ctx_run_args=dict(bb=True), rc=0, out="ni", err="nu"),
|
||||
dict(
|
||||
run_info=dict(
|
||||
cmd=['/mock/bin/testing', '--answer=11', '--bb-here'],
|
||||
),
|
||||
results="0-/-ni-/-nu"
|
||||
),
|
||||
),
|
||||
aa_bb_ignore_none_with_none=(
|
||||
dict(
|
||||
args_bundle=dict(
|
||||
aa=dict(type="int", value=49, fmt_func=fmt.as_opt_eq_val, fmt_arg="--answer"),
|
||||
bb=dict(fmt_func=fmt.as_bool, fmt_arg="--bb-here"),
|
||||
),
|
||||
runner_init_args=dict(default_args_order=['bb', 'aa']),
|
||||
runner_ctx_args=dict(
|
||||
args_order=['aa', 'bb'],
|
||||
ignore_value_none=True, # default
|
||||
),
|
||||
),
|
||||
dict(runner_ctx_run_args=dict(bb=None), rc=0, out="ni", err="nu"),
|
||||
dict(
|
||||
run_info=dict(
|
||||
cmd=['/mock/bin/testing', '--answer=49'],
|
||||
),
|
||||
),
|
||||
),
|
||||
aa_bb_ignore_not_none_with_none=(
|
||||
dict(
|
||||
args_bundle=dict(
|
||||
aa=dict(type="int", value=49, fmt_func=fmt.as_opt_eq_val, fmt_arg="--answer"),
|
||||
bb=dict(fmt_func=fmt.as_bool, fmt_arg="--bb-here"),
|
||||
),
|
||||
runner_init_args=dict(default_args_order=['bb', 'aa']),
|
||||
runner_ctx_args=dict(
|
||||
args_order=['aa', 'bb'],
|
||||
ignore_value_none=False,
|
||||
),
|
||||
),
|
||||
dict(runner_ctx_run_args=dict(aa=None, bb=True), rc=0, out="ni", err="nu"),
|
||||
dict(
|
||||
run_info=dict(
|
||||
cmd=['/mock/bin/testing', '--answer=None', '--bb-here'],
|
||||
),
|
||||
),
|
||||
),
|
||||
)
|
||||
TC_RUNNER_IDS = sorted(TC_RUNNER.keys())
|
||||
|
||||
|
||||
@pytest.mark.parametrize('runner_input, cmd_execution, expected',
|
||||
(TC_RUNNER[tc] for tc in TC_RUNNER_IDS),
|
||||
ids=TC_RUNNER_IDS)
|
||||
def test_runner(runner_input, cmd_execution, expected):
|
||||
arg_spec = {}
|
||||
params = {}
|
||||
arg_formats = {}
|
||||
for k, v in runner_input['args_bundle'].items():
|
||||
try:
|
||||
arg_spec[k] = {'type': v['type']}
|
||||
except KeyError:
|
||||
pass
|
||||
try:
|
||||
params[k] = v['value']
|
||||
except KeyError:
|
||||
pass
|
||||
try:
|
||||
arg_formats[k] = v['fmt_func'](v['fmt_arg'])
|
||||
except KeyError:
|
||||
pass
|
||||
|
||||
orig_results = tuple(cmd_execution[x] for x in ('rc', 'out', 'err'))
|
||||
|
||||
print("arg_spec={0}\nparams={1}\narg_formats={2}\n".format(
|
||||
arg_spec,
|
||||
params,
|
||||
arg_formats,
|
||||
))
|
||||
|
||||
module = MagicMock()
|
||||
type(module).argument_spec = PropertyMock(return_value=arg_spec)
|
||||
type(module).params = PropertyMock(return_value=params)
|
||||
module.get_bin_path.return_value = '/mock/bin/testing'
|
||||
module.run_command.return_value = orig_results
|
||||
|
||||
runner = CmdRunner(
|
||||
module=module,
|
||||
command="testing",
|
||||
arg_formats=arg_formats,
|
||||
**runner_input['runner_init_args']
|
||||
)
|
||||
|
||||
def _assert_run_info(actual, expected):
|
||||
reduced = dict((k, actual[k]) for k in expected.keys())
|
||||
assert reduced == expected, "{0}".format(reduced)
|
||||
|
||||
def _assert_run(runner_input, cmd_execution, expected, ctx, results):
|
||||
_assert_run_info(ctx.run_info, expected['run_info'])
|
||||
assert results == expected.get('results', orig_results)
|
||||
|
||||
exc = expected.get("exc")
|
||||
if exc:
|
||||
with pytest.raises(exc):
|
||||
with runner.context(**runner_input['runner_ctx_args']) as ctx:
|
||||
results = ctx.run(**cmd_execution['runner_ctx_run_args'])
|
||||
_assert_run(runner_input, cmd_execution, expected, ctx, results)
|
||||
|
||||
else:
|
||||
with runner.context(**runner_input['runner_ctx_args']) as ctx:
|
||||
results = ctx.run(**cmd_execution['runner_ctx_run_args'])
|
||||
_assert_run(runner_input, cmd_execution, expected, ctx, results)
|
||||
Loading…
Reference in a new issue