ansible/community.general
1
0
Fork 0
mirror of https://github.com/ansible-collections/community.general.git synced 2024-09-14 20:13:21 +02:00
Code Releases Activity
community.general/tests/unit/plugins/module_utils/test_cmd_runner.py
Alexei Znamensky f5b1b3c6f0
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>
2022-04-25 22:12:00 +02:00

306 lines
12 KiB
Python
Raw Blame History

# -*- 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)
View git blame Copy permalink