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