Some of the repetitive violations were suppressed.
18.98KiB; Python | 2019-11-06 14:54:25+01 | SLOC 347
1
# (c) 2016, Adrian Likins <alikins@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
19
from __future__ import (absolute_import, division, print_function)
20
__metaclass__ = type
21
22
import os
23
24
from units.compat import unittest
25
from units.compat.mock import MagicMock
26
from units.mock.loader import DictDataLoader
27
28
from ansible import errors
29
from ansible.playbook.block import Block
30
from ansible.playbook.handler import Handler
31
from ansible.playbook.task import Task
32
from ansible.playbook.task_include import TaskInclude
33
from ansible.playbook.role.include import RoleInclude
34
35
from ansible.playbook import helpers
36
37
38
class MixinForMocks(object):
39
    def _setup(self):
40
        # This is not a very good mixin, lots of side effects
41 1
        self.fake_loader = DictDataLoader({'include_test.yml': "",
42
                                           'other_include_test.yml': ""})
43 1
        self.mock_tqm = MagicMock(name='MockTaskQueueManager')
44
45 1
        self.mock_play = MagicMock(name='MockPlay')
46 1
        self.mock_play._attributes = []
47
        self.mock_play.collections = None
48
49 1
        self.mock_iterator = MagicMock(name='MockIterator')
50 1
        self.mock_iterator._play = self.mock_play
51
52 1
        self.mock_inventory = MagicMock(name='MockInventory')
53 1
        self.mock_inventory._hosts_cache = dict()
54
55
        def _get_host(host_name):
56
            return None
57
58
        self.mock_inventory.get_host.side_effect = _get_host
59 1
        # TODO: can we use a real VariableManager?
60 1
        self.mock_variable_manager = MagicMock(name='MockVariableManager')
61
        self.mock_variable_manager.get_vars.return_value = dict()
62
63 1
        self.mock_block = MagicMock(name='MockBlock')
64
65
        # On macOS /etc is actually /private/etc, tests fail when performing literal /etc checks
66 1
        self.fake_role_loader = DictDataLoader({os.path.join(os.path.realpath("/etc"), "ansible/roles/bogus_role/tasks/main.yml"): """
67
                                                - shell: echo 'hello world'
68
                                                """})
69
70 1
        self._test_data_path = os.path.dirname(__file__)
71 1
        self.fake_include_loader = DictDataLoader({"/dev/null/includes/test_include.yml": """
72
                                                   - include: other_test_include.yml
73
                                                   - shell: echo 'hello world'
74
                                                   """,
75
                                                   "/dev/null/includes/static_test_include.yml": """
76
                                                   - include: other_test_include.yml
77
                                                   - shell: echo 'hello static world'
78
                                                   """,
79
                                                   "/dev/null/includes/other_test_include.yml": """
80
                                                   - debug:
81
                                                       msg: other_test_include_debug
82
                                                   """})
83
84
85
class TestLoadListOfTasks(unittest.TestCase, MixinForMocks):
86
    def setUp(self):
87
        self._setup()
88
89
    def _assert_is_task_list(self, results):
90
        for result in results:
91
            self.assertIsInstance(result, Task)
92
93
    def _assert_is_task_list_or_blocks(self, results):
94
        self.assertIsInstance(results, list)
95
        for result in results:
96
            self.assertIsInstance(result, (Task, Block))
97
98
    def test_ds_not_list(self):
99
        ds = {}
100
        self.assertRaises(AssertionError, helpers.load_list_of_tasks,
101
                          ds, self.mock_play, block=None, role=None, task_include=None, use_handlers=False, variable_manager=None, loader=None)
102
103
    def test_ds_not_dict(self):
104
        ds = [[]]
105
        self.assertRaises(AssertionError, helpers.load_list_of_tasks,
106
                          ds, self.mock_play, block=None, role=None, task_include=None, use_handlers=False, variable_manager=None, loader=None)
107
108
    def test_empty_task(self):
109
        ds = [{}]
110
        self.assertRaisesRegexp(errors.AnsibleParserError,
111
                                "no module/action detected in task",
112
                                helpers.load_list_of_tasks,
113
                                ds, play=self.mock_play,
114
                                variable_manager=self.mock_variable_manager, loader=self.fake_loader)
115
116
    def test_empty_task_use_handlers(self):
117
        ds = [{}]
118
        self.assertRaisesRegexp(errors.AnsibleParserError,
119
                                "no module/action detected in task.",
120
                                helpers.load_list_of_tasks,
121
                                ds,
122
                                use_handlers=True,
123
                                play=self.mock_play,
124
                                variable_manager=self.mock_variable_manager,
125
                                loader=self.fake_loader)
126
127
    def test_one_bogus_block(self):
