1# -*- coding: utf-8 -*-
2#
3# Copyright (c) 2017 F5 Networks Inc.
4# GNU General Public License v3.0 (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
5
6from __future__ import (absolute_import, division, print_function)
7__metaclass__ = type
8
9import os
10import json
11import pytest
12import sys
13
14if sys.version_info < (2, 7):
15    pytestmark = pytest.mark.skip("F5 Ansible modules require Python >= 2.7")
16
17from ansible.module_utils.basic import AnsibleModule
18
19from ansible_collections.f5networks.f5_modules.plugins.modules.bigip_pool import (
20    ApiParameters, ModuleParameters, ModuleManager, ArgumentSpec
21)
22from ansible_collections.f5networks.f5_modules.plugins.module_utils.common import F5ModuleError
23from ansible_collections.f5networks.f5_modules.tests.unit.compat import unittest
24from ansible_collections.f5networks.f5_modules.tests.unit.compat.mock import Mock, patch
25from ansible_collections.f5networks.f5_modules.tests.unit.modules.utils import set_module_args
26
27
28fixture_path = os.path.join(os.path.dirname(__file__), 'fixtures')
29fixture_data = {}
30
31
32def load_fixture(name):
33    path = os.path.join(fixture_path, name)
34
35    if path in fixture_data:
36        return fixture_data[path]
37
38    with open(path) as f:
39        data = f.read()
40
41    try:
42        data = json.loads(data)
43    except Exception:
44        pass
45
46    fixture_data[path] = data
47    return data
48
49
50class TestParameters(unittest.TestCase):
51    def test_module_parameters(self):
52        args = dict(
53            monitor_type='m_of_n',
54            monitors=['/Common/Fake', '/Common/Fake2'],
55            quorum=1,
56            slow_ramp_time=200,
57            reselect_tries=5,
58            service_down_action='drop'
59        )
60
61        p = ModuleParameters(params=args)
62        assert p.monitor_type == 'm_of_n'
63        assert p.quorum == 1
64        assert p.monitors == 'min 1 of { /Common/Fake /Common/Fake2 }'
65        assert p.slow_ramp_time == 200
66        assert p.reselect_tries == 5
67        assert p.service_down_action == 'drop'
68
69    def test_api_parameters(self):
70        args = dict(
71            monitor="/Common/Fake and /Common/Fake2 ",
72            slowRampTime=200,
73            reselectTries=5,
74            serviceDownAction='drop'
75        )
76
77        p = ApiParameters(params=args)
78        assert p.monitors == '/Common/Fake and /Common/Fake2'
79        assert p.slow_ramp_time == 200
80        assert p.reselect_tries == 5
81        assert p.service_down_action == 'drop'
82
83    def test_unknown_module_lb_method(self):
84        args = dict(
85            lb_method='obscure_hyphenated_fake_method',
86        )
87        with pytest.raises(F5ModuleError):
88            p = ModuleParameters(params=args)
89            assert p.lb_method == 'foo'
90
91    def test_unknown_api_lb_method(self):
92        args = dict(
93            loadBalancingMode='obscure_hypenated_fake_method'
94        )
95        with pytest.raises(F5ModuleError):
96            p = ApiParameters(params=args)
97            assert p.lb_method == 'foo'
98
99
100class TestManager(unittest.TestCase):
101    def setUp(self):
102        self.spec = ArgumentSpec()
103        self.p2 = patch('ansible_collections.f5networks.f5_modules.plugins.modules.bigip_pool.tmos_version')
104        self.p3 = patch('ansible_collections.f5networks.f5_modules.plugins.modules.bigip_pool.send_teem')
105        self.m2 = self.p2.start()
106        self.m2.return_value = '14.1.0'
107        self.m3 = self.p3.start()
108        self.m3.return_value = True
109
110    def tearDown(self):
111        self.p2.stop()
112        self.p3.stop()
113
114    def test_create_pool(self, *args):
115        set_module_args(dict(
116            pool='fake_pool',
117            description='fakepool',
118            service_down_action='drop',
119            lb_method='round-robin',
120            partition='Common',
121            slow_ramp_time=10,
122            reselect_tries=1,
123            provider=dict(
124                server='localhost',
125                password='password',
126                user='admin'
127            )
128        ))
129
130        module = AnsibleModule(
131            argument_spec=self.spec.argument_spec,
132            supports_check_mode=self.spec.supports_check_mode,
133            mutually_exclusive=self.spec.mutually_exclusive,
134            required_one_of=self.spec.required_one_of
135        )
136
137        mm = ModuleManager(module=module)
138        mm.create_on_device = Mock(return_value=True)
139        mm.exists = Mock(return_value=False)
140
141        results = mm.exec_module()
142
143        assert results['changed'] is True
144        assert results['name'] == 'fake_pool'
145        assert results['description'] == 'fakepool'
146        assert results['service_down_action'] == 'drop'
147        assert results['lb_method'] == 'round-robin'
148        assert results['slow_ramp_time'] == 10
149        assert results['reselect_tries'] == 1
150
151    def test_create_pool_monitor_type_missing(self, *args):
152        set_module_args(dict(
153            pool='fake_pool',
154            lb_method='round-robin',
155            partition='Common',
156            monitors=['/Common/tcp', '/Common/http'],
157            provider=dict(
158                server='localhost',
159                password='password',
160                user='admin'
161            )
162        ))
163
164        module = AnsibleModule(
165            argument_spec=self.spec.argument_spec,
166            supports_check_mode=self.spec.supports_check_mode,
167            mutually_exclusive=self.spec.mutually_exclusive,
168            required_one_of=self.spec.required_one_of
169        )
170
171        mm = ModuleManager(module=module)
172        mm.create_on_device = Mock(return_value=True)
173        mm.exists = Mock(return_value=False)
174
175        results = mm.exec_module()
176
177        assert results['changed'] is True
178        assert results['name'] == 'fake_pool'
179        assert results['monitors'] == ['/Common/http', '/Common/tcp']
180        assert results['monitor_type'] == 'and_list'
181
182    def test_create_pool_monitors_missing(self, *args):
183        set_module_args(dict(
184            pool='fake_pool',
185            lb_method='round-robin',
186            partition='Common',
187            monitor_type='and_list',
188            provider=dict(
189                server='localhost',
190                password='password',
191                user='admin'
192            )
193        ))
194
195        module = AnsibleModule(
196            argument_spec=self.spec.argument_spec,
197            supports_check_mode=self.spec.supports_check_mode,
198            mutually_exclusive=self.spec.mutually_exclusive,
199            required_one_of=self.spec.required_one_of
200        )
201
202        mm = ModuleManager(module=module)
203        mm.create_on_device = Mock(return_value=True)
204        mm.exists = Mock(return_value=False)
205
206        msg = "The 'monitors' parameter cannot be empty when " \
207              "'monitor_type' parameter is specified"
208        with pytest.raises(F5ModuleError) as err:
209            mm.exec_module()
210
211        assert str(err.value) == msg
212
213    def test_create_pool_quorum_missing(self, *args):
214        set_module_args(dict(
215            pool='fake_pool',
216            lb_method='round-robin',
217            partition='Common',
218            monitor_type='m_of_n',
219            monitors=['/Common/tcp', '/Common/http'],
220            provider=dict(
221                server='localhost',
222                password='password',
223                user='admin'
224            )
225        ))
226
227        module = AnsibleModule(
228            argument_spec=self.spec.argument_spec,
229            supports_check_mode=self.spec.supports_check_mode,
230            mutually_exclusive=self.spec.mutually_exclusive,
231            required_one_of=self.spec.required_one_of
232        )
233
234        mm = ModuleManager(module=module)
235        mm.create_on_device = Mock(return_value=True)
236        mm.exists = Mock(return_value=False)
237
238        msg = "Quorum value must be specified with monitor_type 'm_of_n'."
239        with pytest.raises(F5ModuleError) as err:
240            mm.exec_module()
241
242        assert str(err.value) == msg
243
244    def test_create_pool_monitor_and_list(self, *args):
245        set_module_args(dict(
246            pool='fake_pool',
247            partition='Common',
248            monitor_type='and_list',
249            monitors=['/Common/tcp', '/Common/http'],
250            provider=dict(
251                server='localhost',
252                password='password',
253                user='admin'
254            )
255        ))
256
257        module = AnsibleModule(
258            argument_spec=self.spec.argument_spec,
259            supports_check_mode=self.spec.supports_check_mode,
260            mutually_exclusive=self.spec.mutually_exclusive,
261            required_one_of=self.spec.required_one_of
262        )
263
264        mm = ModuleManager(module=module)
265        mm.create_on_device = Mock(return_value=True)
266        mm.exists = Mock(return_value=False)
267
268        results = mm.exec_module()
269
270        assert results['changed'] is True
271        assert results['name'] == 'fake_pool'
272        assert results['monitors'] == ['/Common/http', '/Common/tcp']
273        assert results['monitor_type'] == 'and_list'
274
275    def test_create_pool_monitor_m_of_n(self, *args):
276        set_module_args(dict(
277            pool='fake_pool',
278            partition='Common',
279            monitor_type='m_of_n',
280            quorum=1,
281            monitors=['/Common/tcp', '/Common/http'],
282            provider=dict(
283                server='localhost',
284                password='password',
285                user='admin'
286            )
287        ))
288
289        module = AnsibleModule(
290            argument_spec=self.spec.argument_spec,
291            supports_check_mode=self.spec.supports_check_mode,
292            mutually_exclusive=self.spec.mutually_exclusive,
293            required_one_of=self.spec.required_one_of
294        )
295
296        mm = ModuleManager(module=module)
297        mm.create_on_device = Mock(return_value=True)
298        mm.exists = Mock(return_value=False)
299
300        results = mm.exec_module()
301
302        assert results['changed'] is True
303        assert results['name'] == 'fake_pool'
304        assert results['monitors'] == ['/Common/http', '/Common/tcp']
305        assert results['monitor_type'] == 'm_of_n'
306
307    def test_update_monitors(self, *args):
308        set_module_args(dict(
309            name='test_pool',
310            partition='Common',
311            monitor_type='and_list',
312            monitors=['/Common/http', '/Common/tcp'],
313            provider=dict(
314                server='localhost',
315                password='password',
316                user='admin'
317            )
318        ))
319
320        module = AnsibleModule(
321            argument_spec=self.spec.argument_spec,
322            supports_check_mode=self.spec.supports_check_mode,
323            mutually_exclusive=self.spec.mutually_exclusive,
324            required_one_of=self.spec.required_one_of
325        )
326
327        mm = ModuleManager(module=module)
328
329        current = ApiParameters(params=load_fixture('load_ltm_pool.json'))
330
331        mm.update_on_device = Mock(return_value=True)
332        mm.exists = Mock(return_value=True)
333        mm.read_current_from_device = Mock(return_value=current)
334
335        results = mm.exec_module()
336
337        assert results['changed'] is True
338        assert results['monitor_type'] == 'and_list'
339
340    def test_create_pool_monitor_and_list_no_partition(self, *args):
341        set_module_args(dict(
342            pool='fake_pool',
343            monitor_type='and_list',
344            monitors=['tcp', 'http'],
345            provider=dict(
346                server='localhost',
347                password='password',
348                user='admin'
349            )
350        ))
351
352        module = AnsibleModule(
353            argument_spec=self.spec.argument_spec,
354            supports_check_mode=self.spec.supports_check_mode,
355            mutually_exclusive=self.spec.mutually_exclusive,
356            required_one_of=self.spec.required_one_of
357        )
358
359        mm = ModuleManager(module=module)
360        mm.create_on_device = Mock(return_value=True)
361        mm.exists = Mock(return_value=False)
362
363        results = mm.exec_module()
364
365        assert results['changed'] is True
366        assert results['name'] == 'fake_pool'
367        assert results['monitors'] == ['/Common/http', '/Common/tcp']
368        assert results['monitor_type'] == 'and_list'
369
370    def test_create_pool_monitor_m_of_n_no_partition(self, *args):
371        set_module_args(dict(
372            pool='fake_pool',
373            monitor_type='m_of_n',
374            quorum=1,
375            monitors=['tcp', 'http'],
376            provider=dict(
377                server='localhost',
378                password='password',
379                user='admin'
380            )
381        ))
382
383        module = AnsibleModule(
384            argument_spec=self.spec.argument_spec,
385            supports_check_mode=self.spec.supports_check_mode,
386            mutually_exclusive=self.spec.mutually_exclusive,
387            required_one_of=self.spec.required_one_of
388        )
389
390        mm = ModuleManager(module=module)
391        mm.create_on_device = Mock(return_value=True)
392        mm.exists = Mock(return_value=False)
393
394        results = mm.exec_module()
395
396        assert results['changed'] is True
397        assert results['name'] == 'fake_pool'
398        assert results['monitors'] == ['/Common/http', '/Common/tcp']
399        assert results['monitor_type'] == 'm_of_n'
400
401    def test_create_pool_monitor_and_list_custom_partition(self, *args):
402        set_module_args(dict(
403            pool='fake_pool',
404            partition='Testing',
405            monitor_type='and_list',
406            monitors=['tcp', 'http'],
407            provider=dict(
408                server='localhost',
409                password='password',
410                user='admin'
411            )
412        ))
413
414        module = AnsibleModule(
415            argument_spec=self.spec.argument_spec,
416            supports_check_mode=self.spec.supports_check_mode,
417            mutually_exclusive=self.spec.mutually_exclusive,
418            required_one_of=self.spec.required_one_of
419        )
420
421        mm = ModuleManager(module=module)
422        mm.create_on_device = Mock(return_value=True)
423        mm.exists = Mock(return_value=False)
424
425        results = mm.exec_module()
426
427        assert results['changed'] is True
428        assert results['name'] == 'fake_pool'
429        assert results['monitors'] == ['/Testing/http', '/Testing/tcp']
430        assert results['monitor_type'] == 'and_list'
431
432    def test_create_pool_monitor_m_of_n_custom_partition(self, *args):
433        set_module_args(dict(
434            pool='fake_pool',
435            partition='Testing',
436            monitor_type='m_of_n',
437            quorum=1,
438            monitors=['tcp', 'http'],
439            provider=dict(
440                server='localhost',
441                password='password',
442                user='admin'
443            )
444        ))
445
446        module = AnsibleModule(
447            argument_spec=self.spec.argument_spec,
448            supports_check_mode=self.spec.supports_check_mode,
449            mutually_exclusive=self.spec.mutually_exclusive,
450            required_one_of=self.spec.required_one_of
451        )
452
453        mm = ModuleManager(module=module)
454        mm.create_on_device = Mock(return_value=True)
455        mm.exists = Mock(return_value=False)
456
457        results = mm.exec_module()
458
459        assert results['changed'] is True
460        assert results['name'] == 'fake_pool'
461        assert results['monitors'] == ['/Testing/http', '/Testing/tcp']
462        assert results['monitor_type'] == 'm_of_n'
463
464    def test_create_pool_with_metadata(self, *args):
465        set_module_args(dict(
466            pool='fake_pool',
467            metadata=dict(ansible='2.4'),
468            provider=dict(
469                server='localhost',
470                password='password',
471                user='admin'
472            )
473        ))
474
475        module = AnsibleModule(
476            argument_spec=self.spec.argument_spec,
477            supports_check_mode=self.spec.supports_check_mode,
478            mutually_exclusive=self.spec.mutually_exclusive,
479            required_one_of=self.spec.required_one_of
480        )
481
482        mm = ModuleManager(module=module)
483        mm.create_on_device = Mock(return_value=True)
484        mm.exists = Mock(return_value=False)
485
486        results = mm.exec_module()
487
488        assert results['changed'] is True
489        assert results['name'] == 'fake_pool'
490        assert 'metadata' in results
491        assert 'ansible' in results['metadata']
492        assert results['metadata']['ansible'] == '2.4'
493
494    def test_create_aggregate_pools(self, *args):
495        set_module_args(dict(
496            aggregate=[
497                dict(
498                    pool='fake_pool',
499                    description='fakepool',
500                    service_down_action='drop',
501                    lb_method='round-robin',
502                    partition='Common',
503                    slow_ramp_time=10,
504                    reselect_tries=1,
505                ),
506                dict(
507                    pool='fake_pool2',
508                    description='fakepool2',
509                    service_down_action='drop',
510                    lb_method='predictive-node',
511                    partition='Common',
512                    slow_ramp_time=110,
513                    reselect_tries=2,
514                )
515            ],
516            provider=dict(
517                server='localhost',
518                password='password',
519                user='admin'
520            )
521        ))
522
523        module = AnsibleModule(
524            argument_spec=self.spec.argument_spec,
525            supports_check_mode=self.spec.supports_check_mode,
526            mutually_exclusive=self.spec.mutually_exclusive,
527            required_one_of=self.spec.required_one_of
528        )
529
530        mm = ModuleManager(module=module)
531        mm.create_on_device = Mock(return_value=True)
532        mm.exists = Mock(return_value=False)
533
534        results = mm.exec_module()
535
536        assert results['changed'] is True
537