From d4a31e8d26e22f160a6a433fd6f21da8c0435b70 Mon Sep 17 00:00:00 2001 From: James Cammarata Date: Sun, 24 May 2015 07:47:06 -0500 Subject: [PATCH] Adding unit tests for TaskExecutor (v2) --- test/units/executor/test_task_executor.py | 324 ++++++++++++++++++++++ 1 file changed, 324 insertions(+) create mode 100644 test/units/executor/test_task_executor.py diff --git a/test/units/executor/test_task_executor.py b/test/units/executor/test_task_executor.py new file mode 100644 index 00000000000..64ce1d5faa2 --- /dev/null +++ b/test/units/executor/test_task_executor.py @@ -0,0 +1,324 @@ +# (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() + 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)) +