1# -*- coding: utf-8 -*-
2#
3# Copyright: (c) 2020, 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_gtm_dns_listener 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
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            address='10.0.1.0',
54            mask='255.255.255.0',
55            port=56,
56            advertise='no',
57            description='this is description',
58            ip_protocol='tcp'
59        )
60
61        p = ModuleParameters(params=args)
62        assert p.address == '10.0.1.0'
63        assert p.mask == '255.255.255.0'
64        assert p.port == 56
65        assert p.advertise == 'no'
66        assert p.description == 'this is description'
67        assert p.ip_protocol == 'tcp'
68
69    def test_api_parameters(self):
70        args = dict(
71            address='10.0.1.0',
72            mask='255.255.255.0',
73            port=56,
74            advertise='no',
75            description='this is description',
76            ip_protocol='tcp'
77        )
78
79        p = ApiParameters(params=args)
80        assert p.address == '10.0.1.0'
81        assert p.mask == '255.255.255.0'
82        assert p.port == 56
83
84
85class TestManager(unittest.TestCase):
86    def setUp(self):
87        self.spec = ArgumentSpec()
88        self.p2 = patch('ansible_collections.f5networks.f5_modules.plugins.modules.bigip_gtm_dns_listener.tmos_version')
89        self.p3 = patch('ansible_collections.f5networks.f5_modules.plugins.modules.bigip_gtm_dns_listener.send_teem')
90        self.m2 = self.p2.start()
91        self.m2.return_value = '14.1.0'
92        self.m3 = self.p3.start()
93        self.m3.return_value = True
94
95    def tearDown(self):
96        self.p2.stop()
97        self.p3.stop()
98
99    def test_create(self, *args):
100        set_module_args(dict(
101            name='test-dns-listener',
102            address='10.0.1.0',
103            mask='255.255.255.0',
104            port=53,
105            advertise='no',
106            state='present',
107            description='this is description',
108            ip_protocol='tcp',
109            translate_address='yes',
110            translate_port='no',
111            enabled_vlans=[
112                '/Common/vlan1',
113                '/Common/vlan2'
114            ],
115            irules=[
116                '/Common/rule1',
117                '/Common/rule2'
118            ],
119            provider=dict(
120                server='localhost',
121                password='password',
122                user='admin'
123            )
124        ))
125
126        module = AnsibleModule(
127            argument_spec=self.spec.argument_spec,
128            supports_check_mode=self.spec.supports_check_mode
129        )
130
131        # Override methods to force specific logic in the module to happen
132        mm = ModuleManager(module=module)
133        mm.exists = Mock(return_value=False)
134        mm.create_on_device = Mock(return_value=True)
135        results = mm.exec_module()
136
137        assert results['changed'] is True
138        assert results['address'] == '10.0.1.0'
139        assert results['port'] == 53
140        assert results['mask'] == '255.255.255.0'
141        assert results['advertise'] == 'no'
142        assert results['ip_protocol'] == 'tcp'
143        assert results['description'] == 'this is description'
144        assert results['translate_address'] == 'enabled'
145        assert results['translate_port'] == 'disabled'
146        assert results['vlans_enabled'] is True
147        assert results['enabled_vlans'][0] == '/Common/vlan1' and results['vlans'][1] == '/Common/vlan2'
148        assert results['irules'][0] == '/Common/rule1' and results['irules'][1] == '/Common/rule2'
149
150    def test_update(self, *args):
151        set_module_args(dict(
152            name='test-dns-listener',
153            address='10.0.1.0',
154            mask='255.255.255.0',
155            port=53,
156            advertise='no',
157            state='present',
158            description='this is description',
159            ip_protocol='tcp',
160            translate_address='yes',
161            translate_port='no',
162            enabled_vlans=[
163                '/Common/vlan1',
164                '/Common/vlan2'
165            ],
166            irules=[
167                '/Common/rule1',
168                '/Common/rule2'
169            ],
170            provider=dict(
171                server='localhost',
172                password='password',
173                user='admin'
174            )
175        ))
176
177        module = AnsibleModule(
178            argument_spec=self.spec.argument_spec,
179            supports_check_mode=self.spec.supports_check_mode
180        )
181
182        current = ApiParameters(
183            dict(
184                agent_status_traps='disabled'
185            )
186        )
187
188        # Override methods to force specific logic in the module to happen
189        mm = ModuleManager(module=module)
190        mm.exists = Mock(return_value=True)
191        mm.update_on_device = Mock(return_value=True)
192        mm.read_current_from_device = Mock(return_value=current)
193        results = mm.exec_module()
194
195        assert results['changed'] is True
196        assert results['address'] == '10.0.1.0'
197        assert results['port'] == 53
198        assert results['mask'] == '255.255.255.0'
199        assert results['advertise'] == 'no'
200        assert results['ip_protocol'] == 'tcp'
201        assert results['description'] == 'this is description'
202        assert results['translate_address'] == 'enabled'
203        assert results['translate_port'] == 'disabled'
204        assert results['enabled_vlans'][0] == '/Common/vlan1' and results['vlans'][1] == '/Common/vlan2'
205        assert results['irules'][0] == '/Common/rule1' and results['irules'][1] == '/Common/rule2'
206
207    def test_enabled_vlans_all(self, *args):
208        set_module_args(dict(
209            name='test-dns-listener',
210            address='10.0.1.0',
211            mask='255.255.255.0',
212            port=53,
213            advertise='no',
214            state='present',
215            description='this is description',
216            ip_protocol='tcp',
217            translate_address='yes',
218            translate_port='no',
219            enabled_vlans='all',
220            provider=dict(
221                server='localhost',
222                password='password',
223                user='admin'
224            )
225        ))
226
227        module = AnsibleModule(
228            argument_spec=self.spec.argument_spec,
229            supports_check_mode=self.spec.supports_check_mode
230        )
231
232        current = ApiParameters(
233            dict(
234                agent_status_traps='disabled'
235            )
236        )
237
238        # Override methods to force specific logic in the module to happen
239        mm = ModuleManager(module=module)
240        mm.exists = Mock(return_value=True)
241        mm.update_on_device = Mock(return_value=True)
242        mm.read_current_from_device = Mock(return_value=current)
243        results = mm.exec_module()
244
245        assert results['changed'] is True
246        assert results['address'] == '10.0.1.0'
247        assert results['port'] == 53
248        assert results['mask'] == '255.255.255.0'
249        assert results['advertise'] == 'no'
250        assert results['ip_protocol'] == 'tcp'
251        assert results['description'] == 'this is description'
252        assert results['translate_address'] == 'enabled'
253        assert results['translate_port'] == 'disabled'
254        assert results['vlans_enabled'] is True
255        assert 'vlans_disabled' not in results
256
257    def test_disabled_vlans(self, *args):
258        set_module_args(dict(
259            name='test-dns-listener',
260            address='10.0.1.0',
261            mask='255.255.255.0',
262            port=53,
263            advertise='no',
264            state='present',
265            description='this is description',
266            ip_protocol='tcp',
267            translate_address='yes',
268            translate_port='no',
269            disabled_vlans='all',
270            provider=dict(
271                server='localhost',
272                password='password',
273                user='admin'
274            )
275        ))
276
277        module = AnsibleModule(
278            argument_spec=self.spec.argument_spec,
279            supports_check_mode=self.spec.supports_check_mode
280        )
281
282        current = ApiParameters(
283            dict(
284                agent_status_traps='disabled'
285            )
286        )
287
288        # Override methods to force specific logic in the module to happen
289        mm = ModuleManager(module=module)
290        mm.exists = Mock(return_value=True)
291        mm.update_on_device = Mock(return_value=True)
292        mm.read_current_from_device = Mock(return_value=current)
293
294        with pytest.raises(F5ModuleError) as ex:
295            mm.exec_module()
296        assert 'You cannot disable all VLANs. You must name them individually.' in str(ex.value)
297
298    def test_delete(self, *args):
299        set_module_args(dict(
300            name='test-dns-listener',
301            address='192.10.0.2',
302            state='absent',
303            provider=dict(
304                server='localhost',
305                password='password',
306                user='admin'
307            )
308        ))
309
310        module = AnsibleModule(
311            argument_spec=self.spec.argument_spec,
312            supports_check_mode=self.spec.supports_check_mode
313        )
314
315        current = ApiParameters(
316            dict(
317                agent_status_traps='disabled'
318            )
319        )
320
321        # Override methods to force specific logic in the module to happen
322        mm = ModuleManager(module=module)
323        mm.exists = Mock(return_value=False)
324        results = mm.exec_module()
325
326        assert results['changed'] is False
327
328    def test_enable_listener(self, *args):
329        set_module_args(dict(
330            name='test-dns-listener',
331            state='enabled',
332            address='192.10.0.2',
333            provider=dict(
334                server='localhost',
335                password='password',
336                user='admin'
337            )
338        ))
339
340        module = AnsibleModule(
341            argument_spec=self.spec.argument_spec,
342            supports_check_mode=self.spec.supports_check_mode
343        )
344
345        current = ApiParameters(
346            dict(
347                agent_status_traps='disabled'
348            )
349        )
350
351        # Override methods to force specific logic in the module to happen
352        mm = ModuleManager(module=module)
353        mm.exists = Mock(return_value=True)
354        mm.update_on_device = Mock(return_value=True)
355        mm.read_current_from_device = Mock(return_value=current)
356        results = mm.exec_module()
357
358        assert results['changed'] is True
359
360    def test_disable_listener(self, *args):
361        set_module_args(dict(
362            name='test-dns-listener',
363            state='disabled',
364            address='192.10.0.2',
365            provider=dict(
366                server='localhost',
367                password='password',
368                user='admin'
369            )
370        ))
371
372        module = AnsibleModule(
373            argument_spec=self.spec.argument_spec,
374            supports_check_mode=self.spec.supports_check_mode
375        )
376
377        current = ApiParameters(
378            dict(
379                agent_status_traps='disabled'
380            )
381        )
382
383        # Override methods to force specific logic in the module to happen
384        mm = ModuleManager(module=module)
385        mm.exists = Mock(return_value=True)
386        mm.update_on_device = Mock(return_value=True)
387        mm.read_current_from_device = Mock(return_value=current)
388        results = mm.exec_module()
389
390        assert results['changed'] is True
391