internet/community.general
3
0
Fork 0
mirror of https://github.com/ansible-collections/community.general.git synced 2026-05-06 13:22:48 +00:00
Code Issues Packages Projects Releases Wiki Activity

Save the command line arguments into a global context

Browse Source 
* Once cli args are parsed, they're constant.  So, save the parsed args
  into the global context for everyone else to use them from now on.
* Port cli scripts to use the CLIARGS in the context
* Refactor call to parse cli args into the run() method
* Fix unittests for changes to the internals of CLI arg parsing
* Port callback plugins to use context.CLIARGS
  * Got rid of the private self._options attribute
  * Use context.CLIARGS in the individual callback plugins instead.
  * Also output positional arguments in default and unixy plugins
  * Code has been simplified since we're now dealing with a dict rather
    than Optparse.Value
This commit is contained in:
Toshio Kuratomi
2018-12-17 18:10:59 -08:00
parent c18da65089
commit afdbb0d9d5
36 changed files with 1033 additions and 868 deletions
Download Patch File Download Diff File Expand all files Collapse all files

15
test/units/cli/test_adhoc.py
View File

@@ -5,6 +5,8 @@ from __future__ import (absolute_import, division, print_function)
__metaclass__ = type
import pytest
from ansible import context
from ansible.cli.adhoc import AdHocCLI, display
from ansible.errors import AnsibleOptionsError
@@ -22,7 +24,7 @@ def test_with_command():
module_name = 'command'
adhoc_cli = AdHocCLI(args=['-m', module_name, '-vv'])
adhoc_cli.parse()
assert adhoc_cli.options.module_name == module_name
assert context.CLIARGS['module_name'] == module_name
assert display.verbosity == 2
@@ -36,9 +38,8 @@ def test_with_extra_parameters():
def test_simple_command():
""" Test valid command and its run"""
adhoc_cli = AdHocCLI(['/bin/ansible', '-m', 'command', 'localhost'])
adhoc_cli = AdHocCLI(['/bin/ansible', '-m', 'command', 'localhost', '-a', 'echo "hi"'])
adhoc_cli.parse()
adhoc_cli.options.module_args = "echo 'hi'"
ret = adhoc_cli.run()
assert ret == 0
@@ -63,9 +64,8 @@ def test_did_you_mean_playbook():
def test_play_ds_positive():
""" Test _play_ds"""
adhoc_cli = AdHocCLI(args=['/bin/ansible', 'localhost'])
adhoc_cli = AdHocCLI(args=['/bin/ansible', 'localhost', '-m', 'command'])
adhoc_cli.parse()
adhoc_cli.options.module_name = 'command'
ret = adhoc_cli._play_ds('command', 10, 2)
assert ret['name'] == 'Ansible Ad-Hoc'
assert ret['tasks'] == [{'action': {'module': 'command', 'args': {}}, 'async_val': 10, 'poll': 2}]
@@ -73,9 +73,8 @@ def test_play_ds_positive():
def test_play_ds_with_include_role():
""" Test include_role command with poll"""
adhoc_cli = AdHocCLI(args=['/bin/ansible', 'localhost'])
adhoc_cli = AdHocCLI(args=['/bin/ansible', 'localhost', '-m', 'include_role'])
adhoc_cli.parse()
adhoc_cli.options.module_name = 'include_role'
ret = adhoc_cli._play_ds('include_role', None, 2)
assert ret['name'] == 'Ansible Ad-Hoc'
assert ret['gather_facts'] == 'no'
@@ -88,5 +87,5 @@ def test_run_import_playbook():
adhoc_cli.parse()
with pytest.raises(AnsibleOptionsError) as exec_info:
adhoc_cli.run()
assert adhoc_cli.options.module_name == import_playbook
assert context.CLIARGS['module_name'] == import_playbook
assert "'%s' is not a valid action for ad-hoc commands" % import_playbook == str(exec_info.value)

67
test/units/cli/test_galaxy.py
View File

