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
18from ansible.module_utils.six import PY3
19
20from ansible_collections.f5networks.f5_modules.plugins.modules.bigip_irule import (
21    Parameters, ModuleManager, ArgumentSpec, GtmManager, LtmManager
22)
23from ansible_collections.f5networks.f5_modules.tests.unit.compat import unittest
24from ansible_collections.f5networks.f5_modules.tests.unit.compat.mock import Mock, patch, mock_open
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 BigIpObj(object):
51    def __init__(self, **kwargs):
52        self.__dict__.update(kwargs)
53
54
55class TestParameters(unittest.TestCase):
56    def test_module_parameters_ltm(self):
57        content = load_fixture('create_ltm_irule.tcl')
58        args = dict(
59            content=content,
60            module='ltm',
61            name='foo',
62            state='present'
63        )
64        p = Parameters(params=args)
65        assert p.content == content
66
67    def test_module_parameters_gtm(self):
68        content = load_fixture('create_gtm_irule.tcl')
69        args = dict(
70            content=content,
71            module='gtm',
72            name='foo',
73            state='present'
74        )
75        p = Parameters(params=args)
76        assert p.content == content
77
78    def test_api_parameters_ltm(self):
79        content = load_fixture('create_ltm_irule.tcl')
80        args = dict(
81            apiAnonymous=content
82        )
83        p = Parameters(params=args)
84        assert p.content == content
85
86    def test_return_api_params(self):
87        content = load_fixture('create_ltm_irule.tcl')
88        args = dict(
89            content=content,
90            module='ltm',
91            name='foo',
92            state='present'
93        )
94        p = Parameters(params=args)
95        params = p.api_params()
96
97        assert 'apiAnonymous' in params
98
99
100class TestManager(unittest.TestCase):
101
102    def setUp(self):
103        self.spec = ArgumentSpec()
104        self.ltm_irules = []
105        self.gtm_irules = []
106
107        members = load_fixture('load_ltm_irules.json')
108        for item in members:
109            self.ltm_irules.append(BigIpObj(**item))
110
111        members = load_fixture('load_gtm_irules.json')
112        for item in members:
113            self.gtm_irules.append(BigIpObj(**item))
114        self.p2 = patch('ansible_collections.f5networks.f5_modules.plugins.modules.bigip_irule.tmos_version')
115        self.p3 = patch('ansible_collections.f5networks.f5_modules.plugins.modules.bigip_irule.send_teem')
116        self.m2 = self.p2.start()
117        self.m2.return_value = '14.1.0'
118        self.m3 = self.p3.start()
119        self.m3.return_value = True
120
121    def tearDown(self):
122        self.p2.stop()
123        self.p3.stop()
124
125    def test_create_ltm_irule(self, *args):
126        set_module_args(dict(
127            name='foo',
128            module='ltm',
129            content='this is my content',
130            partition='Common',
131            provider=dict(
132                server='localhost',
133                password='password',
134                user='admin'
135            )
136        ))
137
138        module = AnsibleModule(
139            argument_spec=self.spec.argument_spec,
140            supports_check_mode=self.spec.supports_check_mode,
141            mutually_exclusive=self.spec.mutually_exclusive,
142        )
143
144        # Override methods in the specific type of manager
145        tm = LtmManager(module=module, params=module.params)
146        tm.exists = Mock(side_effect=[False, True])
147        tm.create_on_device = Mock(return_value=True)
148
149        # Override methods to force specific logic in the module to happen
150        mm = ModuleManager(module=module)
151        mm.get_manager = Mock(return_value=tm)
152
153        results = mm.exec_module()
154
155        assert results['changed'] is True
156        assert results['content'] == 'this is my content'
157
158    def test_create_gtm_irule(self, *args):
159        set_module_args(dict(
160            name='foo',
161            module='gtm',
162            content='this is my content',
163            partition='Common',
164            provider=dict(
165                server='localhost',
166                password='password',
167                user='admin'
168            )
169        ))
170
171        module = AnsibleModule(
172            argument_spec=self.spec.argument_spec,
173            supports_check_mode=self.spec.supports_check_mode,
174            mutually_exclusive=self.spec.mutually_exclusive,
175        )
176
177        # Override methods in the specific type of manager
178        tm = GtmManager(module=module, params=module.params)
179        tm.exists = Mock(side_effect=[False, True])
180        tm.create_on_device = Mock(return_value=True)
181
182        # Override methods to force specific logic in the module to happen
183        mm = ModuleManager(module=module)
184        mm.get_manager = Mock(return_value=tm)
185
186        results = mm.exec_module()
187
188        assert results['changed'] is True
189        assert results['content'] == 'this is my content'
190
191    def test_create_gtm_irule_src(self, *args):
192        set_module_args(dict(
193            name='foo',
194            module='gtm',
195            src='{0}/create_ltm_irule.tcl'.format(fixture_path),
196            partition='Common',
197            provider=dict(
198                server='localhost',
199                password='password',
200                user='admin'
201            )
202        ))
203
204        module = AnsibleModule(
205            argument_spec=self.spec.argument_spec,
206            supports_check_mode=self.spec.supports_check_mode,
207            mutually_exclusive=self.spec.mutually_exclusive,
208        )
209
210        if PY3:
211            builtins_name = 'builtins'
212        else:
213            builtins_name = '__builtin__'
214
215        with patch(builtins_name + '.open', mock_open(read_data='this is my content'), create=True):
216            # Override methods in the specific type of manager
217            tm = GtmManager(module=module, params=module.params)
218            tm.exists = Mock(side_effect=[False, True])
219            tm.create_on_device = Mock(return_value=True)
220
221            # Override methods to force specific logic in the module to happen
222            mm = ModuleManager(module=module)
223            mm.get_manager = Mock(return_value=tm)
224
225            results = mm.exec_module()
226
227        assert results['changed'] is True
228        assert results['content'] == 'this is my content'
229        assert results['module'] == 'gtm'
230        assert results['src'] == '{0}/create_ltm_irule.tcl'.format(fixture_path)
231        assert len(results.keys()) == 4
232
233    def test_module_mutual_exclusion(self, *args):
234        set_module_args(dict(
235            content='foo',
236            module='ltm',
237            name='foo',
238            state='present',
239            src='/path/to/irules/foo.tcl',
240            partition='Common',
241            provider=dict(
242                server='localhost',
243                password='password',
244                user='admin'
245            )
246        ))
247
248        with patch('ansible.module_utils.basic.AnsibleModule.fail_json', unsafe=True) as mo:
249            AnsibleModule(
250                argument_spec=self.spec.argument_spec,
251                supports_check_mode=self.spec.supports_check_mode,
252                mutually_exclusive=self.spec.mutually_exclusive,
253            )
254            mo.assert_called_once()
255