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
19try:
20    from library.modules.bigip_monitor_udp import Parameters
21    from library.modules.bigip_monitor_udp import ModuleManager
22    from library.modules.bigip_monitor_udp import ArgumentSpec
23
24    from library.module_utils.network.f5.common import F5ModuleError
25
26    # In Ansible 2.8, Ansible changed import paths.
27    from test.units.compat import unittest
28    from test.units.compat.mock import Mock
29
30    from test.units.modules.utils import set_module_args
31except ImportError:
32    from ansible.modules.network.f5.bigip_monitor_udp import Parameters
33    from ansible.modules.network.f5.bigip_monitor_udp import ModuleManager
34    from ansible.modules.network.f5.bigip_monitor_udp import ArgumentSpec
35
36    from ansible.module_utils.network.f5.common import F5ModuleError
37
38    # Ansible 2.8 imports
39    from units.compat import unittest
40    from units.compat.mock import Mock
41
42    from units.modules.utils import set_module_args
43
44
45fixture_path = os.path.join(os.path.dirname(__file__), 'fixtures')
46fixture_data = {}
47
48
49def load_fixture(name):
50    path = os.path.join(fixture_path, name)
51
52    if path in fixture_data:
53        return fixture_data[path]
54
55    with open(path) as f:
56        data = f.read()
57
58    try:
59        data = json.loads(data)
60    except Exception:
61        pass
62
63    fixture_data[path] = data
64    return data
65
66
67class TestParameters(unittest.TestCase):
68    def test_module_parameters(self):
69        args = dict(
70            name='foo',
71            parent='parent',
72            send='this is a send string',
73            receive='this is a receive string',
74            ip='10.10.10.10',
75            port=80,
76            interval=20,
77            timeout=30,
78            time_until_up=60,
79            partition='Common'
80        )
81
82        p = Parameters(params=args)
83        assert p.name == 'foo'
84        assert p.parent == '/Common/parent'
85        assert p.send == 'this is a send string'
86        assert p.receive == 'this is a receive string'
87        assert p.ip == '10.10.10.10'
88        assert p.type == 'udp'
89        assert p.port == 80
90        assert p.destination == '10.10.10.10:80'
91        assert p.interval == 20
92        assert p.timeout == 30
93        assert p.time_until_up == 60
94
95    def test_module_parameters_ints_as_strings(self):
96        args = dict(
97            name='foo',
98            parent='parent',
99            send='this is a send string',
100            receive='this is a receive string',
101            ip='10.10.10.10',
102            port='80',
103            interval='20',
104            timeout='30',
105            time_until_up='60',
106            partition='Common'
107        )
108
109        p = Parameters(params=args)
110        assert p.name == 'foo'
111        assert p.parent == '/Common/parent'
112        assert p.send == 'this is a send string'
113        assert p.receive == 'this is a receive string'
114        assert p.ip == '10.10.10.10'
115        assert p.type == 'udp'
116        assert p.port == 80
117        assert p.destination == '10.10.10.10:80'
118        assert p.interval == 20
119        assert p.timeout == 30
120        assert p.time_until_up == 60
121
122    def test_api_parameters(self):
123        args = dict(
124            name='foo',
125            defaultsFrom='/Common/parent',
126            send='this is a send string',
127            recv='this is a receive string',
128            destination='10.10.10.10:80',
129            interval=20,
130            timeout=30,
131            timeUntilUp=60
132        )
133
134        p = Parameters(params=args)
135        assert p.name == 'foo'
136        assert p.parent == '/Common/parent'
137        assert p.send == 'this is a send string'
138        assert p.receive == 'this is a receive string'
139        assert p.ip == '10.10.10.10'
140        assert p.type == 'udp'
141        assert p.port == 80
142        assert p.destination == '10.10.10.10:80'
143        assert p.interval == 20
144        assert p.timeout == 30
145        assert p.time_until_up == 60
146
147
148class TestManager(unittest.TestCase):
149
150    def setUp(self):
151        self.spec = ArgumentSpec()
152
153    def test_create_monitor(self, *args):
154        set_module_args(dict(
155            name='foo',
156            parent='parent',
157            send='this is a send string',
158            receive='this is a receive string',
159            ip='10.10.10.10',
160            port=80,
161            interval=20,
162            timeout=30,
163            time_until_up=60,
164            partition='Common',
165            provider=dict(
166                server='localhost',
167                password='password',
168                user='admin'
169            )
170        ))
171
172        module = AnsibleModule(
173            argument_spec=self.spec.argument_spec,
174            supports_check_mode=self.spec.supports_check_mode
175        )
176
177        # Override methods in the specific type of manager
178        mm = ModuleManager(module=module)
179        mm.exists = Mock(side_effect=[False, True])
180        mm.create_on_device = Mock(return_value=True)
181
182        results = mm.exec_module()
183
184        assert results['changed'] is True
185        assert results['parent'] == '/Common/parent'
186
187    def test_create_monitor_idempotent(self, *args):
188        set_module_args(dict(
189            name='asdf',
190            parent='udp',
191            send='default send string',
192            receive='hello world',
193            ip='1.1.1.1',
194            port=389,
195            interval=5,
196            timeout=16,
197            time_until_up=0,
198            partition='Common',
199            provider=dict(
200                server='localhost',
201                password='password',
202                user='admin'
203            )
204        ))
205
206        current = Parameters(params=load_fixture('load_ltm_monitor_udp.json'))
207        module = AnsibleModule(
208            argument_spec=self.spec.argument_spec,
209            supports_check_mode=self.spec.supports_check_mode
210        )
211
212        # Override methods in the specific type of manager
213        mm = ModuleManager(module=module)
214        mm.exists = Mock(return_value=True)
215        mm.read_current_from_device = Mock(return_value=current)
216
217        results = mm.exec_module()
218
219        assert results['changed'] is False
220
221    def test_update_port(self, *args):
222        set_module_args(dict(
223            name='asdf',
224            port=800,
225            partition='Common',
226            provider=dict(
227                server='localhost',
228                password='password',
229                user='admin'
230            )
231        ))
232
233        current = Parameters(params=load_fixture('load_ltm_monitor_udp.json'))
234        module = AnsibleModule(
235            argument_spec=self.spec.argument_spec,
236            supports_check_mode=self.spec.supports_check_mode
237        )
238
239        # Override methods in the specific type of manager
240        mm = ModuleManager(module=module)
241        mm.exists = Mock(return_value=True)
242        mm.read_current_from_device = Mock(return_value=current)
243        mm.update_on_device = Mock(return_value=True)
244
245        results = mm.exec_module()
246
247        assert results['changed'] is True
248        assert results['port'] == 800
249
250    def test_update_interval(self, *args):
251        set_module_args(dict(
252            name='foo',
253            interval=10,
254            partition='Common',
255            provider=dict(
256                server='localhost',
257                password='password',
258                user='admin'
259            )
260        ))
261
262        current = Parameters(params=load_fixture('load_ltm_monitor_udp.json'))
263        module = AnsibleModule(
264            argument_spec=self.spec.argument_spec,
265            supports_check_mode=self.spec.supports_check_mode
266        )
267
268        # Override methods in the specific type of manager
269        mm = ModuleManager(module=module)
270        mm.exists = Mock(return_value=True)
271        mm.read_current_from_device = Mock(return_value=current)
272        mm.update_on_device = Mock(return_value=True)
273
274        results = mm.exec_module()
275
276        assert results['changed'] is True
277        assert results['interval'] == 10
278
279    def test_update_interval_larger_than_existing_timeout(self, *args):
280        set_module_args(dict(
281            name='asdf',
282            interval=30,
283            partition='Common',
284            provider=dict(
285                server='localhost',
286                password='password',
287                user='admin'
288            )
289        ))
290
291        current = Parameters(params=load_fixture('load_ltm_monitor_udp.json'))
292        module = AnsibleModule(
293            argument_spec=self.spec.argument_spec,
294            supports_check_mode=self.spec.supports_check_mode
295        )
296
297        # Override methods in the specific type of manager
298        mm = ModuleManager(module=module)
299        mm.exists = Mock(return_value=True)
300        mm.read_current_from_device = Mock(return_value=current)
301        mm.update_on_device = Mock(return_value=True)
302
303        with pytest.raises(F5ModuleError) as ex:
304            mm.exec_module()
305
306        assert "must be less than" in str(ex.value)
307
308    def test_update_interval_larger_than_new_timeout(self, *args):
309        set_module_args(dict(
310            name='asdf',
311            interval=10,
312            timeout=5,
313            partition='Common',
314            provider=dict(
315                server='localhost',
316                password='password',
317                user='admin'
318            )
319        ))
320
321        current = Parameters(params=load_fixture('load_ltm_monitor_udp.json'))
322        module = AnsibleModule(
323            argument_spec=self.spec.argument_spec,
324            supports_check_mode=self.spec.supports_check_mode
325        )
326
327        # Override methods in the specific type of manager
328        mm = ModuleManager(module=module)
329        mm.exists = Mock(return_value=True)
330        mm.read_current_from_device = Mock(return_value=current)
331        mm.update_on_device = Mock(return_value=True)
332
333        with pytest.raises(F5ModuleError) as ex:
334            mm.exec_module()
335
336        assert "must be less than" in str(ex.value)
337
338    def test_update_send(self, *args):
339        set_module_args(dict(
340            name='asdf',
341            send='this is another send string',
342            partition='Common',
343            provider=dict(
344                server='localhost',
345                password='password',
346                user='admin'
347            )
348        ))
349
350        current = Parameters(params=load_fixture('load_ltm_monitor_udp.json'))
351        module = AnsibleModule(
352            argument_spec=self.spec.argument_spec,
353            supports_check_mode=self.spec.supports_check_mode
354        )
355
356        # Override methods in the specific type of manager
357        mm = ModuleManager(module=module)
358        mm.exists = Mock(return_value=True)
359        mm.read_current_from_device = Mock(return_value=current)
360        mm.update_on_device = Mock(return_value=True)
361
362        results = mm.exec_module()
363
364        assert results['changed'] is True
365        assert results['send'] == 'this is another send string'
366
367    def test_update_receive(self, *args):
368        set_module_args(dict(
369            name='asdf',
370            receive='this is another receive string',
371            partition='Common',
372            provider=dict(
373                server='localhost',
374                password='password',
375                user='admin'
376            )
377        ))
378
379        current = Parameters(params=load_fixture('load_ltm_monitor_udp.json'))
380        module = AnsibleModule(
381            argument_spec=self.spec.argument_spec,
382            supports_check_mode=self.spec.supports_check_mode
383        )
384
385        # Override methods in the specific type of manager
386        mm = ModuleManager(module=module)
387        mm.exists = Mock(return_value=True)
388        mm.read_current_from_device = Mock(return_value=current)
389        mm.update_on_device = Mock(return_value=True)
390
391        results = mm.exec_module()
392
393        assert results['changed'] is True
394        assert results['receive'] == 'this is another receive string'
395
396    def test_update_timeout(self, *args):
397        set_module_args(dict(
398            name='asdf',
399            timeout=300,
400            partition='Common',
401            provider=dict(
402                server='localhost',
403                password='password',
404                user='admin'
405            )
406        ))
407
408        current = Parameters(params=load_fixture('load_ltm_monitor_udp.json'))
409        module = AnsibleModule(
410            argument_spec=self.spec.argument_spec,
411            supports_check_mode=self.spec.supports_check_mode
412        )
413
414        # Override methods in the specific type of manager
415        mm = ModuleManager(module=module)
416        mm.exists = Mock(return_value=True)
417        mm.read_current_from_device = Mock(return_value=current)
418        mm.update_on_device = Mock(return_value=True)
419
420        results = mm.exec_module()
421
422        assert results['changed'] is True
423        assert results['timeout'] == 300
424
425    def test_update_time_until_up(self, *args):
426        set_module_args(dict(
427            name='asdf',
428            time_until_up=300,
429            partition='Common',
430            provider=dict(
431                server='localhost',
432                password='password',
433                user='admin'
434            )
435        ))
436
437        current = Parameters(params=load_fixture('load_ltm_monitor_udp.json'))
438        module = AnsibleModule(
439            argument_spec=self.spec.argument_spec,
440            supports_check_mode=self.spec.supports_check_mode
441        )
442
443        # Override methods in the specific type of manager
444        mm = ModuleManager(module=module)
445        mm.exists = Mock(return_value=True)
446        mm.read_current_from_device = Mock(return_value=current)
447        mm.update_on_device = Mock(return_value=True)
448
449        results = mm.exec_module()
450
451        assert results['changed'] is True
452        assert results['time_until_up'] == 300
453