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