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_iapp_service import (
20    Parameters, ApiParameters, ModuleParameters, ModuleManager, ArgumentSpec
21)
22from ansible_collections.f5networks.f5_modules.tests.unit.compat import unittest
23from ansible_collections.f5networks.f5_modules.tests.unit.compat.mock import Mock, patch
24from ansible_collections.f5networks.f5_modules.tests.unit.modules.utils import set_module_args
25
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
51    def test_module_parameters_keys(self):
52        args = load_fixture('create_iapp_service_parameters_f5_http.json')
53        p = ModuleParameters(params=args)
54
55        # Assert the top-level keys
56        assert p.name == 'http_example'
57        assert p.partition == 'Common'
58        assert p.template == '/Common/f5.http'
59        assert p.device_group is None
60        assert p.inheritedTrafficGroup == 'true'
61        assert p.inheritedDevicegroup == 'true'
62        assert p.traffic_group == '/Common/traffic-group-local-only'
63
64    def test_module_parameters_lists(self):
65        args = load_fixture('create_iapp_service_parameters_f5_http.json')
66        p = ModuleParameters(params=args)
67
68        assert 'lists' in p._values
69
70        assert p.lists[0]['name'] == 'irules__irules'
71        assert p.lists[0]['encrypted'] == 'no'
72        assert len(p.lists[0]['value']) == 1
73        assert p.lists[0]['value'][0] == '/Common/lgyft'
74
75        assert p.lists[1]['name'] == 'net__client_vlan'
76        assert p.lists[1]['encrypted'] == 'no'
77        assert len(p.lists[1]['value']) == 1
78        assert p.lists[1]['value'][0] == '/Common/net2'
79
80    def test_module_parameters_tables(self):
81        args = load_fixture('create_iapp_service_parameters_f5_http.json')
82        p = ModuleParameters(params=args)
83
84        assert 'tables' in p._values
85
86        assert 'columnNames' in p.tables[0]
87        assert len(p.tables[0]['columnNames']) == 1
88        assert p.tables[0]['columnNames'][0] == 'name'
89
90        assert 'name' in p.tables[0]
91        assert p.tables[0]['name'] == 'pool__hosts'
92
93        assert 'rows' in p.tables[0]
94        assert len(p.tables[0]['rows']) == 1
95        assert 'row' in p.tables[0]['rows'][0]
96        assert len(p.tables[0]['rows'][0]['row']) == 1
97        assert p.tables[0]['rows'][0]['row'][0] == 'demo.example.com'
98
99        assert len(p.tables[1]['rows']) == 2
100        assert 'row' in p.tables[0]['rows'][0]
101        assert len(p.tables[1]['rows'][0]['row']) == 2
102        assert p.tables[1]['rows'][0]['row'][0] == '10.1.1.1'
103        assert p.tables[1]['rows'][0]['row'][1] == '0'
104        assert p.tables[1]['rows'][1]['row'][0] == '10.1.1.2'
105        assert p.tables[1]['rows'][1]['row'][1] == '0'
106
107    def test_module_parameters_variables(self):
108        args = load_fixture('create_iapp_service_parameters_f5_http.json')
109        p = ModuleParameters(params=args)
110
111        assert 'variables' in p._values
112        assert len(p.variables) == 34
113
114        # Assert one configuration value
115        assert 'name' in p.variables[0]
116        assert 'value' in p.variables[0]
117        assert p.variables[0]['name'] == 'afm__dos_security_profile'
118        assert p.variables[0]['value'] == '/#do_not_use#'
119
120        # Assert a second configuration value
121        assert 'name' in p.variables[1]
122        assert 'value' in p.variables[1]
123        assert p.variables[1]['name'] == 'afm__policy'
124        assert p.variables[1]['value'] == '/#do_not_use#'
125
126    def test_module_strict_updates_from_top_level(self):
127        # Assumes the user did not provide any parameters
128
129        args = dict(
130            strict_updates=True
131        )
132        p = ModuleParameters(params=args)
133        assert p.strict_updates == 'enabled'
134
135        args = dict(
136            strict_updates=False
137        )
138        p = ModuleParameters(params=args)
139        assert p.strict_updates == 'disabled'
140
141    def test_module_strict_updates_override_from_top_level(self):
142        args = dict(
143            strict_updates=True,
144            parameters=dict(
145                strictUpdates='disabled'
146            )
147        )
148        p = ModuleParameters(params=args)
149        assert p.strict_updates == 'enabled'
150
151        args = dict(
152            strict_updates=False,
153            parameters=dict(
154                strictUpdates='enabled'
155            )
156        )
157        p = ModuleParameters(params=args)
158        assert p.strict_updates == 'disabled'
159
160    def test_module_strict_updates_only_parameters(self):
161        args = dict(
162            parameters=dict(
163                strictUpdates='disabled'
164            )
165        )
166        p = ModuleParameters(params=args)
167        assert p.strict_updates == 'disabled'
168
169        args = dict(
170            parameters=dict(
171                strictUpdates='enabled'
172            )
173        )
174        p = ModuleParameters(params=args)
175        assert p.strict_updates == 'enabled'
176
177    def test_api_strict_updates_from_top_level(self):
178        args = dict(
179            strictUpdates='enabled'
180        )
181        p = ApiParameters(params=args)
182        assert p.strict_updates == 'enabled'
183
184        args = dict(
185            strictUpdates='disabled'
186        )
187        p = ApiParameters(params=args)
188        assert p.strict_updates == 'disabled'
189
190    def test_api_parameters_variables(self):
191        args = dict(
192            variables=[
193                dict(
194                    name="client__http_compression",
195                    encrypted="no",
196                    value="/#create_new#"
197                )
198            ]
199        )
200        p = ApiParameters(params=args)
201        assert p.variables[0]['name'] == 'client__http_compression'
202
203    def test_api_parameters_tables(self):
204        args = dict(
205            tables=[
206                {
207                    "name": "pool__members",
208                    "columnNames": [
209                        "addr",
210                        "port",
211                        "connection_limit"
212                    ],
213                    "rows": [
214                        {
215                            "row": [
216                                "12.12.12.12",
217                                "80",
218                                "0"
219                            ]
220                        },
221                        {
222                            "row": [
223                                "13.13.13.13",
224                                "443",
225                                10
226                            ]
227                        }
228                    ]
229                }
230            ]
231        )
232        p = ApiParameters(params=args)
233        assert p.tables[0]['name'] == 'pool__members'
234        assert p.tables[0]['columnNames'] == ['addr', 'port', 'connection_limit']
235        assert len(p.tables[0]['rows']) == 2
236        assert 'row' in p.tables[0]['rows'][0]
237        assert 'row' in p.tables[0]['rows'][1]
238        assert p.tables[0]['rows'][0]['row'] == ['12.12.12.12', '80', '0']
239        assert p.tables[0]['rows'][1]['row'] == ['13.13.13.13', '443', '10']
240
241    def test_api_parameters_device_group(self):
242        args = dict(
243            deviceGroup='none'
244        )
245        p = ApiParameters(params=args)
246        assert p.device_group is None
247
248    def test_api_parameters_inherited_traffic_group(self):
249        args = dict(
250            inheritedTrafficGroup='true'
251        )
252        p = ApiParameters(params=args)
253        assert p.inheritedTrafficGroup == 'true'
254
255    def test_api_parameters_inherited_devicegroup(self):
256        args = dict(
257            inheritedDevicegroup='true'
258        )
259        p = ApiParameters(params=args)
260        assert p.inheritedDevicegroup == 'true'
261
262    def test_api_parameters_traffic_group(self):
263        args = dict(
264            trafficGroup='/Common/traffic-group-local-only'
265        )
266        p = ApiParameters(params=args)
267        assert p.traffic_group == '/Common/traffic-group-local-only'
268
269    def test_module_template_same_partition(self):
270        args = dict(
271            template='foo',
272            partition='bar'
273        )
274        p = ModuleParameters(params=args)
275        assert p.template == '/bar/foo'
276
277    def test_module_template_same_partition_full_path(self):
278        args = dict(
279            template='/bar/foo',
280            partition='bar'
281        )
282        p = ModuleParameters(params=args)
283        assert p.template == '/bar/foo'
284
285    def test_module_template_different_partition_full_path(self):
286        args = dict(
287            template='/Common/foo',
288            partition='bar'
289        )
290        p = ModuleParameters(params=args)
291        assert p.template == '/Common/foo'
292
293
294class TestManager(unittest.TestCase):
295
296    def setUp(self):
297        self.spec = ArgumentSpec()
298        self.p2 = patch('ansible_collections.f5networks.f5_modules.plugins.modules.bigip_iapp_service.tmos_version')
299        self.p3 = patch('ansible_collections.f5networks.f5_modules.plugins.modules.bigip_iapp_service.send_teem')
300        self.m2 = self.p2.start()
301        self.m2.return_value = '14.1.0'
302        self.m3 = self.p3.start()
303        self.m3.return_value = True
304
305    def tearDown(self):
306        self.p2.stop()
307        self.p3.stop()
308
309    def test_create_service(self, *args):
310        parameters = load_fixture('create_iapp_service_parameters_f5_http.json')
311        set_module_args(dict(
312            name='foo',
313            template='f5.http',
314            parameters=parameters,
315            state='present',
316            provider=dict(
317                server='localhost',
318                password='password',
319                user='admin'
320            )
321        ))
322
323        module = AnsibleModule(
324            argument_spec=self.spec.argument_spec,
325            supports_check_mode=self.spec.supports_check_mode
326        )
327        mm = ModuleManager(module=module)
328
329        # Override methods to force specific logic in the module to happen
330        mm.exists = Mock(return_value=False)
331        mm.create_on_device = Mock(return_value=True)
332        mm.template_exists = Mock(return_value=True)
333
334        results = mm.exec_module()
335        assert results['changed'] is True
336
337    def test_update_agent_status_traps(self, *args):
338        parameters = load_fixture('update_iapp_service_parameters_f5_http.json')
339        set_module_args(dict(
340            name='foo',
341            template='f5.http',
342            parameters=parameters,
343            state='present',
344            provider=dict(
345                server='localhost',
346                password='password',
347                user='admin'
348            )
349        ))
350
351        # Configure the parameters that would be returned by querying the
352        # remote device
353        parameters = load_fixture('create_iapp_service_parameters_f5_http.json')
354        current = Parameters(parameters)
355
356        module = AnsibleModule(
357            argument_spec=self.spec.argument_spec,
358            supports_check_mode=self.spec.supports_check_mode
359        )
360        mm = ModuleManager(module=module)
361
362        # Override methods to force specific logic in the module to happen
363        mm.exists = Mock(return_value=True)
364        mm.update_on_device = Mock(return_value=True)
365        mm.read_current_from_device = Mock(return_value=current)
366
367        results = mm.exec_module()
368        assert results['changed'] is True
369