# (c) 2012-2014, Michael DeHaan # # This file is part of Ansible # # Ansible is free software: you can redistribute it and/or modify # it under the terms of the GNU General Public License as published by # the Free Software Foundation, either version 3 of the License, or # (at your option) any later version. # # Ansible is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU General Public License for more details. # # You should have received a copy of the GNU General Public License # along with Ansible. If not, see . # Make coding more python3-ish from __future__ import (absolute_import, division, print_function) __metaclass__ = type from ansible.compat.tests import unittest from ansible.compat.tests.mock import patch, MagicMock from ansible.errors import AnsibleError, AnsibleParserError from ansible.executor.connection_info import ConnectionInformation from ansible.executor.task_executor import TaskExecutor from ansible.plugins import action_loader from units.mock.loader import DictDataLoader class TestTaskExecutor(unittest.TestCase): def setUp(self): pass def tearDown(self): pass def test_task_executor_init(self): fake_loader = DictDataLoader({}) mock_host = MagicMock() mock_task = MagicMock() mock_conn_info = MagicMock() mock_shared_loader = MagicMock() new_stdin = None job_vars = dict() te = TaskExecutor( host = mock_host, task = mock_task, job_vars = job_vars, connection_info = mock_conn_info, new_stdin = new_stdin, loader = fake_loader, shared_loader_obj = mock_shared_loader, ) def test_task_executor_run(self): fake_loader = DictDataLoader({}) mock_host = MagicMock() mock_task = MagicMock() mock_task._role._role_path = '/path/to/role/foo' mock_conn_info = MagicMock() mock_shared_loader = MagicMock() new_stdin = None job_vars = dict() te = TaskExecutor( host = mock_host, task = mock_task, job_vars = job_vars, connection_info = mock_conn_info, new_stdin = new_stdin, loader = fake_loader, shared_loader_obj = mock_shared_loader, ) te._get_loop_items = MagicMock(return_value=None) te._execute = MagicMock(return_value=dict()) res = te.run() te._get_loop_items = MagicMock(return_value=[]) res = te.run() te._get_loop_items = MagicMock(return_value=['a','b','c']) te._run_loop = MagicMock(return_value=[dict(item='a', changed=True), dict(item='b', failed=True), dict(item='c')]) res = te.run() te._get_loop_items = MagicMock(side_effect=AnsibleError("")) res = te.run() self.assertIn("failed", res) def test_task_executor_get_loop_items(self): fake_loader = DictDataLoader({}) mock_host = MagicMock() mock_task = MagicMock() mock_task.loop = 'items' mock_task.loop_args = ['a', 'b', 'c'] mock_conn_info = MagicMock() mock_shared_loader = MagicMock() new_stdin = None job_vars = dict() te = TaskExecutor( host = mock_host, task = mock_task, job_vars = job_vars, connection_info = mock_conn_info, new_stdin = new_stdin, loader = fake_loader, shared_loader_obj = mock_shared_loader, ) items = te._get_loop_items() self.assertEqual(items, ['a', 'b', 'c']) def test_task_executor_run_loop(self): items = ['a', 'b', 'c'] fake_loader = DictDataLoader({}) mock_host = MagicMock() def _copy(): new_item = MagicMock() return new_item mock_task = MagicMock() mock_task.copy.side_effect = _copy mock_conn_info = MagicMock() mock_shared_loader = MagicMock() new_stdin = None job_vars = dict() te = TaskExecutor( host = mock_host, task = mock_task, job_vars = job_vars, connection_info = mock_conn_info, new_stdin = new_stdin, loader = fake_loader, shared_loader_obj = mock_shared_loader, ) def _execute(variables): return dict(item=variables.get('item')) te._squash_items = MagicMock(return_value=items) te._execute = MagicMock(side_effect=_execute) res = te._run_loop(items) self.assertEqual(len(res), 3) def test_task_executor_squash_items(self): items = ['a', 'b', 'c'] fake_loader = DictDataLoader({}) mock_host = MagicMock() def _evaluate_conditional(templar, variables): item = variables.get('item') if item == 'b': return False return True mock_task = MagicMock() mock_task.evaluate_conditional.side_effect = _evaluate_conditional mock_conn_info = MagicMock() mock_shared_loader = None new_stdin = None job_vars = dict() te = TaskExecutor( host = mock_host, task = mock_task, job_vars = job_vars, connection_info = mock_conn_info, new_stdin = new_stdin, loader = fake_loader, shared_loader_obj = mock_shared_loader, ) mock_task.action = 'foo' new_items = te._squash_items(items=items, variables=job_vars) self.assertEqual(new_items, ['a', 'b', 'c']) mock_task.action = 'yum' new_items = te._squash_items(items=items, variables=job_vars) self.assertEqual(new_items, ['a,c']) def test_task_executor_execute(self): fake_loader = DictDataLoader({}) mock_host = MagicMock() mock_task = MagicMock() mock_task.args = dict() mock_task.retries = 0 mock_task.delay = -1 mock_task.register = 'foo' mock_task.until = None mock_task.changed_when = None mock_task.failed_when = None mock_task.post_validate.return_value = None mock_conn_info = MagicMock() mock_conn_info.post_validate.return_value = None mock_conn_info.update_vars.return_value = None mock_connection = MagicMock() mock_connection.set_host_overrides.return_value = None mock_connection._connect.return_value = None mock_action = MagicMock() shared_loader = None new_stdin = None job_vars = dict(omit="XXXXXXXXXXXXXXXXXXX") te = TaskExecutor( host = mock_host, task = mock_task, job_vars = job_vars, connection_info = mock_conn_info, new_stdin = new_stdin, loader = fake_loader, shared_loader_obj = shared_loader, ) te._get_connection = MagicMock(return_value=mock_connection) te._get_action_handler = MagicMock(return_value=mock_action) mock_action.run.return_value = dict(ansible_facts=dict()) res = te._execute() mock_task.changed_when = "1 == 1" res = te._execute() mock_task.changed_when = None mock_task.failed_when = "1 == 1" res = te._execute() mock_task.failed_when = None mock_task.evaluate_conditional.return_value = False res = te._execute() mock_task.evaluate_conditional.return_value = True mock_task.args = dict(_raw_params='foo.yml', a='foo', b='bar') mock_task.action = 'include' res = te._execute() def test_task_executor_poll_async_result(self): fake_loader = DictDataLoader({}) mock_host = MagicMock() mock_task = MagicMock() mock_task.async = 3 mock_task.poll = 1 mock_conn_info = MagicMock() mock_connection = MagicMock() mock_action = MagicMock() shared_loader = None new_stdin = None job_vars = dict(omit="XXXXXXXXXXXXXXXXXXX") te = TaskExecutor( host = mock_host, task = mock_task, job_vars = job_vars, connection_info = mock_conn_info, new_stdin = new_stdin, loader = fake_loader, shared_loader_obj = shared_loader, ) te._connection = MagicMock() def _get(*args, **kwargs): mock_action = MagicMock() mock_action.run.return_value = dict(stdout='') return mock_action # testing with some bad values in the result passed to poll async, # and with a bad value returned from the mock action with patch.object(action_loader, 'get', _get): mock_templar = MagicMock() res = te._poll_async_result(result=dict(), templar=mock_templar) self.assertIn('failed', res) res = te._poll_async_result(result=dict(ansible_job_id=1), templar=mock_templar) self.assertIn('failed', res) def _get(*args, **kwargs): mock_action = MagicMock() mock_action.run.return_value = dict(finished=1) return mock_action # now testing with good values with patch.object(action_loader, 'get', _get): mock_templar = MagicMock() res = te._poll_async_result(result=dict(ansible_job_id=1), templar=mock_templar) self.assertEqual(res, dict(finished=1))