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_static_route import (
20    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
26fixture_path = os.path.join(os.path.dirname(__file__), 'fixtures')
27fixture_data = {}
28
29
30def load_fixture(name):
31    path = os.path.join(fixture_path, name)
32
33    if path in fixture_data:
34        return fixture_data[path]
35
36    with open(path) as f:
37        data = f.read()
38
39    try:
40        data = json.loads(data)
41    except Exception:
42        pass
43
44    fixture_data[path] = data
45    return data
46
47
48class TestParameters(unittest.TestCase):
49    def test_module_parameters(self):
50        args = dict(
51            vlan="foo",
52            gateway_address="10.10.10.10"
53        )
54        p = ModuleParameters(params=args)
55        assert p.vlan == '/Common/foo'
56        assert p.gateway_address == '10.10.10.10'
57
58    def test_api_parameters(self):
59        args = dict(
60            tmInterface="foo",
61            gw="10.10.10.10"
62        )
63        p = ApiParameters(params=args)
64        assert p.vlan == 'foo'
65        assert p.gateway_address == '10.10.10.10'
66
67    def test_reject_parameter_types(self):
68        # boolean true
69        args = dict(reject=True)
70        p = ModuleParameters(params=args)
71        assert p.reject is True
72
73        # boolean false
74        args = dict(reject=False)
75        p = ModuleParameters(params=args)
76        assert p.reject is None
77
78        # string
79        args = dict(reject="yes")
80        p = ModuleParameters(params=args)
81        assert p.reject is True
82
83        # integer
84        args = dict(reject=1)
85        p = ModuleParameters(params=args)
86        assert p.reject is True
87
88        # none
89        args = dict(reject=None)
90        p = ModuleParameters(params=args)
91        assert p.reject is None
92
93    def test_destination_parameter_types(self):
94        # cidr address
95        args = dict(
96            destination="10.10.10.10",
97            netmask='32'
98        )
99        p = ModuleParameters(params=args)
100        assert p.destination == '10.10.10.10/32'
101
102        # netmask
103        args = dict(
104            destination="10.10.10.10",
105            netmask="255.255.255.255"
106        )
107        p = ModuleParameters(params=args)
108        assert p.destination == '10.10.10.10/32'
109
110    def test_vlan_with_partition(self):
111        args = dict(
112            vlan="/Common/foo",
113            gateway_address="10.10.10.10"
114        )
115        p = ModuleParameters(params=args)
116        assert p.vlan == '/Common/foo'
117        assert p.gateway_address == '10.10.10.10'
118
119    def test_api_route_domain(self):
120        args = dict(
121            destination="1.1.1.1/32%2"
122        )
123        p = ApiParameters(params=args)
124        assert p.route_domain == 2
125
126        args = dict(
127            destination="2700:bc00:1f10:101::6/64%2"
128        )
129        p = ApiParameters(params=args)
130        assert p.route_domain == 2
131
132
133class TestManager(unittest.TestCase):
134    def setUp(self):
135        self.spec = ArgumentSpec()
136        self.p2 = patch('ansible_collections.f5networks.f5_modules.plugins.modules.bigip_static_route.tmos_version')
137        self.p3 = patch('ansible_collections.f5networks.f5_modules.plugins.modules.bigip_static_route.send_teem')
138        self.m2 = self.p2.start()
139        self.m2.return_value = '14.1.0'
140        self.m3 = self.p3.start()
141        self.m3.return_value = True
142
143    def tearDown(self):
144        self.p2.stop()
145        self.p3.stop()
146
147    def test_create_blackhole(self, *args):
148        set_module_args(dict(
149            name='test-route',
150            state='present',
151            destination='10.10.10.10',
152            netmask='255.255.255.255',
153            reject='yes',
154            provider=dict(
155                server='localhost',
156                password='password',
157                user='admin'
158            )
159        ))
160
161        module = AnsibleModule(
162            argument_spec=self.spec.argument_spec,
163            mutually_exclusive=self.spec.mutually_exclusive,
164            supports_check_mode=self.spec.supports_check_mode
165        )
166        mm = ModuleManager(module=module)
167
168        # Override methods to force specific logic in the module to happen
169        mm.exists = Mock(return_value=False)
170        mm.create_on_device = Mock(return_value=True)
171
172        results = mm.exec_module()
173        assert results['changed'] is True
174
175    def test_create_route_to_pool(self, *args):
176        set_module_args(dict(
177            name='test-route',
178            state='present',
179            destination='10.10.10.10',
180            netmask='255.255.255.255',
181            pool="test-pool",
182            provider=dict(
183                server='localhost',
184                password='password',
185                user='admin'
186            )
187        ))
188
189        module = AnsibleModule(
190            argument_spec=self.spec.argument_spec,
191            mutually_exclusive=self.spec.mutually_exclusive,
192            supports_check_mode=self.spec.supports_check_mode
193        )
194        mm = ModuleManager(module=module)
195
196        # Override methods to force specific logic in the module to happen
197        mm.exists = Mock(return_value=False)
198        mm.create_on_device = Mock(return_value=True)
199        results = mm.exec_module()
200
201        assert results['changed'] is True
202        assert results['pool'] == 'test-pool'
203
204    def test_create_route_to_vlan(self, *args):
205        set_module_args(dict(
206            name='test-route',
207            state='present',
208            destination='10.10.10.10',
209            netmask='255.255.255.255',
210            vlan="test-vlan",
211            provider=dict(
212                server='localhost',
213                password='password',
214                user='admin'
215            )
216        ))
217
218        module = AnsibleModule(
219            argument_spec=self.spec.argument_spec,
220            mutually_exclusive=self.spec.mutually_exclusive,
221            supports_check_mode=self.spec.supports_check_mode
222        )
223        mm = ModuleManager(module=module)
224
225        # Override methods to force specific logic in the module to happen
226        mm.exists = Mock(return_value=False)
227        mm.create_on_device = Mock(return_value=True)
228        results = mm.exec_module()
229
230        assert results['changed'] is True
231        assert results['vlan'] == '/Common/test-vlan'
232
233    def test_update_description(self, *args):
234        set_module_args(dict(
235            name='test-route',
236            state='present',
237            description='foo description',
238            provider=dict(
239                server='localhost',
240                password='password',
241                user='admin'
242            )
243        ))
244
245        module = AnsibleModule(
246            argument_spec=self.spec.argument_spec,
247            mutually_exclusive=self.spec.mutually_exclusive,
248            supports_check_mode=self.spec.supports_check_mode
249        )
250        mm = ModuleManager(module=module)
251
252        # Override methods to force specific logic in the module to happen
253        current = ApiParameters(params=load_fixture('load_net_route_description.json'))
254        mm.exists = Mock(return_value=True)
255        mm.update_on_device = Mock(return_value=True)
256        mm.read_current_from_device = Mock(return_value=current)
257        results = mm.exec_module()
258
259        assert results['changed'] is True
260        assert results['description'] == 'foo description'
261
262    def test_update_description_idempotent(self, *args):
263        set_module_args(dict(
264            name='test-route',
265            state='present',
266            description='asdasd',
267            provider=dict(
268                server='localhost',
269                password='password',
270                user='admin'
271            )
272        ))
273
274        module = AnsibleModule(
275            argument_spec=self.spec.argument_spec,
276            mutually_exclusive=self.spec.mutually_exclusive,
277            supports_check_mode=self.spec.supports_check_mode
278        )
279        mm = ModuleManager(module=module)
280
281        # Override methods to force specific logic in the module to happen
282        current = ApiParameters(params=load_fixture('load_net_route_description.json'))
283        mm.exists = Mock(return_value=True)
284        mm.update_on_device = Mock(return_value=True)
285        mm.read_current_from_device = Mock(return_value=current)
286        results = mm.exec_module()
287
288        # There is no assert for the description, because it should
289        # not have changed
290        assert results['changed'] is False
291
292    def test_delete(self, *args):
293        set_module_args(dict(
294            name='test-route',
295            state='absent',
296            provider=dict(
297                server='localhost',
298                password='password',
299                user='admin'
300            )
301        ))
302
303        module = AnsibleModule(
304            argument_spec=self.spec.argument_spec,
305            mutually_exclusive=self.spec.mutually_exclusive,
306            supports_check_mode=self.spec.supports_check_mode
307        )
308        mm = ModuleManager(module=module)
309
310        # Override methods to force specific logic in the module to happen
311        mm.exists = Mock(side_effect=[True, False])
312        mm.remove_from_device = Mock(return_value=True)
313        results = mm.exec_module()
314
315        assert results['changed'] is True
316        assert 'description' not in results
317
318    def test_invalid_unknown_params(self, *args):
319        set_module_args(dict(
320            name='test-route',
321            state='present',
322            foo="bar",
323            provider=dict(
324                server='localhost',
325                password='password',
326                user='admin'
327            )
328        ))
329        with patch('ansible.module_utils.basic.AnsibleModule.fail_json') as mo:
330            mo.return_value = True
331            AnsibleModule(
332                argument_spec=self.spec.argument_spec,
333                mutually_exclusive=self.spec.mutually_exclusive,
334                supports_check_mode=self.spec.supports_check_mode
335            )
336            assert mo.call_count == 1
337
338    def test_create_with_route_domain(self, *args):
339        set_module_args(dict(
340            name='test-route',
341            state='present',
342            destination='10.10.10.10',
343            netmask='255.255.255.255',
344            route_domain=1,
345            reject='yes',
346            provider=dict(
347                server='localhost',
348                password='password',
349                user='admin'
350            )
351        ))
352
353        module = AnsibleModule(
354            argument_spec=self.spec.argument_spec,
355            mutually_exclusive=self.spec.mutually_exclusive,
356            supports_check_mode=self.spec.supports_check_mode
357        )
358        mm = ModuleManager(module=module)
359
360        # Override methods to force specific logic in the module to happen
361        mm.exists = Mock(return_value=False)
362        mm.create_on_device = Mock(return_value=True)
363
364        results = mm.exec_module()
365        assert results['changed'] is True
366        assert results['route_domain'] == 1
367        assert results['destination'] == '10.10.10.10%1/32'
368