128
        ds = [{'block': None}]
129
        self.assertRaisesRegexp(errors.AnsibleParserError,
130
                                "A malformed block was encountered",
131
                                helpers.load_list_of_tasks,
132
                                ds, play=self.mock_play,
133
                                variable_manager=self.mock_variable_manager, loader=self.fake_loader)
134
135
    def test_unknown_action(self):
136
        action_name = 'foo_test_unknown_action'
137
        ds = [{'action': action_name}]
138
        res = helpers.load_list_of_tasks(ds, play=self.mock_play,
139
                                         variable_manager=self.mock_variable_manager, loader=self.fake_loader)
140
        self._assert_is_task_list_or_blocks(res)
141
        self.assertEqual(res[0].action, action_name)
142
143
    def test_block_unknown_action(self):
144
        action_name = 'foo_test_block_unknown_action'
145
        ds = [{
146
            'block': [{'action': action_name}]
147
        }]
148
        res = helpers.load_list_of_tasks(ds, play=self.mock_play,
149
                                         variable_manager=self.mock_variable_manager, loader=self.fake_loader)
150
        self._assert_is_task_list_or_blocks(res)
151
        self.assertIsInstance(res[0], Block)
152
        self._assert_default_block(res[0])
153
154
    def _assert_default_block(self, block):
155
        # the expected defaults
156
        self.assertIsInstance(block.block, list)
157
        self.assertEqual(len(block.block), 1)
158
        self.assertIsInstance(block.rescue, list)
159
        self.assertEqual(len(block.rescue), 0)
160
        self.assertIsInstance(block.always, list)
161
        self.assertEqual(len(block.always), 0)
162
163
    def test_block_unknown_action_use_handlers(self):
164
        ds = [{
165
            'block': [{'action': 'foo_test_block_unknown_action'}]
166
        }]
167
        res = helpers.load_list_of_tasks(ds, play=self.mock_play, use_handlers=True,
168
                                         variable_manager=self.mock_variable_manager, loader=self.fake_loader)
169
        self._assert_is_task_list_or_blocks(res)
170
        self.assertIsInstance(res[0], Block)
171
        self._assert_default_block(res[0])
172
173
    def test_one_bogus_block_use_handlers(self):
174
        ds = [{'block': True}]
175
        self.assertRaisesRegexp(errors.AnsibleParserError,
176
                                "A malformed block was encountered",
177
                                helpers.load_list_of_tasks,
178
                                ds, play=self.mock_play, use_handlers=True,
179
                                variable_manager=self.mock_variable_manager, loader=self.fake_loader)
180
181
    def test_one_bogus_include(self):
182
        ds = [{'include': 'somefile.yml'}]
183
        res = helpers.load_list_of_tasks(ds, play=self.mock_play,
184
                                         variable_manager=self.mock_variable_manager, loader=self.fake_loader)
185
        self.assertIsInstance(res, list)
186
        self.assertEqual(len(res), 0)
187
188
    def test_one_bogus_include_use_handlers(self):
189
        ds = [{'include': 'somefile.yml'}]
190
        res = helpers.load_list_of_tasks(ds, play=self.mock_play, use_handlers=True,
191
                                         variable_manager=self.mock_variable_manager, loader=self.fake_loader)
192
        self.assertIsInstance(res, list)
193
        self.assertEqual(len(res), 0)
194
195
    def test_one_bogus_include_static(self):
196
        ds = [{'include': 'somefile.yml',
197
               'static': 'true'}]
198
        res = helpers.load_list_of_tasks(ds, play=self.mock_play,
199
                                         variable_manager=self.mock_variable_manager, loader=self.fake_loader)
200
        self.assertIsInstance(res, list)
201
        self.assertEqual(len(res), 0)
202
203
    def test_one_include(self):
204
        ds = [{'include': '/dev/null/includes/other_test_include.yml'}]
205
        res = helpers.load_list_of_tasks(ds, play=self.mock_play,
206
                                         variable_manager=self.mock_variable_manager, loader=self.fake_include_loader)
207
        self.assertEqual(len(res), 1)
208
        self._assert_is_task_list_or_blocks(res)
209
210
    def test_one_parent_include(self):
211
        ds = [{'include': '/dev/null/includes/test_include.yml'}]
212
        res = helpers.load_list_of_tasks(ds, play=self.mock_play,
213
                                         variable_manager=self.mock_variable_manager, loader=self.fake_include_loader)
214
        self._assert_is_task_list_or_blocks(res)
215
        self.assertIsInstance(res[0], Block)
216
        self.assertIsInstance(res[0]._parent, TaskInclude)
217
218
    # TODO/FIXME: do this non deprecated way
219
    def test_one_include_tags(self):
