0871d955fe
Previous changes addressed a corner case, which unfortunately introduced
another bug. This patch adds a new flag to the host state (did_rescue) which
is set to true when the rescue portion of a block completes. This flag is
then checked in _check_failed_state() when the fail_state != FAILED_NONE.
This lead to the discovery of another bug - current strategies are not advancing
hosts to ITERATING_COMPLETE after doing a peek at the next task, leaving the
host state in the run_state of the final task. To address this, before gathering
the list of failed hosts in StrategyBase.run(), a final pass through the iterator
for all hosts is done to ensure each host is in its final state. This way, no
strategy derived from StrategyBase has to worry about it and it's handled.
Fixes #17983
(cherry picked from commit ca5b361ad8)
519 lines
20 KiB
Python
519 lines
20 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
|
|
|
|
import uuid
|
|
|
|
from ansible.compat.tests import unittest
|
|
from ansible.compat.tests.mock import patch, MagicMock
|
|
|
|
from ansible.errors import AnsibleError, AnsibleParserError
|
|
from ansible.plugins.strategy import StrategyBase
|
|
from ansible.executor.process.worker import WorkerProcess
|
|
from ansible.executor.task_queue_manager import TaskQueueManager
|
|
from ansible.executor.task_result import TaskResult
|
|
from ansible.playbook.block import Block
|
|
from ansible.playbook.handler import Handler
|
|
from ansible.inventory.host import Host
|
|
|
|
from six.moves import queue as Queue
|
|
from units.mock.loader import DictDataLoader
|
|
|
|
class TestStrategyBase(unittest.TestCase):
|
|
|
|
def setUp(self):
|
|
pass
|
|
|
|
def tearDown(self):
|
|
pass
|
|
|
|
def test_strategy_base_init(self):
|
|
queue_items = []
|
|
def _queue_empty(*args, **kwargs):
|
|
return len(queue_items) == 0
|
|
def _queue_get(*args, **kwargs):
|
|
if len(queue_items) == 0:
|
|
raise Queue.Empty
|
|
else:
|
|
return queue_items.pop()
|
|
def _queue_put(item, *args, **kwargs):
|
|
queue_items.append(item)
|
|
|
|
mock_queue = MagicMock()
|
|
mock_queue.empty.side_effect = _queue_empty
|
|
mock_queue.get.side_effect = _queue_get
|
|
mock_queue.put.side_effect = _queue_put
|
|
|
|
mock_tqm = MagicMock(TaskQueueManager)
|
|
mock_tqm._final_q = mock_queue
|
|
mock_tqm._options = MagicMock()
|
|
mock_tqm._notified_handlers = {}
|
|
mock_tqm._listening_handlers = {}
|
|
strategy_base = StrategyBase(tqm=mock_tqm)
|
|
strategy_base.cleanup()
|
|
|
|
def test_strategy_base_run(self):
|
|
queue_items = []
|
|
def _queue_empty(*args, **kwargs):
|
|
return len(queue_items) == 0
|
|
def _queue_get(*args, **kwargs):
|
|
if len(queue_items) == 0:
|
|
raise Queue.Empty
|
|
else:
|
|
return queue_items.pop()
|
|
def _queue_put(item, *args, **kwargs):
|
|
queue_items.append(item)
|
|
|
|
mock_queue = MagicMock()
|
|
mock_queue.empty.side_effect = _queue_empty
|
|
mock_queue.get.side_effect = _queue_get
|
|
mock_queue.put.side_effect = _queue_put
|
|
|
|
mock_tqm = MagicMock(TaskQueueManager)
|
|
mock_tqm._final_q = mock_queue
|
|
mock_tqm._stats = MagicMock()
|
|
mock_tqm._notified_handlers = {}
|
|
mock_tqm._listening_handlers = {}
|
|
mock_tqm.send_callback.return_value = None
|
|
|
|
for attr in ('RUN_OK', 'RUN_ERROR', 'RUN_FAILED_HOSTS', 'RUN_UNREACHABLE_HOSTS'):
|
|
setattr(mock_tqm, attr, getattr(TaskQueueManager, attr))
|
|
|
|
mock_iterator = MagicMock()
|
|
mock_iterator._play = MagicMock()
|
|
mock_iterator._play.handlers = []
|
|
|
|
mock_play_context = MagicMock()
|
|
|
|
mock_tqm._failed_hosts = dict()
|
|
mock_tqm._unreachable_hosts = dict()
|
|
mock_tqm._options = MagicMock()
|
|
mock_tqm._notified_handlers = {}
|
|
mock_tqm._listening_handlers = {}
|
|
strategy_base = StrategyBase(tqm=mock_tqm)
|
|
|
|
mock_host = MagicMock()
|
|
mock_host.name = 'host1'
|
|
|
|
self.assertEqual(strategy_base.run(iterator=mock_iterator, play_context=mock_play_context), mock_tqm.RUN_OK)
|
|
self.assertEqual(strategy_base.run(iterator=mock_iterator, play_context=mock_play_context, result=TaskQueueManager.RUN_ERROR), mock_tqm.RUN_ERROR)
|
|
mock_tqm._failed_hosts = dict(host1=True)
|
|
mock_iterator.get_failed_hosts.return_value = [mock_host]
|
|
self.assertEqual(strategy_base.run(iterator=mock_iterator, play_context=mock_play_context, result=False), mock_tqm.RUN_FAILED_HOSTS)
|
|
mock_tqm._unreachable_hosts = dict(host1=True)
|
|
mock_iterator.get_failed_hosts.return_value = []
|
|
self.assertEqual(strategy_base.run(iterator=mock_iterator, play_context=mock_play_context, result=False), mock_tqm.RUN_UNREACHABLE_HOSTS)
|
|
strategy_base.cleanup()
|
|
|
|
def test_strategy_base_get_hosts(self):
|
|
queue_items = []
|
|
def _queue_empty(*args, **kwargs):
|
|
return len(queue_items) == 0
|
|
def _queue_get(*args, **kwargs):
|
|
if len(queue_items) == 0:
|
|
raise Queue.Empty
|
|
else:
|
|
return queue_items.pop()
|
|
def _queue_put(item, *args, **kwargs):
|
|
queue_items.append(item)
|
|
|
|
mock_queue = MagicMock()
|
|
mock_queue.empty.side_effect = _queue_empty
|
|
mock_queue.get.side_effect = _queue_get
|
|
mock_queue.put.side_effect = _queue_put
|
|
|
|
mock_hosts = []
|
|
for i in range(0, 5):
|
|
mock_host = MagicMock()
|
|
mock_host.name = "host%02d" % (i+1)
|
|
mock_host.has_hostkey = True
|
|
mock_hosts.append(mock_host)
|
|
|
|
mock_inventory = MagicMock()
|
|
mock_inventory.get_hosts.return_value = mock_hosts
|
|
|
|
mock_tqm = MagicMock()
|
|
mock_tqm._final_q = mock_queue
|
|
mock_tqm._notified_handlers = {}
|
|
mock_tqm._listening_handlers = {}
|
|
mock_tqm.get_inventory.return_value = mock_inventory
|
|
|
|
mock_play = MagicMock()
|
|
mock_play.hosts = ["host%02d" % (i+1) for i in range(0, 5)]
|
|
|
|
strategy_base = StrategyBase(tqm=mock_tqm)
|
|
|
|
mock_tqm._failed_hosts = []
|
|
mock_tqm._unreachable_hosts = []
|
|
self.assertEqual(strategy_base.get_hosts_remaining(play=mock_play), mock_hosts)
|
|
|
|
mock_tqm._failed_hosts = ["host01"]
|
|
self.assertEqual(strategy_base.get_hosts_remaining(play=mock_play), mock_hosts[1:])
|
|
self.assertEqual(strategy_base.get_failed_hosts(play=mock_play), [mock_hosts[0]])
|
|
|
|
mock_tqm._unreachable_hosts = ["host02"]
|
|
self.assertEqual(strategy_base.get_hosts_remaining(play=mock_play), mock_hosts[2:])
|
|
strategy_base.cleanup()
|
|
|
|
@patch.object(WorkerProcess, 'run')
|
|
def test_strategy_base_queue_task(self, mock_worker):
|
|
def fake_run(self):
|
|
return
|
|
|
|
mock_worker.run.side_effect = fake_run
|
|
|
|
fake_loader = DictDataLoader()
|
|
mock_var_manager = MagicMock()
|
|
mock_host = MagicMock()
|
|
mock_host.has_hostkey = True
|
|
mock_inventory = MagicMock()
|
|
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,
|
|
)
|
|
tqm._initialize_processes(3)
|
|
tqm.hostvars = dict()
|
|
|
|
try:
|
|
strategy_base = StrategyBase(tqm=tqm)
|
|
strategy_base._queue_task(host=mock_host, task=MagicMock(), task_vars=dict(), play_context=MagicMock())
|
|
self.assertEqual(strategy_base._cur_worker, 1)
|
|
self.assertEqual(strategy_base._pending_results, 1)
|
|
strategy_base._queue_task(host=mock_host, task=MagicMock(), task_vars=dict(), play_context=MagicMock())
|
|
self.assertEqual(strategy_base._cur_worker, 2)
|
|
self.assertEqual(strategy_base._pending_results, 2)
|
|
strategy_base._queue_task(host=mock_host, task=MagicMock(), task_vars=dict(), play_context=MagicMock())
|
|
self.assertEqual(strategy_base._cur_worker, 0)
|
|
self.assertEqual(strategy_base._pending_results, 3)
|
|
finally:
|
|
tqm.cleanup()
|
|
|
|
|
|
def test_strategy_base_process_pending_results(self):
|
|
mock_tqm = MagicMock()
|
|
mock_tqm._terminated = False
|
|
mock_tqm._failed_hosts = dict()
|
|
mock_tqm._unreachable_hosts = dict()
|
|
mock_tqm.send_callback.return_value = None
|
|
mock_tqm._notified_handlers = {}
|
|
mock_tqm._listening_handlers = {}
|
|
|
|
queue_items = []
|
|
def _queue_empty(*args, **kwargs):
|
|
return len(queue_items) == 0
|
|
def _queue_get(*args, **kwargs):
|
|
if len(queue_items) == 0:
|
|
raise Queue.Empty
|
|
else:
|
|
return queue_items.pop()
|
|
def _queue_put(item, *args, **kwargs):
|
|
queue_items.append(item)
|
|
|
|
mock_queue = MagicMock()
|
|
mock_queue.empty.side_effect = _queue_empty
|
|
mock_queue.get.side_effect = _queue_get
|
|
mock_queue.put.side_effect = _queue_put
|
|
mock_tqm._final_q = mock_queue
|
|
|
|
mock_tqm._stats = MagicMock()
|
|
mock_tqm._stats.increment.return_value = None
|
|
|
|
mock_play = MagicMock()
|
|
|
|
mock_host = MagicMock()
|
|
mock_host.name = 'test01'
|
|
mock_host.vars = dict()
|
|
mock_host.has_hostkey = True
|
|
|
|
mock_task = MagicMock()
|
|
mock_task._role = None
|
|
mock_task.ignore_errors = False
|
|
mock_task._uuid = uuid.uuid4()
|
|
mock_task.loop = None
|
|
|
|
mock_handler_task = MagicMock(Handler)
|
|
mock_handler_task.name = 'test handler'
|
|
mock_handler_task.action = 'foo'
|
|
mock_handler_task.get_name.return_value = "test handler"
|
|
mock_handler_task.has_triggered.return_value = False
|
|
|
|
mock_iterator = MagicMock()
|
|
mock_iterator._play = mock_play
|
|
mock_iterator.mark_host_failed.return_value = None
|
|
mock_iterator.get_next_task_for_host.return_value = (None, None)
|
|
mock_iterator.get_original_task.return_value = mock_task
|
|
|
|
mock_handler_block = MagicMock()
|
|
mock_handler_block.block = [mock_handler_task]
|
|
mock_handler_block.rescue = []
|
|
mock_handler_block.always = []
|
|
mock_play.handlers = [mock_handler_block]
|
|
|
|
mock_tqm._notified_handlers = {mock_handler_task: []}
|
|
mock_tqm._listening_handlers = {}
|
|
|
|
mock_group = MagicMock()
|
|
mock_group.add_host.return_value = None
|
|
|
|
def _get_host(host_name):
|
|
if host_name == 'test01':
|
|
return mock_host
|
|
return None
|
|
def _get_group(group_name):
|
|
if group_name in ('all', 'foo'):
|
|
return mock_group
|
|
return None
|
|
|
|
mock_inventory = MagicMock()
|
|
mock_inventory._hosts_cache = dict()
|
|
mock_inventory.get_host.side_effect = _get_host
|
|
mock_inventory.get_group.side_effect = _get_group
|
|
mock_inventory.clear_pattern_cache.return_value = None
|
|
mock_inventory.clear_group_dict_cache.return_value = None
|
|
mock_inventory.get_host_vars.return_value = {}
|
|
|
|
mock_var_mgr = MagicMock()
|
|
mock_var_mgr.set_host_variable.return_value = None
|
|
mock_var_mgr.set_host_facts.return_value = None
|
|
|
|
strategy_base = StrategyBase(tqm=mock_tqm)
|
|
strategy_base._inventory = mock_inventory
|
|
strategy_base._variable_manager = mock_var_mgr
|
|
strategy_base._blocked_hosts = dict()
|
|
|
|
def _has_dead_workers():
|
|
return False
|
|
|
|
strategy_base._tqm.has_dead_workers = _has_dead_workers
|
|
results = strategy_base._wait_on_pending_results(iterator=mock_iterator)
|
|
self.assertEqual(len(results), 0)
|
|
|
|
task_result = TaskResult(host=mock_host.name, task=mock_task._uuid, return_data=dict(changed=True))
|
|
queue_items.append(task_result)
|
|
strategy_base._blocked_hosts['test01'] = True
|
|
strategy_base._pending_results = 1
|
|
results = strategy_base._wait_on_pending_results(iterator=mock_iterator)
|
|
self.assertEqual(len(results), 1)
|
|
self.assertEqual(results[0], task_result)
|
|
self.assertEqual(strategy_base._pending_results, 0)
|
|
self.assertNotIn('test01', strategy_base._blocked_hosts)
|
|
|
|
task_result = TaskResult(host=mock_host.name, task=mock_task._uuid, return_data='{"failed":true}')
|
|
queue_items.append(task_result)
|
|
strategy_base._blocked_hosts['test01'] = True
|
|
strategy_base._pending_results = 1
|
|
mock_iterator.is_failed.return_value = True
|
|
results = strategy_base._wait_on_pending_results(iterator=mock_iterator)
|
|
self.assertEqual(len(results), 1)
|
|
self.assertEqual(results[0], task_result)
|
|
self.assertEqual(strategy_base._pending_results, 0)
|
|
self.assertNotIn('test01', strategy_base._blocked_hosts)
|
|
#self.assertIn('test01', mock_tqm._failed_hosts)
|
|
#del mock_tqm._failed_hosts['test01']
|
|
mock_iterator.is_failed.return_value = False
|
|
|
|
task_result = TaskResult(host=mock_host.name, task=mock_task._uuid, return_data='{"unreachable": true}')
|
|
queue_items.append(task_result)
|
|
strategy_base._blocked_hosts['test01'] = True
|
|
strategy_base._pending_results = 1
|
|
results = strategy_base._wait_on_pending_results(iterator=mock_iterator)
|
|
self.assertEqual(len(results), 1)
|
|
self.assertEqual(results[0], task_result)
|
|
self.assertEqual(strategy_base._pending_results, 0)
|
|
self.assertNotIn('test01', strategy_base._blocked_hosts)
|
|
self.assertIn('test01', mock_tqm._unreachable_hosts)
|
|
del mock_tqm._unreachable_hosts['test01']
|
|
|
|
task_result = TaskResult(host=mock_host.name, task=mock_task._uuid, return_data='{"skipped": true}')
|
|
queue_items.append(task_result)
|
|
strategy_base._blocked_hosts['test01'] = True
|
|
strategy_base._pending_results = 1
|
|
results = strategy_base._wait_on_pending_results(iterator=mock_iterator)
|
|
self.assertEqual(len(results), 1)
|
|
self.assertEqual(results[0], task_result)
|
|
self.assertEqual(strategy_base._pending_results, 0)
|
|
self.assertNotIn('test01', strategy_base._blocked_hosts)
|
|
|
|
queue_items.append(TaskResult(host=mock_host.name, task=mock_task._uuid, return_data=dict(add_host=dict(host_name='newhost01', new_groups=['foo']))))
|
|
strategy_base._blocked_hosts['test01'] = True
|
|
strategy_base._pending_results = 1
|
|
results = strategy_base._wait_on_pending_results(iterator=mock_iterator)
|
|
self.assertEqual(len(results), 1)
|
|
self.assertEqual(strategy_base._pending_results, 0)
|
|
self.assertNotIn('test01', strategy_base._blocked_hosts)
|
|
|
|
queue_items.append(TaskResult(host=mock_host.name, task=mock_task._uuid, return_data=dict(add_group=dict(group_name='foo'))))
|
|
strategy_base._blocked_hosts['test01'] = True
|
|
strategy_base._pending_results = 1
|
|
results = strategy_base._wait_on_pending_results(iterator=mock_iterator)
|
|
self.assertEqual(len(results), 1)
|
|
self.assertEqual(strategy_base._pending_results, 0)
|
|
self.assertNotIn('test01', strategy_base._blocked_hosts)
|
|
|
|
queue_items.append(TaskResult(host=mock_host.name, task=mock_task._uuid, return_data=dict(changed=True, _ansible_notify=['test handler'])))
|
|
strategy_base._blocked_hosts['test01'] = True
|
|
strategy_base._pending_results = 1
|
|
results = strategy_base._wait_on_pending_results(iterator=mock_iterator)
|
|
self.assertEqual(len(results), 1)
|
|
self.assertEqual(strategy_base._pending_results, 0)
|
|
self.assertNotIn('test01', strategy_base._blocked_hosts)
|
|
self.assertIn(mock_handler_task, strategy_base._notified_handlers)
|
|
self.assertIn(mock_host, strategy_base._notified_handlers[mock_handler_task])
|
|
|
|
#queue_items.append(('set_host_var', mock_host, mock_task, None, 'foo', 'bar'))
|
|
#results = strategy_base._process_pending_results(iterator=mock_iterator)
|
|
#self.assertEqual(len(results), 0)
|
|
#self.assertEqual(strategy_base._pending_results, 1)
|
|
|
|
#queue_items.append(('set_host_facts', mock_host, mock_task, None, 'foo', dict()))
|
|
#results = strategy_base._process_pending_results(iterator=mock_iterator)
|
|
#self.assertEqual(len(results), 0)
|
|
#self.assertEqual(strategy_base._pending_results, 1)
|
|
|
|
#queue_items.append(('bad'))
|
|
#self.assertRaises(AnsibleError, strategy_base._process_pending_results, iterator=mock_iterator)
|
|
strategy_base.cleanup()
|
|
|
|
def test_strategy_base_load_included_file(self):
|
|
fake_loader = DictDataLoader({
|
|
"test.yml": """
|
|
- debug: msg='foo'
|
|
""",
|
|
"bad.yml": """
|
|
""",
|
|
})
|
|
|
|
queue_items = []
|
|
def _queue_empty(*args, **kwargs):
|
|
return len(queue_items) == 0
|
|
def _queue_get(*args, **kwargs):
|
|
if len(queue_items) == 0:
|
|
raise Queue.Empty
|
|
else:
|
|
return queue_items.pop()
|
|
def _queue_put(item, *args, **kwargs):
|
|
queue_items.append(item)
|
|
|
|
mock_queue = MagicMock()
|
|
mock_queue.empty.side_effect = _queue_empty
|
|
mock_queue.get.side_effect = _queue_get
|
|
mock_queue.put.side_effect = _queue_put
|
|
|
|
mock_tqm = MagicMock()
|
|
mock_tqm._final_q = mock_queue
|
|
mock_tqm._notified_handlers = {}
|
|
mock_tqm._listening_handlers = {}
|
|
|
|
strategy_base = StrategyBase(tqm=mock_tqm)
|
|
strategy_base._loader = fake_loader
|
|
strategy_base.cleanup()
|
|
|
|
mock_play = MagicMock()
|
|
|
|
mock_block = MagicMock()
|
|
mock_block._play = mock_play
|
|
mock_block.vars = dict()
|
|
|
|
mock_task = MagicMock()
|
|
mock_task._block = mock_block
|
|
mock_task._role = None
|
|
|
|
mock_iterator = MagicMock()
|
|
mock_iterator.mark_host_failed.return_value = None
|
|
|
|
mock_inc_file = MagicMock()
|
|
mock_inc_file._task = mock_task
|
|
|
|
mock_inc_file._filename = "test.yml"
|
|
res = strategy_base._load_included_file(included_file=mock_inc_file, iterator=mock_iterator)
|
|
|
|
mock_inc_file._filename = "bad.yml"
|
|
res = strategy_base._load_included_file(included_file=mock_inc_file, iterator=mock_iterator)
|
|
self.assertEqual(res, [])
|
|
|
|
@patch.object(WorkerProcess, 'run')
|
|
def test_strategy_base_run_handlers(self, mock_worker):
|
|
def fake_run(*args):
|
|
return
|
|
mock_worker.side_effect = fake_run
|
|
mock_play_context = MagicMock()
|
|
|
|
mock_handler_task = MagicMock(Handler)
|
|
mock_handler_task.action = 'foo'
|
|
mock_handler_task.get_name.return_value = "test handler"
|
|
mock_handler_task.has_triggered.return_value = False
|
|
mock_handler_task.listen = None
|
|
mock_handler_task._role = None
|
|
|
|
mock_handler = MagicMock()
|
|
mock_handler.block = [mock_handler_task]
|
|
mock_handler.flag_for_host.return_value = False
|
|
|
|
mock_play = MagicMock()
|
|
mock_play.handlers = [mock_handler]
|
|
|
|
mock_host = MagicMock(Host)
|
|
mock_host.name = "test01"
|
|
mock_host.has_hostkey = True
|
|
|
|
mock_inventory = MagicMock()
|
|
mock_inventory.get_hosts.return_value = [mock_host]
|
|
|
|
mock_var_mgr = MagicMock()
|
|
mock_var_mgr.get_vars.return_value = dict()
|
|
|
|
mock_iterator = MagicMock()
|
|
mock_iterator._play = mock_play
|
|
mock_iterator.get_original_task.return_value = mock_handler_task
|
|
|
|
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,
|
|
)
|
|
tqm._initialize_processes(3)
|
|
tqm._initialize_notified_handlers(mock_play)
|
|
tqm.hostvars = dict()
|
|
|
|
try:
|
|
strategy_base = StrategyBase(tqm=tqm)
|
|
|
|
strategy_base._inventory = mock_inventory
|
|
strategy_base._notified_handlers = {mock_handler_task: [mock_host]}
|
|
|
|
task_result = TaskResult(Host('host01'), Handler(), dict(changed=False))
|
|
tqm._final_q.put(task_result)
|
|
|
|
result = strategy_base.run_handlers(iterator=mock_iterator, play_context=mock_play_context)
|
|
finally:
|
|
strategy_base.cleanup()
|
|
tqm.cleanup()
|