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