1# -*- coding: utf-8 -*-
2#
3# Copyright: (c) 2019, 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_cgnat_lsn_pool 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
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            name='foo',
53            description='foobar',
54            client_conn_limit=300,
55            harpin_mode='yes',
56            icmp_echo='no',
57            inbound_connections='explicit',
58            mode='pba',
59            persistence_mode='address-port',
60            persistence_timeout=180,
61            route_advertisement='yes',
62            pba_block_idle_timeout=50,
63            pba_block_lifetime=1800,
64            pba_block_size=10,
65            pba_client_block_limit=1,
66            pba_zombie_timeout=10,
67            port_range_low=1080,
68            port_range_high=1090,
69            egress_intf_enabled='yes',
70            egress_interfaces=['tunnel1', 'tunnel2'],
71            members=['10.10.10.0/25', '10.10.10.128/25'],
72            backup_members=['12.12.12.0/25', '12.12.12.128/25'],
73            log_profile='foo_profile',
74            log_publisher='baz_publisher',
75            partition='Common'
76        )
77
78        p = ModuleParameters(params=args)
79        assert p.name == 'foo'
80        assert p.description == 'foobar'
81        assert p.client_conn_limit == 300
82        assert p.harpin_mode == 'enabled'
83        assert p.icmp_echo == 'disabled'
84        assert p.inbound_connections == 'explicit'
85        assert p.mode == 'pba'
86        assert p.persistence_mode == 'address-port'
87        assert p.persistence_timeout == 180
88        assert p.route_advertisement == 'enabled'
89        assert p.pba_block_idle_timeout == 50
90        assert p.pba_block_lifetime == 1800
91        assert p.pba_block_size == 10
92        assert p.pba_client_block_limit == 1
93        assert p.pba_zombie_timeout == 10
94        assert p.port_range_low == 1080
95        assert p.port_range_high == 1090
96        assert p.egress_intf_enabled == 'yes'
97        assert '/Common/tunnel1' and '/Common/tunnel2' in p.egress_interfaces
98        assert '12.12.10.0/25' and '12.12.12.128/25' in p.backup_members
99        assert '10.10.10.0/25' and '10.10.10.128/25' in p.members
100        assert p.log_profile == '/Common/foo_profile'
101        assert p.log_publisher == '/Common/baz_publisher'
102
103    def test_api_parameters(self):
104        args = load_fixture('load_cgnat_lsn_pool.json')
105
106        p = ApiParameters(params=args)
107        assert p.name == 'test_pool'
108        assert p.client_conn_limit == 0
109        assert p.harpin_mode == 'disabled'
110        assert p.icmp_echo == 'disabled'
111        assert p.inbound_connections == 'disabled'
112        assert p.mode == 'deterministic'
113        assert p.persistence_mode == 'address'
114        assert p.persistence_timeout == 300
115        assert p.route_advertisement == 'disabled'
116        assert p.pba_block_idle_timeout == 3600
117        assert p.pba_block_lifetime == 0
118        assert p.pba_block_size == 64
119        assert p.pba_client_block_limit == 1
120        assert p.pba_zombie_timeout == 0
121        assert p.port_range_low == 1025
122        assert p.port_range_high == 6954
123        assert p.egress_intf_enabled == 'no'
124        assert '/Common/http-tunnel' in p.egress_interfaces
125        assert '11.11.11.1/32' in p.backup_members
126        assert '101.10.10.0/24' in p.members
127        assert p.log_profile == '/Common/lsn_log_profile'
128        assert p.log_publisher == '/Common/default-ipsec-log-publisher'
129
130
131class TestManager(unittest.TestCase):
132    def setUp(self):
133        self.spec = ArgumentSpec()
134        self.p2 = patch('ansible_collections.f5networks.f5_modules.plugins.modules.bigip_cgnat_lsn_pool.tmos_version')
135        self.p3 = patch('ansible_collections.f5networks.f5_modules.plugins.modules.bigip_cgnat_lsn_pool.send_teem')
136        self.m2 = self.p2.start()
137        self.m2.return_value = '14.1.0'
138        self.m3 = self.p3.start()
139        self.m3.return_value = True
140
141    def tearDown(self):
142        self.p2.stop()
143        self.p3.stop()
144
145    def test_create_LSN_pool(self, *args):
146        set_module_args(dict(
147            name='foo',
148            description='foobar',
149            client_conn_limit=300,
150            harpin_mode='yes',
151            icmp_echo='no',
152            inbound_connections='explicit',
153            mode='pba',
154            persistence_mode='address-port',
155            persistence_timeout=180,
156            route_advertisement='yes',
157            pba_block_idle_timeout=50,
158            pba_block_lifetime=1800,
159            pba_block_size=10,
160            pba_client_block_limit=1,
161            pba_zombie_timeout=10,
162            port_range_low=1080,
163            port_range_high=1090,
164            egress_intf_enabled='yes',
165            egress_interfaces=['tunnel1', 'tunnel2'],
166            members=['10.10.10.0/25', '10.10.10.128/25'],
167            backup_members=['12.12.12.0/25', '12.12.12.128/25'],
168            log_profile='foo_profile',
169            log_publisher='baz_publisher',
170            partition='Common',
171            provider=dict(
172                server='localhost',
173                password='password',
174                user='admin'
175            )
176        ))
177
178        module = AnsibleModule(
179            argument_spec=self.spec.argument_spec,
180            supports_check_mode=self.spec.supports_check_mode,
181            required_together=self.spec.required_together,
182        )
183
184        # Override methods in the specific type of manager
185        mm = ModuleManager(module=module)
186        mm.exists = Mock(return_value=False)
187        mm.create_on_device = Mock(return_value=True)
188
189        results = mm.exec_module()
190
191        assert results['changed'] is True
192        assert results['description'] == 'foobar'
193        assert results['client_conn_limit'] == 300
194        assert results['harpin_mode'] == 'yes'
195        assert results['icmp_echo'] == 'no'
196        assert results['inbound_connections'] == 'explicit'
197        assert results['mode'] == 'pba'
198        assert results['persistence_mode'] == 'address-port'
199        assert results['persistence_timeout'] == 180
200        assert results['route_advertisement'] == 'yes'
201        assert results['pba_block_idle_timeout'] == 50
202        assert results['pba_block_lifetime'] == 1800
203        assert results['pba_block_size'] == 10
204        assert results['pba_client_block_limit'] == 1
205        assert results['pba_zombie_timeout'] == 10
206        assert results['port_range_low'] == 1080
207        assert results['port_range_high'] == 1090
208        assert results['egress_intf_enabled'] == 'yes'
209        assert '/Common/tunnel1' and '/Common/tunnel2' in results['egress_interfaces']
210        assert '12.12.10.0/25' and '12.12.12.128/25' in results['backup_members']
211        assert '10.10.10.0/25' and '10.10.10.128/25' in results['members']
212        assert results['log_profile'] == '/Common/foo_profile'
213        assert results['log_publisher'] == '/Common/baz_publisher'
214
215    def test_update_LSN_pool(self, *args):
216        set_module_args(dict(
217            name='test_pool',
218            description='foobar',
219            mode='napt',
220            members=['15.15.15.0/25'],
221            backup_members='',
222            harpin_mode='yes',
223            route_advertisement='yes',
224            provider=dict(
225                server='localhost',
226                password='password',
227                user='admin'
228            )
229        ))
230        current = ApiParameters(params=load_fixture('load_cgnat_lsn_pool.json'))
231
232        module = AnsibleModule(
233            argument_spec=self.spec.argument_spec,
234            supports_check_mode=self.spec.supports_check_mode,
235            required_together=self.spec.required_together,
236        )
237
238        # Override methods in the specific type of manager
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
244        results = mm.exec_module()
245
246        assert results['changed'] is True
247        assert results['description'] == 'foobar'
248        assert results['mode'] == 'napt'
249        assert results['members'] == ['15.15.15.0/25']
250        assert results['backup_members'] == []
251        assert results['harpin_mode'] == 'yes'
252        assert results['route_advertisement'] == 'yes'
253