wbrawner/ansible
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Properly load module_prefix for collections in task_executor.py (#60420)

Browse source 
* Properly load module_prefix for collections

Now that we are using collections for tasks, we need to properly split
the name and load the prefix properly.

Signed-off-by: Paul Belanger <pabelanger@redhat.com>

* Add unit tests for task_executor

This commit adds missing unit tests for action handler in test_task_executor.

Signed-off-by: Daniel Mellado <dmellado@redhat.com>
This commit is contained in:
Paul Belanger 2019-08-29 12:04:26 -04:00 committed by Matt Davis
parent 6a8dcb8fcb
commit f23bb6f7bd
2 changed files with 105 additions and 1 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
lib/ansible/executor/task_executor.py
View file

@ -1019,7 +1019,7 @@ class TaskExecutor:
Returns the correct action plugin to handle the requestion task action
'''
module_prefix = self._task.action.split('_')[0]
module_prefix = self._task.action.split('.')[-1].split('_')[0]
collections = self._task.collections

104
test/units/executor/test_task_executor.py
View file

@ -19,6 +19,8 @@
from __future__ import (absolute_import, division, print_function)
__metaclass__ = type
import mock
from units.compat import unittest
from units.compat.mock import patch, MagicMock
from ansible.errors import AnsibleError
@ -355,6 +357,108 @@ class TestTaskExecutor(unittest.TestCase):
self.assertEqual(new_items, items)
self.assertEqual(mock_task.args, {'name': '{{item.name}}', 'state': '{{item.state}}'})
def test_task_executor_get_action_handler(self):
te = TaskExecutor(
host=MagicMock(),
task=MagicMock(),
job_vars={},
play_context=MagicMock(),
new_stdin=None,
loader=DictDataLoader({}),
shared_loader_obj=MagicMock(),
final_q=MagicMock(),
)
action_loader = te._shared_loader_obj.action_loader
action_loader.has_plugin.return_value = True
action_loader.get.return_value = mock.sentinel.handler
mock_connection = MagicMock()
mock_templar = MagicMock()
action = 'namespace.prefix_sufix'
te._task.action = action
handler = te._get_action_handler(mock_connection, mock_templar)
self.assertIs(mock.sentinel.handler, handler)
action_loader.has_plugin.assert_called_once_with(
action, collection_list=te._task.collections)
action_loader.get.assert_called_once_with(
te._task.action, task=te._task, connection=mock_connection,
play_context=te._play_context, loader=te._loader,
templar=mock_templar, shared_loader_obj=te._shared_loader_obj,
collection_list=te._task.collections)
def test_task_executor_get_handler_prefix(self):
te = TaskExecutor(
host=MagicMock(),
task=MagicMock(),
job_vars={},
play_context=MagicMock(),
new_stdin=None,
loader=DictDataLoader({}),
shared_loader_obj=MagicMock(),
final_q=MagicMock(),
)
action_loader = te._shared_loader_obj.action_loader
action_loader.has_plugin.return_value = False
action_loader.get.return_value = mock.sentinel.handler
action_loader.__contains__.return_value = True
mock_connection = MagicMock()
mock_templar = MagicMock()
action = 'namespace.netconf_sufix'
te._task.action = action
handler = te._get_action_handler(mock_connection, mock_templar)
self.assertIs(mock.sentinel.handler, handler)
action_loader.has_plugin.assert_called_once_with(
action, collection_list=te._task.collections)
action_loader.get.assert_called_once_with(
'netconf', task=te._task, connection=mock_connection,
play_context=te._play_context, loader=te._loader,
templar=mock_templar, shared_loader_obj=te._shared_loader_obj,
collection_list=te._task.collections)
def test_task_executor_get_handler_normal(self):
te = TaskExecutor(
host=MagicMock(),
task=MagicMock(),
job_vars={},
play_context=MagicMock(),
new_stdin=None,
loader=DictDataLoader({}),
shared_loader_obj=MagicMock(),
final_q=MagicMock(),
)
action_loader = te._shared_loader_obj.action_loader
action_loader.has_plugin.return_value = False
action_loader.get.return_value = mock.sentinel.handler
action_loader.__contains__.return_value = False
mock_connection = MagicMock()
mock_templar = MagicMock()
action = 'namespace.prefix_sufix'
te._task.action = action
handler = te._get_action_handler(mock_connection, mock_templar)
self.assertIs(mock.sentinel.handler, handler)
action_loader.has_plugin.assert_called_once_with(
action, collection_list=te._task.collections)
action_loader.get.assert_called_once_with(
'normal', task=te._task, connection=mock_connection,
play_context=te._play_context, loader=te._loader,
templar=mock_templar, shared_loader_obj=te._shared_loader_obj,
collection_list=None)
def test_task_executor_execute(self):
fake_loader = DictDataLoader({})