ansible/test/units/executor/test_task_executor.py
2015-05-24 07:47:06 -05:00

324 lines
9.6 KiB
Python

# (c) 2012-2014, Michael DeHaan <michael.dehaan@gmail.com>
#
# 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 <http://www.gnu.org/licenses/>.
# 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))