220
        ds = [{'include': '/dev/null/includes/other_test_include.yml',
221
               'tags': ['test_one_include_tags_tag1', 'and_another_tagB']
222
               }]
223
        res = helpers.load_list_of_tasks(ds, play=self.mock_play,
224
                                         variable_manager=self.mock_variable_manager, loader=self.fake_include_loader)
225
        self._assert_is_task_list_or_blocks(res)
226
        self.assertIsInstance(res[0], Block)
227
        self.assertIn('test_one_include_tags_tag1', res[0].tags)
228
        self.assertIn('and_another_tagB', res[0].tags)
229
230
    # TODO/FIXME: do this non deprecated way
231
    def test_one_parent_include_tags(self):
232
        ds = [{'include': '/dev/null/includes/test_include.yml',
233
               # 'vars': {'tags': ['test_one_parent_include_tags_tag1', 'and_another_tag2']}
234
               'tags': ['test_one_parent_include_tags_tag1', 'and_another_tag2']
235
               }
236
              ]
237
        res = helpers.load_list_of_tasks(ds, play=self.mock_play,
238
                                         variable_manager=self.mock_variable_manager, loader=self.fake_include_loader)
239
        self._assert_is_task_list_or_blocks(res)
240
        self.assertIsInstance(res[0], Block)
241
        self.assertIn('test_one_parent_include_tags_tag1', res[0].tags)
242
        self.assertIn('and_another_tag2', res[0].tags)
243
244
    # It would be useful to be able to tell what kind of deprecation we encountered and where we encountered it.
245
    def test_one_include_tags_deprecated_mixed(self):
246
        ds = [{'include': "/dev/null/includes/other_test_include.yml",
247
               'vars': {'tags': "['tag_on_include1', 'tag_on_include2']"},
248
               'tags': 'mixed_tag1, mixed_tag2'
249
               }]
250
        self.assertRaisesRegexp(errors.AnsibleParserError, 'Mixing styles',
251
                                helpers.load_list_of_tasks,
252
                                ds, play=self.mock_play,
253
                                variable_manager=self.mock_variable_manager, loader=self.fake_include_loader)
254
255
    def test_one_include_tags_deprecated_include(self):
256
        ds = [{'include': '/dev/null/includes/other_test_include.yml',
257
               'vars': {'tags': ['include_tag1_deprecated', 'and_another_tagB_deprecated']}
258
               }]
259
        res = helpers.load_list_of_tasks(ds, play=self.mock_play,
260
                                         variable_manager=self.mock_variable_manager, loader=self.fake_include_loader)
261
        self._assert_is_task_list_or_blocks(res)
262
        self.assertIsInstance(res[0], Block)
263
        self.assertIn('include_tag1_deprecated', res[0].tags)
264
        self.assertIn('and_another_tagB_deprecated', res[0].tags)
265
266
    def test_one_include_use_handlers(self):
267
        ds = [{'include': '/dev/null/includes/other_test_include.yml'}]
268
        res = helpers.load_list_of_tasks(ds, play=self.mock_play,
269
                                         use_handlers=True,
270
                                         variable_manager=self.mock_variable_manager, loader=self.fake_include_loader)
271
        self._assert_is_task_list_or_blocks(res)
272
        self.assertIsInstance(res[0], Handler)
273
274
    def test_one_parent_include_use_handlers(self):
275
        ds = [{'include': '/dev/null/includes/test_include.yml'}]
276
        res = helpers.load_list_of_tasks(ds, play=self.mock_play,
277
                                         use_handlers=True,
278
                                         variable_manager=self.mock_variable_manager, loader=self.fake_include_loader)
279
        self._assert_is_task_list_or_blocks(res)
280
        self.assertIsInstance(res[0], Handler)
281
282
        # default for Handler
283
        self.assertEqual(res[0].listen, [])
284
285
    # TODO/FIXME: this doesn't seen right
286
    #  figure out how to get the non-static errors to be raised, this seems to just ignore everything
287
    def test_one_include_not_static(self):
288
        ds = [{
289
            'include': '/dev/null/includes/static_test_include.yml',
290
            'static': False
291
        }]
292
        # a_block = Block()
293
        ti_ds = {'include': '/dev/null/includes/ssdftatic_test_include.yml'}
294
        a_task_include = TaskInclude()
295
        ti = a_task_include.load(ti_ds)
296
        res = helpers.load_list_of_tasks(ds, play=self.mock_play,
297
                                         block=ti,
298
                                         variable_manager=self.mock_variable_manager, loader=self.fake_include_loader)
299
        self._assert_is_task_list_or_blocks(res)
300
        self.assertIsInstance(res[0], Task)
301
        self.assertEqual(res[0].args['_raw_params'], '/dev/null/includes/static_test_include.yml')