@@ -26,6 +26,8 @@ import tarfile
import tempfile
import yaml
from ansible import arguments
from ansible import context
from ansible.cli.galaxy import GalaxyCLI
from units.compat import unittest
from units.compat.mock import call, patch
@@ -47,7 +49,6 @@ class TestGalaxy(unittest.TestCase):
# creating framework for a role
gc = GalaxyCLI(args=["ansible-galaxy", "init", "--offline", "delete_me"])
gc.parse()
gc.run()
cls.role_dir = "./delete_me"
cls.role_name = "delete_me"
@@ -96,8 +97,14 @@ class TestGalaxy(unittest.TestCase):
shutil.rmtree(cls.temp_dir)
def setUp(self):
# Reset the stored command line args
arguments.GlobalCLIArgs._Singleton__instance = None
self.default_args = ['ansible-galaxy']
def tearDown(self):
# Reset the stored command line args
arguments.GlobalCLIArgs._Singleton__instance = None
def test_init(self):
galaxy_cli = GalaxyCLI(args=self.default_args)
self.assertTrue(isinstance(galaxy_cli, GalaxyCLI))
@@ -120,12 +127,11 @@ class TestGalaxy(unittest.TestCase):
def test_run(self):
''' verifies that the GalaxyCLI object's api is created and that execute() is called. '''
gc = GalaxyCLI(args=["ansible-galaxy", "install", "--ignore-errors", "imaginary_role"])
gc.parse()
with patch.object(ansible.cli.CLI, "execute", return_value=None) as mock_ex:
with patch.object(ansible.cli.CLI, "run", return_value=None) as mock_run:
gc.run()
# testing
self.assertIsInstance(gc.galaxy, ansible.galaxy.Galaxy)
self.assertEqual(mock_run.call_count, 1)
self.assertTrue(isinstance(gc.api, ansible.galaxy.api.GalaxyAPI))
self.assertEqual(mock_ex.call_count, 1)
@@ -133,15 +139,16 @@ class TestGalaxy(unittest.TestCase):
def test_execute_remove(self):
# installing role
gc = GalaxyCLI(args=["ansible-galaxy", "install", "-p", self.role_path, "-r", self.role_req, '--force'])
gc.parse()
gc.run()
# location where the role was installed
role_file = os.path.join(self.role_path, self.role_name)
# removing role
# Have to reset the arguments in the context object manually since we're doing the
# equivalent of running the command line program twice
arguments.GlobalCLIArgs._Singleton__instance = None
gc = GalaxyCLI(args=["ansible-galaxy", "remove", role_file, self.role_name])
gc.parse()
gc.run()
# testing role was removed
@@ -151,7 +158,6 @@ class TestGalaxy(unittest.TestCase):
def test_exit_without_ignore_without_flag(self):
''' tests that GalaxyCLI exits with the error specified if the --ignore-errors flag is not used '''
gc = GalaxyCLI(args=["ansible-galaxy", "install", "--server=None", "fake_role_name"])
gc.parse()
with patch.object(ansible.utils.display.Display, "display", return_value=None) as mocked_display:
# testing that error expected is raised
self.assertRaises(AnsibleError, gc.run)
@@ -161,7 +167,6 @@ class TestGalaxy(unittest.TestCase):
''' tests that GalaxyCLI exits without the error specified if the --ignore-errors flag is used '''
# testing with --ignore-errors flag
gc = GalaxyCLI(args=["ansible-galaxy", "install", "--server=None", "fake_role_name", "--ignore-errors"])
gc.parse()
with patch.object(ansible.utils.display.Display, "display", return_value=None) as mocked_display:
gc.run()
self.assertTrue(mocked_display.called_once_with("- downloading role 'fake_role_name', owned by "))
@@ -172,7 +177,6 @@ class TestGalaxy(unittest.TestCase):
# checking that the common results of parse() for all possible actions have been created/called
self.assertIsInstance(galaxycli_obj.parser, ansible.cli.SortedOptParser)
self.assertIsInstance(galaxycli_obj.galaxy, ansible.galaxy.Galaxy)
formatted_call = {
'import': 'usage: %prog import [options] github_user github_repo',
'delete': 'usage: %prog delete [options] github_user github_repo',
@@ -206,74 +210,74 @@ class TestGalaxy(unittest.TestCase):
''' testing the options parser when the action 'delete' is given '''
gc = GalaxyCLI(args=["ansible-galaxy", "delete"])
self.run_parse_common(gc, "delete")
self.assertEqual(gc.options.verbosity, 0)
self.assertEqual(context.CLIARGS['verbosity'], 0)
def test_parse_import(self):
''' testing the options parser when the action 'import' is given '''
gc = GalaxyCLI(args=["ansible-galaxy", "import"])
self.run_parse_common(gc, "import")
self.assertEqual(gc.options.wait, True)
self.assertEqual(gc.options.reference, None)
self.assertEqual(gc.options.check_status, False)
self.assertEqual(gc.options.verbosity, 0)
self.assertEqual(context.CLIARGS['wait'], True)
self.assertEqual(context.CLIARGS['reference'], None)
self.assertEqual(context.CLIARGS['check_status'], False)
self.assertEqual(context.CLIARGS['verbosity'], 0)
def test_parse_info(self):
''' testing the options parser when the action 'info' is given '''
gc = GalaxyCLI(args=["ansible-galaxy", "info"])
self.run_parse_common(gc, "info")
self.assertEqual(gc.options.offline, False)
self.assertEqual(context.CLIARGS['offline'], False)
def test_parse_init(self):
''' testing the options parser when the action 'init' is given '''
gc = GalaxyCLI(args=["ansible-galaxy", "init"])
self.run_parse_common(gc, "init")
self.assertEqual(gc.options.offline, False)
self.assertEqual(gc.options.force, False)
self.assertEqual(context.CLIARGS['offline'], False)
self.assertEqual(context.CLIARGS['force'], False)
def test_parse_install(self):
''' testing the options parser when the action 'install' is given '''
gc = GalaxyCLI(args=["ansible-galaxy", "install"])
self.run_parse_common(gc, "install")
self.assertEqual(gc.options.ignore_errors, False)
self.assertEqual(gc.options.no_deps, False)
self.assertEqual(gc.options.role_file, None)
self.assertEqual(gc.options.force, False)
self.assertEqual(context.CLIARGS['ignore_errors'], False)
self.assertEqual(context.CLIARGS['no_deps'], False)
self.assertEqual(context.CLIARGS['role_file'], None)
self.assertEqual(context.CLIARGS['force'], False)
def test_parse_list(self):
''' testing the options parser when the action 'list' is given '''
gc = GalaxyCLI(args=["ansible-galaxy", "list"])
self.run_parse_common(gc, "list")
self.assertEqual(gc.options.verbosity, 0)
self.assertEqual(context.CLIARGS['verbosity'], 0)
def test_parse_login(self):
''' testing the options parser when the action 'login' is given '''
gc = GalaxyCLI(args=["ansible-galaxy", "login"])
self.run_parse_common(gc, "login")
self.assertEqual(gc.options.verbosity, 0)
self.assertEqual(gc.options.token, None)
self.assertEqual(context.CLIARGS['verbosity'], 0)
self.assertEqual(context.CLIARGS['token'], None)
def test_parse_remove(self):
''' testing the options parser when the action 'remove' is given '''
gc = GalaxyCLI(args=["ansible-galaxy", "remove"])
self.run_parse_common(gc, "remove")
self.assertEqual(gc.options.verbosity, 0)
self.assertEqual(context.CLIARGS['verbosity'], 0)
def test_parse_search(self):
''' testing the options parswer when the action 'search' is given '''
gc = GalaxyCLI(args=["ansible-galaxy", "search"])
self.run_parse_common(gc, "search")
self.assertEqual(gc.options.platforms, None)
self.assertEqual(gc.options.galaxy_tags, None)
self.assertEqual(gc.options.author, None)
self.assertEqual(context.CLIARGS['platforms'], None)
self.assertEqual(context.CLIARGS['galaxy_tags'], None)
self.assertEqual(context.CLIARGS['author'], None)
def test_parse_setup(self):
''' testing the options parser when the action 'setup' is given '''
gc = GalaxyCLI(args=["ansible-galaxy", "setup"])
self.run_parse_common(gc, "setup")
self.assertEqual(gc.options.verbosity, 0)
self.assertEqual(gc.options.remove_id, None)
self.assertEqual(gc.options.setup_list, False)
self.assertEqual(context.CLIARGS['verbosity'], 0)
self.assertEqual(context.CLIARGS['remove_id'], None)
self.assertEqual(context.CLIARGS['setup_list'], False)
class ValidRoleTests(object):
@@ -299,7 +303,6 @@ class ValidRoleTests(object):
# create role using default skeleton
gc = GalaxyCLI(args=['ansible-galaxy', 'init', '-c', '--offline'] + galaxy_args + ['--init-path', cls.test_dir, cls.role_name])
gc.parse()
gc.run()
cls.gc = gc
@@ -466,4 +469,4 @@ class TestGalaxyInitSkeleton(unittest.TestCase, ValidRoleTests):
self.assertTrue(os.path.exists(os.path.join(self.role_dir, 'templates_extra', 'templates.txt')))
def test_skeleton_option(self):
self.assertEquals(self.role_skeleton_path, self.gc.options.role_skeleton, msg='Skeleton path was not parsed properly from the command line')
self.assertEquals(self.role_skeleton_path, context.CLIARGS['role_skeleton'], msg='Skeleton path was not parsed properly from the command line')

3
test/units/cli/test_playbook.py
View File

@@ -22,6 +22,7 @@ __metaclass__ = type
from units.compat import unittest
from units.mock.loader import DictDataLoader
from ansible import context
from ansible.inventory.manager import InventoryManager
from ansible.vars.manager import VariableManager
@@ -32,7 +33,7 @@ class TestPlaybookCLI(unittest.TestCase):
def test_flush_cache(self):
cli = PlaybookCLI(args=["ansible-playbook", "--flush-cache", "foobar.yml"])
cli.parse()
self.assertTrue(cli.options.flush_cache)
self.assertTrue(context.CLIARGS['flush_cache'])
variable_manager = VariableManager()
fake_loader = DictDataLoader({'foobar.yml': ""})

14
test/units/executor/test_playbook_executor.py
View File

@@ -21,6 +21,8 @@ __metaclass__ = type
from units.compat import unittest
from units.compat.mock import MagicMock
from ansible import arguments
from ansible.executor.playbook_executor import PlaybookExecutor
from ansible.playbook import Playbook
from ansible.template import Templar
@@ -31,10 +33,12 @@ from units.mock.loader import DictDataLoader
class TestPlaybookExecutor(unittest.TestCase):
def setUp(self):
pass
# Reset command line args for every test
arguments.CLIArgs._Singleton__instance = None
def tearDown(self):
pass
# And cleanup after ourselves too
arguments.CLIArgs._Singleton__instance = None
def test_get_serialized_batches(self):
fake_loader = DictDataLoader({
@@ -77,11 +81,6 @@ class TestPlaybookExecutor(unittest.TestCase):
mock_inventory = MagicMock()
mock_var_manager = MagicMock()
# fake out options to use the syntax CLI switch, which will ensure
# the PlaybookExecutor doesn't create a TaskQueueManager
mock_options = MagicMock()
mock_options.syntax.value = True
templar = Templar(loader=fake_loader)
pbe = PlaybookExecutor(
@@ -89,7 +88,6 @@ class TestPlaybookExecutor(unittest.TestCase):
inventory=mock_inventory,
variable_manager=mock_var_manager,
loader=fake_loader,
options=mock_options,
passwords=[],
)

11
test/units/executor/test_task_queue_manager_callbacks.py
View File

@@ -20,6 +20,9 @@ from __future__ import (absolute_import, division, print_function)
from units.compat import unittest
from units.compat.mock import MagicMock
from ansible import arguments
from ansible import context
from ansible.executor.task_queue_manager import TaskQueueManager
from ansible.playbook import Playbook
from ansible.plugins.callback import CallbackBase
@@ -32,10 +35,11 @@ class TestTaskQueueManagerCallbacks(unittest.TestCase):
inventory = MagicMock()
variable_manager = MagicMock()
loader = MagicMock()
options = MagicMock()
passwords = []
self._tqm = TaskQueueManager(inventory, variable_manager, loader, options, passwords)
# Reset the stored command line args
arguments.GlobalCLIArgs._Singleton__instance = None
self._tqm = TaskQueueManager(inventory, variable_manager, loader, passwords)
self._playbook = Playbook(loader)
# we use a MagicMock to register the result of the call we
@@ -46,7 +50,8 @@ class TestTaskQueueManagerCallbacks(unittest.TestCase):
self._register = MagicMock()
def tearDown(self):
pass
# Reset the stored command line args
arguments.GlobalCLIArgs._Singleton__instance = None
def test_task_queue_manager_callbacks_v2_playbook_on_start(self):
"""

29
test/units/playbook/test_play_context.py
View File

@@ -11,8 +11,10 @@ import os
import pytest
from ansible import arguments
from ansible import constants as C
from ansible.cli import CLI
from ansible import context
from ansible import cli
from units.compat import unittest
from ansible.errors import AnsibleError, AnsibleParserError
from ansible.module_utils.six.moves import shlex_quote
@@ -23,7 +25,7 @@ from units.mock.loader import DictDataLoader
@pytest.fixture
def parser():
parser = CLI.base_parser(runas_opts=True, meta_opts=True,
parser = cli.base_parser(runas_opts=True, meta_opts=True,
runtask_opts=True, vault_opts=True,
async_opts=True, connect_opts=True,
subset_opts=True, check_opts=True,
@@ -31,9 +33,18 @@ def parser():
return parser
def test_play_context(mocker, parser):
@pytest.fixture
def reset_cli_args():
arguments.GlobalCLIArgs._Singleton__instance = None
yield
arguments.GlobalCLIArgs._Singleton__instance = None
def test_play_context(mocker, parser, reset_cli_args):
(options, args) = parser.parse_args(['-vv', '--check'])
play_context = PlayContext(options=options)
options.args = args
context._init_global_context(options)
play_context = PlayContext()
assert play_context._attributes['connection'] == C.DEFAULT_TRANSPORT
assert play_context.remote_addr is None
@@ -56,7 +67,7 @@ def test_play_context(mocker, parser):
mock_play.become_user = 'mockroot'
mock_play.no_log = True
play_context = PlayContext(play=mock_play, options=options)
play_context = PlayContext(play=mock_play)
assert play_context.connection == 'mock'
assert play_context.remote_user == 'mock'
assert play_context.password == ''
@@ -83,7 +94,7 @@ def test_play_context(mocker, parser):
mock_templar = mocker.MagicMock()
play_context = PlayContext(play=mock_play, options=options)
play_context = PlayContext(play=mock_play)
play_context = play_context.set_task_and_variable_override(task=mock_task, variables=all_vars, templar=mock_templar)
assert play_context.connection == 'mock_inventory'
@@ -100,9 +111,11 @@ def test_play_context(mocker, parser):
assert play_context.no_log is False
def test_play_context_make_become_cmd(parser):
def test_play_context_make_become_cmd(mocker, parser, reset_cli_args):
(options, args) = parser.parse_args([])
play_context = PlayContext(options=options)
options.args = args
context._init_global_context(options)
play_context = PlayContext()
default_cmd = "/bin/foo"
default_exe = "/bin/bash"

10
test/units/plugins/strategy/test_strategy_base.py
View File

@@ -66,7 +66,6 @@ class TestStrategyBase(unittest.TestCase):
mock_tqm = MagicMock(TaskQueueManager)
mock_tqm._final_q = mock_queue
mock_tqm._options = MagicMock()
strategy_base = StrategyBase(tqm=mock_tqm)
strategy_base.cleanup()
@@ -106,7 +105,6 @@ class TestStrategyBase(unittest.TestCase):
mock_tqm._failed_hosts = dict()
mock_tqm._unreachable_hosts = dict()
mock_tqm._options = MagicMock()
strategy_base = StrategyBase(tqm=mock_tqm)
mock_host = MagicMock()
@@ -187,15 +185,13 @@ class TestStrategyBase(unittest.TestCase):
mock_host.has_hostkey = True
mock_inventory = MagicMock()
mock_inventory.get.return_value = mock_host
mock_options = MagicMock()
mock_options.module_path = None
tqm = TaskQueueManager(
inventory=mock_inventory,
variable_manager=mock_var_manager,
loader=fake_loader,
options=mock_options,
passwords=None,
forks=5,
)
tqm._initialize_processes(3)
tqm.hostvars = dict()
@@ -520,15 +516,13 @@ class TestStrategyBase(unittest.TestCase):
mock_iterator._play = mock_play
fake_loader = DictDataLoader()
mock_options = MagicMock()
mock_options.module_path = None
tqm = TaskQueueManager(
inventory=mock_inventory,
variable_manager=mock_var_mgr,
loader=fake_loader,
options=mock_options,
passwords=None,
forks=5,
)
tqm._initialize_processes(3)
tqm._initialize_notified_handlers(mock_play)

5
test/units/plugins/strategy/test_strategy_linear.py
View File

@@ -80,15 +80,12 @@ class TestStrategyLinear(unittest.TestCase):
all_vars=dict(),
)
mock_options = MagicMock()
mock_options.module_path = None
tqm = TaskQueueManager(
inventory=inventory,
variable_manager=mock_var_manager,
loader=fake_loader,
options=mock_options,
passwords=None,
forks=5,
)
tqm._initialize_processes(3)
strategy = StrategyModule(tqm)

19
test/units/test_arguments.py
View File

@@ -27,7 +27,7 @@ MAKE_IMMUTABLE_DATA = ((u'くらとみ', u'くらとみ'),
arguments.ImmutableDict({u'café': (1, frozenset(u'ñ'))})),
([set((1, 2)), {u'くらとみ': 3}],
(frozenset((1, 2)), arguments.ImmutableDict({u'くらとみ': 3}))),
)
)
@pytest.mark.parametrize('data, expected', MAKE_IMMUTABLE_DATA)
@@ -35,6 +35,17 @@ def test_make_immutable(data, expected):
assert arguments._make_immutable(data) == expected
def test_cliargs_from_dict():
old_dict = {'tags': [u'production', u'webservers'],
'check_mode': True,
'start_at_task': u'Start with くらとみ'}
expected = frozenset((('tags', (u'production', u'webservers')),
('check_mode', True),
('start_at_task', u'Start with くらとみ')))
assert frozenset(arguments.CLIArgs(old_dict).items()) == expected
def test_cliargs():
class FakeOptions:
pass
@@ -47,7 +58,7 @@ def test_cliargs():
('check_mode', True),
('start_at_task', u'Start with くらとみ')))
assert frozenset(arguments.CLIArgs(options).items()) == expected
assert frozenset(arguments.CLIArgs.from_options(options).items()) == expected
@pytest.mark.skipIf(argparse is None)
@@ -69,8 +80,8 @@ def test_cliargs_argparse():
def test_cliargs_optparse():
parser = optparse.OptionParser(description='Process some integers.')
parser.add_option('--sum', dest='accumulate', action='store_const',
const=sum, default=max,
help='sum the integers (default: find the max)')
const=sum, default=max,
help='sum the integers (default: find the max)')
opts, args = parser.parse_args([u'--sum', u'1', u'2'])
opts.integers = args

30
test/units/test_context.py Normal file
View File

@@ -0,0 +1,30 @@
# -*- coding: utf-8 -*-
# Copyright: (c) 2018, Toshio Kuratomi <tkuratomi@ansible.com>
# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
# Make coding more python3-ish
from __future__ import (absolute_import, division)
__metaclass__ = type
import pytest
from ansible import context
class FakeOptions:
pass
def test_set_global_context():
options = FakeOptions()
options.tags = [u'production', u'webservers']
options.check_mode = True
options.start_at_task = u'Start with くらとみ'
expected = frozenset((('tags', (u'production', u'webservers')),
('check_mode', True),
('start_at_task', u'Start with くらとみ')))
context._init_global_context(options)
assert frozenset(context.CLIARGS.items()) == expected