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