302
303
    # TODO/FIXME: This two get stuck trying to make a mock_block into a TaskInclude
304
#    def test_one_include(self):
305
#        ds = [{'include': 'other_test_include.yml'}]
306
#        res = helpers.load_list_of_tasks(ds, play=self.mock_play,
307
#                                         block=self.mock_block,
308
#                                         variable_manager=self.mock_variable_manager, loader=self.fake_include_loader)
309
#        print(res)
310
311
#    def test_one_parent_include(self):
312
#        ds = [{'include': 'test_include.yml'}]
313
#        res = helpers.load_list_of_tasks(ds, play=self.mock_play,
314
#                                         block=self.mock_block,
315
#                                         variable_manager=self.mock_variable_manager, loader=self.fake_include_loader)
316
#        print(res)
317
318
    def test_one_bogus_include_role(self):
319
        ds = [{'include_role': {'name': 'bogus_role'}}]
320
        res = helpers.load_list_of_tasks(ds, play=self.mock_play,
321
                                         block=self.mock_block,
322
                                         variable_manager=self.mock_variable_manager, loader=self.fake_role_loader)
323
        self.assertEqual(len(res), 1)
324
        self._assert_is_task_list_or_blocks(res)
325
326
    def test_one_bogus_include_role_use_handlers(self):
327
        ds = [{'include_role': {'name': 'bogus_role'}}]
328
        res = helpers.load_list_of_tasks(ds, play=self.mock_play, use_handlers=True,
329
                                         block=self.mock_block,
330
                                         variable_manager=self.mock_variable_manager,
331
                                         loader=self.fake_role_loader)
332
        self.assertEqual(len(res), 1)
333
        self._assert_is_task_list_or_blocks(res)
334
335
336
class TestLoadListOfRoles(unittest.TestCase, MixinForMocks):
337
    def setUp(self):
338
        self._setup()
339
340
    def test_ds_not_list(self):
341
        ds = {}
342
        self.assertRaises(AssertionError, helpers.load_list_of_roles,
343
                          ds, self.mock_play)
344
345
    def test_empty_role(self):
346
        ds = [{}]
347
        self.assertRaisesRegexp(errors.AnsibleError,
348
                                "role definitions must contain a role name",
349
                                helpers.load_list_of_roles,
350
                                ds, self.mock_play,
351
                                variable_manager=self.mock_variable_manager, loader=self.fake_role_loader)
352
353
    def test_empty_role_just_name(self):
354
        ds = [{'name': 'bogus_role'}]
355
        res = helpers.load_list_of_roles(ds, self.mock_play,
356
                                         variable_manager=self.mock_variable_manager, loader=self.fake_role_loader)
357
        self.assertIsInstance(res, list)
358
        for r in res:
359
            self.assertIsInstance(r, RoleInclude)
360
361
    def test_block_unknown_action(self):
362
        ds = [{
363
            'block': [{'action': 'foo_test_block_unknown_action'}]
364
        }]
365
        ds = [{'name': 'bogus_role'}]
366
        res = helpers.load_list_of_roles(ds, self.mock_play,
367
                                         variable_manager=self.mock_variable_manager, loader=self.fake_role_loader)
368
        self.assertIsInstance(res, list)
369
        for r in res:
370
            self.assertIsInstance(r, RoleInclude)
371
372
373
class TestLoadListOfBlocks(unittest.TestCase, MixinForMocks):
374
    def setUp(self):
375
        self._setup()
376
377
    def test_ds_not_list(self):
378
        ds = {}
379
        mock_play = MagicMock(name='MockPlay')
380
        self.assertRaises(AssertionError, helpers.load_list_of_blocks,
381
                          ds, mock_play, parent_block=None, role=None, task_include=None, use_handlers=False, variable_manager=None, loader=None)
382
383
    def test_empty_block(self):
384
        ds = [{}]
385
        mock_play = MagicMock(name='MockPlay')
386
        self.assertRaisesRegexp(errors.AnsibleParserError,
387
                                "no module/action detected in task",
388
                                helpers.load_list_of_blocks,
389
                                ds, mock_play,
390
                                parent_block=None,
391
                                role=None,
392
                                task_include=None,
393
                                use_handlers=False,
394
                                variable_manager=None,
395
                                loader=None)
396
397
    def test_block_unknown_action(self):
398
        ds = [{'action': 'foo'}]
399
        mock_play = MagicMock(name='MockPlay')
400
        res = helpers.load_list_of_blocks(ds, mock_play, parent_block=None, role=None, task_include=None, use_handlers=False, variable_manager=None,
401
                                          loader=None)
402
403
        self.assertIsInstance(res, list)
404
        for block in res:
405
            self.assertIsInstance(block, Block)