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