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_half_open 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            ip='10.10.10.10',
56            port=80,
57            interval=20,
58            timeout=30,
59            time_until_up=60,
60            partition='Common'
61        )
62
63        p = Parameters(params=args)
64        assert p.name == 'foo'
65        assert p.parent == '/Common/parent'
66        assert p.ip == '10.10.10.10'
67        assert p.port == 80
68        assert p.type == 'tcp_half_open'
69        assert p.destination == '10.10.10.10:80'
70        assert p.interval == 20
71        assert p.timeout == 30
72        assert p.time_until_up == 60
73
74    def test_module_parameters_ints_as_strings(self):
75        args = dict(
76            name='foo',
77            parent='parent',
78            ip='10.10.10.10',
79            port=80,
80            interval='20',
81            timeout='30',
82            time_until_up='60',
83            partition='Common'
84        )
85
86        p = Parameters(params=args)
87        assert p.name == 'foo'
88        assert p.parent == '/Common/parent'
89        assert p.ip == '10.10.10.10'
90        assert p.port == 80
91        assert p.type == 'tcp_half_open'
92        assert p.destination == '10.10.10.10:80'
93        assert p.interval == 20
94        assert p.timeout == 30
95        assert p.time_until_up == 60
96
97    def test_api_parameters(self):
98        args = dict(
99            name='foo',
100            defaultsFrom='/Common/parent',
101            destination='10.10.10.10:80',
102            interval=20,
103            timeout=30,
104            timeUntilUp=60
105        )
106
107        p = Parameters(params=args)
108        assert p.name == 'foo'
109        assert p.parent == '/Common/parent'
110        assert p.ip == '10.10.10.10'
111        assert p.port == 80
112        assert p.type == 'tcp_half_open'
113        assert p.destination == '10.10.10.10:80'
114        assert p.interval == 20
115        assert p.timeout == 30
116        assert p.time_until_up == 60
117
118
119class TestManager(unittest.TestCase):
120    def setUp(self):
121        self.spec = ArgumentSpec()
122        self.p2 = patch('ansible_collections.f5networks.f5_modules.plugins.modules.bigip_monitor_tcp_half_open.tmos_version')
123        self.p3 = patch('ansible_collections.f5networks.f5_modules.plugins.modules.bigip_monitor_tcp_half_open.send_teem')
124        self.m2 = self.p2.start()
125        self.m2.return_value = '14.1.0'
126        self.m3 = self.p3.start()
127        self.m3.return_value = True
128
129    def tearDown(self):
130        self.p2.stop()
131        self.p3.stop()
132
133    def test_create_monitor(self, *args):
134        set_module_args(dict(
135            name='foo',
136            ip='10.10.10.10',
137            port=80,
138            interval=20,
139            timeout=30,
140            time_until_up=60,
141            provider=dict(
142                server='localhost',
143                password='password',
144                user='admin'
145            )
146        ))
147
148        module = AnsibleModule(
149            argument_spec=self.spec.argument_spec,
150            supports_check_mode=self.spec.supports_check_mode
151        )
152
153        # Override methods in the specific type of manager
154        mm = ModuleManager(module=module)
155        mm.exists = Mock(side_effect=[False, True])
156        mm.create_on_device = Mock(return_value=True)
157
158        results = mm.exec_module()
159
160        assert results['changed'] is True
161
162    def test_create_monitor_idempotent(self, *args):
163        set_module_args(dict(
164            name='foo',
165            ip='10.10.10.10',
166            port=80,
167            interval=20,
168            timeout=30,
169            time_until_up=60,
170            provider=dict(
171                server='localhost',
172                password='password',
173                user='admin'
174            )
175        ))
176
177        current = Parameters(params=load_fixture('load_ltm_monitor_tcp_half_open.json'))
178        module = AnsibleModule(
179            argument_spec=self.spec.argument_spec,
180            supports_check_mode=self.spec.supports_check_mode
181        )
182
183        # Override methods in the specific type of manager
184        mm = ModuleManager(module=module)
185        mm.exists = Mock(return_value=True)
186        mm.read_current_from_device = Mock(return_value=current)
187
188        results = mm.exec_module()
189
190        assert results['changed'] is False
191
192    def test_update_interval(self, *args):
193        set_module_args(dict(
194            name='foo',
195            interval=10,
196            provider=dict(
197                server='localhost',
198                password='password',
199                user='admin'
200            )
201        ))
202
203        current = Parameters(params=load_fixture('load_ltm_monitor_tcp_half_open.json'))
204        module = AnsibleModule(
205            argument_spec=self.spec.argument_spec,
206            supports_check_mode=self.spec.supports_check_mode
207        )
208
209        # Override methods in the specific type of manager
210        mm = ModuleManager(module=module)
211        mm.exists = Mock(return_value=True)
212        mm.read_current_from_device = Mock(return_value=current)
213        mm.update_on_device = Mock(return_value=True)
214
215        results = mm.exec_module()
216
217        assert results['changed'] is True
218        assert results['interval'] == 10
219
220    def test_update_interval_larger_than_existing_timeout(self, *args):
221        set_module_args(dict(
222            name='foo',
223            interval=30,
224            provider=dict(
225                server='localhost',
226                password='password',
227                user='admin'
228            )
229        ))
230
231        current = Parameters(params=load_fixture('load_ltm_monitor_tcp_half_open.json'))
232        module = AnsibleModule(
233            argument_spec=self.spec.argument_spec,
234            supports_check_mode=self.spec.supports_check_mode
235        )
236
237        # Override methods in the specific type of manager
238        mm = ModuleManager(module=module)
239        mm.exists = Mock(return_value=True)
240        mm.read_current_from_device = Mock(return_value=current)
241        mm.update_on_device = Mock(return_value=True)
242
243        with pytest.raises(F5ModuleError) as ex:
244            mm.exec_module()
245
246        assert "must be less than" in str(ex.value)
247
248    def test_update_interval_larger_than_new_timeout(self, *args):
249        set_module_args(dict(
250            name='foo',
251            interval=10,
252            timeout=5,
253            provider=dict(
254                server='localhost',
255                password='password',
256                user='admin'
257            )
258        ))
259
260        current = Parameters(params=load_fixture('load_ltm_monitor_tcp_half_open.json'))
261        module = AnsibleModule(
262            argument_spec=self.spec.argument_spec,
263            supports_check_mode=self.spec.supports_check_mode
264        )
265
266        # Override methods in the specific type of manager
267        mm = ModuleManager(module=module)
268        mm.exists = Mock(return_value=True)
269        mm.read_current_from_device = Mock(return_value=current)
270        mm.update_on_device = Mock(return_value=True)
271
272        with pytest.raises(F5ModuleError) as ex:
273            mm.exec_module()
274
275        assert "must be less than" in str(ex.value)
276
277    def test_update_timeout(self, *args):
278        set_module_args(dict(
279            name='foo',
280            timeout=300,
281            provider=dict(
282                server='localhost',
283                password='password',
284                user='admin'
285            )
286        ))
287
288        current = Parameters(params=load_fixture('load_ltm_monitor_tcp_half_open.json'))
289        module = AnsibleModule(
290            argument_spec=self.spec.argument_spec,
291            supports_check_mode=self.spec.supports_check_mode
292        )
293
294        # Override methods in the specific type of manager
295        mm = ModuleManager(module=module)
296        mm.exists = Mock(return_value=True)
297        mm.read_current_from_device = Mock(return_value=current)
298        mm.update_on_device = Mock(return_value=True)
299
300        results = mm.exec_module()
301        assert results['changed'] is True
302        assert results['timeout'] == 300
303
304    def test_update_time_until_up(self, *args):
305        set_module_args(dict(
306            name='foo',
307            time_until_up=300,
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_half_open.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        results = mm.exec_module()
328
329        assert results['changed'] is True
330        assert results['time_until_up'] == 300
331