1# (c) 2016, Steve Kuznetsov <skuznets@redhat.com>
2#
3# This file is part of Ansible
4#
5# Ansible is free software: you can redistribute it and/or modify
6# it under the terms of the GNU General Public License as published by
7# the Free Software Foundation, either version 3 of the License, or
8# (at your option) any later version.
9#
10# Ansible is distributed in the hope that it will be useful,
11# but WITHOUT ANY WARRANTY; without even the implied warranty of
12# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
13# GNU General Public License for more details.
14#
15# You should have received a copy of the GNU General Public License
16# along with Ansible.  If not, see <http://www.gnu.org/licenses/>.
17
18# Make coding more python3-ish
19from __future__ import (absolute_import, division, print_function)
20
21from units.compat import unittest
22from units.compat.mock import MagicMock
23
24from ansible.executor.task_queue_manager import TaskQueueManager
25from ansible.playbook import Playbook
26from ansible.plugins.callback import CallbackBase
27from ansible.utils import context_objects as co
28
29__metaclass__ = type
30
31
32class TestTaskQueueManagerCallbacks(unittest.TestCase):
33    def setUp(self):
34        inventory = MagicMock()
35        variable_manager = MagicMock()
36        loader = MagicMock()
37        passwords = []
38
39        # Reset the stored command line args
40        co.GlobalCLIArgs._Singleton__instance = None
41        self._tqm = TaskQueueManager(inventory, variable_manager, loader, passwords)
42        self._playbook = Playbook(loader)
43
44        # we use a MagicMock to register the result of the call we
45        # expect to `v2_playbook_on_call`. We don't mock out the
46        # method since we're testing code that uses `inspect` to
47        # look at that method's argspec and we want to ensure this
48        # test is easy to reason about.
49        self._register = MagicMock()
50
51    def tearDown(self):
52        # Reset the stored command line args
53        co.GlobalCLIArgs._Singleton__instance = None
54
55    def test_task_queue_manager_callbacks_v2_playbook_on_start(self):
56        """
57        Assert that no exceptions are raised when sending a Playbook
58        start callback to a current callback module plugin.
59        """
60        register = self._register
61
62        class CallbackModule(CallbackBase):
63            """
64            This is a callback module with the current
65            method signature for `v2_playbook_on_start`.
66            """
67            CALLBACK_VERSION = 2.0
68            CALLBACK_TYPE = 'notification'
69            CALLBACK_NAME = 'current_module'
70
71            def v2_playbook_on_start(self, playbook):
72                register(self, playbook)
73
74        callback_module = CallbackModule()
75        self._tqm._callback_plugins.append(callback_module)
76        self._tqm.send_callback('v2_playbook_on_start', self._playbook)
77        register.assert_called_once_with(callback_module, self._playbook)
78
79    def test_task_queue_manager_callbacks_v2_playbook_on_start_wrapped(self):
80        """
81        Assert that no exceptions are raised when sending a Playbook
82        start callback to a wrapped current callback module plugin.
83        """
84        register = self._register
85
86        def wrap_callback(func):
87            """
88            This wrapper changes the exposed argument
89            names for a method from the original names
90            to (*args, **kwargs). This is used in order
91            to validate that wrappers which change par-
92            ameter names do not break the TQM callback
93            system.
94
95            :param func: function to decorate
96            :return: decorated function
97            """
98
99            def wrapper(*args, **kwargs):
100                return func(*args, **kwargs)
101
102            return wrapper
103
104        class WrappedCallbackModule(CallbackBase):
105            """
106            This is a callback module with the current
107            method signature for `v2_playbook_on_start`
108            wrapped in order to change the signature.
109            """
110            CALLBACK_VERSION = 2.0
111            CALLBACK_TYPE = 'notification'
112            CALLBACK_NAME = 'current_module'
113
114            @wrap_callback
115            def v2_playbook_on_start(self, playbook):
116                register(self, playbook)
117
118        callback_module = WrappedCallbackModule()
119        self._tqm._callback_plugins.append(callback_module)
120        self._tqm.send_callback('v2_playbook_on_start', self._playbook)
121        register.assert_called_once_with(callback_module, self._playbook)
122