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_device_connectivity import (
20    ApiParameters, ModuleParameters, 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
27fixture_path = os.path.join(os.path.dirname(__file__), 'fixtures')
28fixture_data = {}
29
30
31def load_fixture(name):
32    path = os.path.join(fixture_path, name)
33
34    if path in fixture_data:
35        return fixture_data[path]
36
37    with open(path) as f:
38        data = f.read()
39
40    try:
41        data = json.loads(data)
42    except Exception:
43        pass
44
45    fixture_data[path] = data
46    return data
47
48
49class TestParameters(unittest.TestCase):
50    def test_module_parameters(self):
51        args = dict(
52            multicast_port='1010',
53            multicast_address='10.10.10.10',
54            multicast_interface='eth0',
55            failover_multicast=True,
56            unicast_failover=[
57                dict(
58                    address='20.20.20.20',
59                    port='1234'
60                )
61            ],
62            mirror_primary_address='1.2.3.4',
63            mirror_secondary_address='5.6.7.8',
64            config_sync_ip='4.3.2.1',
65            state='present',
66        )
67        p = ModuleParameters(params=args)
68        assert p.multicast_port == 1010
69        assert p.multicast_address == '10.10.10.10'
70        assert p.multicast_interface == 'eth0'
71        assert p.failover_multicast is True
72        assert p.mirror_primary_address == '1.2.3.4'
73        assert p.mirror_secondary_address == '5.6.7.8'
74        assert p.config_sync_ip == '4.3.2.1'
75        assert len(p.unicast_failover) == 1
76        assert 'effectiveIp' in p.unicast_failover[0]
77        assert 'effectivePort' in p.unicast_failover[0]
78        assert 'port' in p.unicast_failover[0]
79        assert 'ip' in p.unicast_failover[0]
80        assert p.unicast_failover[0]['effectiveIp'] == '20.20.20.20'
81        assert p.unicast_failover[0]['ip'] == '20.20.20.20'
82        assert p.unicast_failover[0]['port'] == 1234
83        assert p.unicast_failover[0]['effectivePort'] == 1234
84
85    def test_api_parameters(self):
86        params = load_fixture('load_tm_cm_device.json')
87        p = ApiParameters(params=params)
88        assert p.multicast_port == 62960
89        assert p.multicast_address == '224.0.0.245'
90        assert p.multicast_interface == 'eth0'
91        assert p.mirror_primary_address == '10.2.2.2'
92        assert p.mirror_secondary_address == '10.2.3.2'
93        assert p.config_sync_ip == '10.2.2.2'
94        assert len(p.unicast_failover) == 2
95        assert 'effectiveIp' in p.unicast_failover[0]
96        assert 'effectivePort' in p.unicast_failover[0]
97        assert 'port' in p.unicast_failover[0]
98        assert 'ip' in p.unicast_failover[0]
99        assert p.unicast_failover[0]['effectiveIp'] == 'management-ip'
100        assert p.unicast_failover[0]['ip'] == 'management-ip'
101        assert p.unicast_failover[0]['port'] == 1026
102        assert p.unicast_failover[0]['effectivePort'] == 1026
103
104
105class TestManager(unittest.TestCase):
106
107    def setUp(self):
108        self.spec = ArgumentSpec()
109        self.p2 = patch('ansible_collections.f5networks.f5_modules.plugins.modules.bigip_device_connectivity.tmos_version')
110        self.p3 = patch('ansible_collections.f5networks.f5_modules.plugins.modules.bigip_device_connectivity.send_teem')
111        self.m2 = self.p2.start()
112        self.m2.return_value = '14.1.0'
113        self.m3 = self.p3.start()
114        self.m3.return_value = True
115
116    def tearDown(self):
117        self.p2.stop()
118        self.p3.stop()
119
120    def test_update_settings(self, *args):
121        set_module_args(dict(
122            config_sync_ip="10.1.30.1",
123            mirror_primary_address="10.1.30.1",
124            unicast_failover=[
125                dict(
126                    address="10.1.30.1"
127                )
128            ],
129            provider=dict(
130                server='localhost',
131                password='password',
132                user='admin'
133            )
134        ))
135
136        # Configure the parameters that would be returned by querying the
137        # remote device
138        current = ApiParameters(params=load_fixture('load_tm_cm_device_default.json'))
139
140        module = AnsibleModule(
141            argument_spec=self.spec.argument_spec,
142            supports_check_mode=self.spec.supports_check_mode,
143            required_together=self.spec.required_together
144        )
145        mm = ModuleManager(module=module)
146
147        # Override methods to force specific logic in the module to happen
148        mm.update_on_device = Mock(return_value=True)
149        mm.read_current_from_device = Mock(return_value=current)
150
151        results = mm.exec_module()
152
153        assert results['changed'] is True
154        assert results['config_sync_ip'] == '10.1.30.1'
155        assert results['mirror_primary_address'] == '10.1.30.1'
156        assert len(results.keys()) == 4
157
158    def test_set_primary_mirror_address_none(self, *args):
159        set_module_args(dict(
160            mirror_primary_address="none",
161            provider=dict(
162                server='localhost',
163                password='password',
164                user='admin'
165            )
166        ))
167
168        # Configure the parameters that would be returned by querying the
169        # remote device
170        current = ApiParameters(params=load_fixture('load_tm_cm_device.json'))
171
172        module = AnsibleModule(
173            argument_spec=self.spec.argument_spec,
174            supports_check_mode=self.spec.supports_check_mode,
175            required_together=self.spec.required_together
176        )
177        mm = ModuleManager(module=module)
178
179        # Override methods to force specific logic in the module to happen
180        mm.update_on_device = Mock(return_value=True)
181        mm.read_current_from_device = Mock(return_value=current)
182
183        results = mm.exec_module()
184
185        assert results['changed'] is True
186        assert results['mirror_primary_address'] == 'none'
187        assert len(results.keys()) == 2
188
189    def test_set_secondary_mirror_address_none(self, *args):
190        set_module_args(dict(
191            mirror_secondary_address="none",
192            provider=dict(
193                server='localhost',
194                password='password',
195                user='admin'
196            )
197        ))
198
199        # Configure the parameters that would be returned by querying the
200        # remote device
201        current = ApiParameters(params=load_fixture('load_tm_cm_device.json'))
202
203        module = AnsibleModule(
204            argument_spec=self.spec.argument_spec,
205            supports_check_mode=self.spec.supports_check_mode,
206            required_together=self.spec.required_together
207        )
208        mm = ModuleManager(module=module)
209
210        # Override methods to force specific logic in the module to happen
211        mm.update_on_device = Mock(return_value=True)
212        mm.read_current_from_device = Mock(return_value=current)
213
214        results = mm.exec_module()
215
216        assert results['changed'] is True
217        assert results['mirror_secondary_address'] == 'none'
218        assert len(results.keys()) == 2
219
220    def test_set_multicast_address_none(self, *args):
221        set_module_args(dict(
222            multicast_address="none",
223            multicast_port=62960,
224            multicast_interface="eth0",
225            provider=dict(
226                server='localhost',
227                password='password',
228                user='admin'
229            )
230        ))
231
232        # Configure the parameters that would be returned by querying the
233        # remote device
234        current = ApiParameters(params=load_fixture('load_tm_cm_device.json'))
235
236        module = AnsibleModule(
237            argument_spec=self.spec.argument_spec,
238            supports_check_mode=self.spec.supports_check_mode,
239            required_together=self.spec.required_together
240        )
241        mm = ModuleManager(module=module)
242
243        # Override methods to force specific logic in the module to happen
244        mm.update_on_device = Mock(return_value=True)
245        mm.read_current_from_device = Mock(return_value=current)
246
247        results = mm.exec_module()
248
249        assert results['changed'] is True
250        assert results['multicast_address'] == 'none'
251        assert len(results.keys()) == 2
252
253    def test_set_multicast_port_negative(self, *args):
254        set_module_args(dict(
255            multicast_port=-1,
256            multicast_address="224.0.0.245",
257            multicast_interface="eth0",
258            provider=dict(
259                server='localhost',
260                password='password',
261                user='admin'
262            )
263        ))
264
265        # Configure the parameters that would be returned by querying the
266        # remote device
267        current = ApiParameters(params=load_fixture('load_tm_cm_device.json'))
268
269        module = AnsibleModule(
270            argument_spec=self.spec.argument_spec,
271            supports_check_mode=self.spec.supports_check_mode,
272            required_together=self.spec.required_together
273        )
274        mm = ModuleManager(module=module)
275
276        # Override methods to force specific logic in the module to happen
277        mm.update_on_device = Mock(return_value=True)
278        mm.read_current_from_device = Mock(return_value=current)
279
280        with pytest.raises(F5ModuleError) as ex:
281            mm.exec_module()
282
283        assert 'must be between' in str(ex.value)
284
285    def test_set_multicast_address(self, *args):
286        set_module_args(dict(
287            multicast_address="10.1.1.1",
288            multicast_port=62960,
289            multicast_interface="eth0",
290            provider=dict(
291                server='localhost',
292                password='password',
293                user='admin'
294            )
295        ))
296
297        # Configure the parameters that would be returned by querying the
298        # remote device
299        current = ApiParameters(params=load_fixture('load_tm_cm_device.json'))
300
301        module = AnsibleModule(
302            argument_spec=self.spec.argument_spec,
303            supports_check_mode=self.spec.supports_check_mode,
304            required_together=self.spec.required_together
305        )
306        mm = ModuleManager(module=module)
307
308        # Override methods to force specific logic in the module to happen
309        mm.update_on_device = Mock(return_value=True)
310        mm.read_current_from_device = Mock(return_value=current)
311
312        results = mm.exec_module()
313
314        assert results['changed'] is True
315        assert results['multicast_address'] == '10.1.1.1'
316        assert len(results.keys()) == 2
317
318    def test_unset_unicast_failover(self, *args):
319        set_module_args(dict(
320            unicast_failover=[],
321            provider=dict(
322                server='localhost',
323                password='password',
324                user='admin'
325            )
326        ))
327
328        # Configure the parameters that would be returned by querying the
329        # remote device
330        current = ApiParameters(params=load_fixture('load_tm_cm_device.json'))
331
332        module = AnsibleModule(
333            argument_spec=self.spec.argument_spec,
334            supports_check_mode=self.spec.supports_check_mode,
335            required_together=self.spec.required_together
336        )
337        mm = ModuleManager(module=module)
338
339        # Override methods to force specific logic in the module to happen
340        mm.update_on_device = Mock(return_value=True)
341        mm.read_current_from_device = Mock(return_value=current)
342
343        results = mm.exec_module()
344
345        assert results['changed'] is True
346        assert results['unicast_failover'] == 'none'
347        assert len(results.keys()) == 2
348
349    def test_unset_config_sync_ip(self, *args):
350        set_module_args(dict(
351            config_sync_ip="none",
352            provider=dict(
353                server='localhost',
354                password='password',
355                user='admin'
356            )
357        ))
358
359        # Configure the parameters that would be returned by querying the
360        # remote device
361        current = ApiParameters(params=load_fixture('load_tm_cm_device.json'))
362
363        module = AnsibleModule(
364            argument_spec=self.spec.argument_spec,
365            supports_check_mode=self.spec.supports_check_mode,
366            required_together=self.spec.required_together
367        )
368        mm = ModuleManager(module=module)
369
370        # Override methods to force specific logic in the module to happen
371        mm.update_on_device = Mock(return_value=True)
372        mm.read_current_from_device = Mock(return_value=current)
373
374        results = mm.exec_module()
375
376        assert results['changed'] is True
377        assert results['config_sync_ip'] == 'none'
378        assert len(results.keys()) == 2
379