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_snat_translation 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='my-snat-translation', 53 address='1.1.1.1', 54 arp='yes', 55 connection_limit=300, 56 description='None', 57 ip_idle_timeout='50', 58 partition='Common', 59 state='present', 60 traffic_group='test', 61 tcp_idle_timeout='20', 62 udp_idle_timeout='100', 63 ) 64 65 p = ModuleParameters(params=args) 66 assert p.name == 'my-snat-translation' 67 assert p.address == '1.1.1.1' 68 assert p.arp == 'enabled' 69 assert p.connection_limit == 300 70 assert p.description == 'None' 71 assert p.ip_idle_timeout == '50' 72 assert p.partition == 'Common' 73 assert p.state == 'present' 74 assert p.traffic_group == '/Common/test' 75 assert p.tcp_idle_timeout == '20' 76 assert p.udp_idle_timeout == '100' 77 78 def test_api_parameters(self): 79 args = load_fixture('load_ltm_snat_translation_default.json') 80 p = ApiParameters(params=args) 81 assert p.address == '1.1.1.1' 82 assert p.arp == 'no' 83 assert p.connection_limit == 0 84 assert p.description == 'My description' 85 assert p.ip_idle_timeout == '50' 86 assert p.partition == 'Common' 87 assert p.traffic_group == '/Common/test' 88 assert p.tcp_idle_timeout == '20' 89 assert p.udp_idle_timeout == '100' 90 91 def test_module_parameters_arp_yes(self): 92 args = dict( 93 arp='yes' 94 ) 95 p = ModuleParameters(params=args) 96 assert p.arp == 'enabled' 97 98 def test_module_parameters_arp_no(self): 99 args = dict( 100 arp='no' 101 ) 102 p = ModuleParameters(params=args) 103 assert p.arp == 'disabled' 104 105 def test_module_parameters_connection_limit_none(self): 106 args = dict( 107 connection_limit=0 108 ) 109 p = ModuleParameters(params=args) 110 assert p.connection_limit == 0 111 112 def test_module_parameters_connection_limit_int(self): 113 args = dict( 114 connection_limit=500 115 ) 116 p = ModuleParameters(params=args) 117 assert p.connection_limit == 500 118 119 def test_module_parameters_description_none(self): 120 args = dict( 121 description='none' 122 ) 123 p = ModuleParameters(params=args) 124 assert p.description == '' 125 126 def test_module_parameters_description_empty(self): 127 args = dict( 128 description='' 129 ) 130 p = ModuleParameters(params=args) 131 assert p.description == '' 132 133 def test_module_parameters_description_string_value(self): 134 args = dict( 135 description='My Snat Translation' 136 ) 137 p = ModuleParameters(params=args) 138 assert p.description == 'My Snat Translation' 139 140 def test_module_parameters_ip_idle_timeout_indefinite(self): 141 args = dict( 142 ip_idle_timeout='indefinite' 143 ) 144 p = ModuleParameters(params=args) 145 assert p.ip_idle_timeout == 'indefinite' 146 147 def test_module_parameters_ip_idle_timeout_string_value(self): 148 args = dict( 149 ip_idle_timeout='65000' 150 ) 151 p = ModuleParameters(params=args) 152 assert p.ip_idle_timeout == '65000' 153 154 def test_module_no_partition_prefix_parameters(self): 155 args = dict( 156 partition='Common', 157 address='10.10.10.10', 158 traffic_group='traffic-group-1' 159 ) 160 p = ModuleParameters(params=args) 161 assert p.partition == 'Common' 162 assert p.address == '10.10.10.10' 163 assert p.traffic_group == '/Common/traffic-group-1' 164 165 def test_module_partition_prefix_parameters(self): 166 args = dict( 167 partition='Common', 168 address='10.10.10.10', 169 traffic_group='/Common/traffic-group-1' 170 ) 171 p = ModuleParameters(params=args) 172 assert p.partition == 'Common' 173 assert p.address == '10.10.10.10' 174 assert p.traffic_group == '/Common/traffic-group-1' 175 176 def test_module_parameters_state_present(self): 177 args = dict( 178 state='present' 179 ) 180 p = ModuleParameters(params=args) 181 assert p.state == 'present' 182 assert p.enabled is True 183 184 def test_module_parameters_state_absent(self): 185 args = dict( 186 state='absent' 187 ) 188 p = ModuleParameters(params=args) 189 assert p.state == 'absent' 190 191 def test_module_parameters_state_enabled(self): 192 args = dict( 193 state='enabled' 194 ) 195 p = ModuleParameters(params=args) 196 assert p.state == 'enabled' 197 assert p.enabled is True 198 199 def test_module_parameters_state_disabled(self): 200 args = dict( 201 state='disabled' 202 ) 203 p = ModuleParameters(params=args) 204 assert p.state == 'disabled' 205 assert p.disabled is True 206 207 def test_module_parameters_tcp_idle_timeout_indefinite(self): 208 args = dict( 209 tcp_idle_timeout='indefinite' 210 ) 211 p = ModuleParameters(params=args) 212 assert p.tcp_idle_timeout == 'indefinite' 213 214 def test_module_parameters_tcp_idle_timeout_string_value(self): 215 args = dict( 216 tcp_idle_timeout='65000' 217 ) 218 p = ModuleParameters(params=args) 219 assert p.tcp_idle_timeout == '65000' 220 221 def test_module_parameters_udp_idle_timeout_indefinite(self): 222 args = dict( 223 udp_idle_timeout='indefinite' 224 ) 225 p = ModuleParameters(params=args) 226 assert p.udp_idle_timeout == 'indefinite' 227 228 def test_module_parameters_udp_idle_timeout_string_value(self): 229 args = dict( 230 udp_idle_timeout='65000' 231 ) 232 p = ModuleParameters(params=args) 233 assert p.udp_idle_timeout == '65000' 234 235 236class TestManager(unittest.TestCase): 237 def setUp(self): 238 self.spec = ArgumentSpec() 239 self.p2 = patch('ansible_collections.f5networks.f5_modules.plugins.modules.bigip_snat_translation.tmos_version') 240 self.p3 = patch('ansible_collections.f5networks.f5_modules.plugins.modules.bigip_snat_translation.send_teem') 241 self.m2 = self.p2.start() 242 self.m2.return_value = '14.1.0' 243 self.m3 = self.p3.start() 244 self.m3.return_value = True 245 246 def tearDown(self): 247 self.p2.stop() 248 self.p3.stop() 249 250 def test_create_snat_translation(self, *args): 251 set_module_args(dict( 252 name='my-snat-translation', 253 address='1.1.1.1', 254 arp='yes', 255 connection_limit=300, 256 description='My description', 257 ip_idle_timeout='50', 258 state='present', 259 tcp_idle_timeout='20', 260 udp_idle_timeout='100', 261 provider=dict( 262 server='localhost', 263 password='password', 264 user='admin' 265 ) 266 )) 267 268 module = AnsibleModule( 269 argument_spec=self.spec.argument_spec, 270 supports_check_mode=self.spec.supports_check_mode, 271 required_if=self.spec.required_if 272 ) 273 mm = ModuleManager(module=module) 274 275 # Override methods to force specific logic in the module to happen 276 mm.exists = Mock(side_effect=[False, True]) 277 mm.create_on_device = Mock(return_value=True) 278 279 results = mm.exec_module() 280 281 assert results['changed'] is True 282 assert results['address'] == '1.1.1.1' 283 assert results['arp'] == 'yes' 284 assert results['connection_limit'] == 300 285 assert results['description'] == 'My description' 286 assert results['ip_idle_timeout'] == '50' 287 assert results['tcp_idle_timeout'] == '20' 288 assert results['udp_idle_timeout'] == '100' 289 290 def test_update_snat_translation(self, *args): 291 set_module_args(dict( 292 name='my-snat-translation', 293 address='1.1.1.1', 294 arp='yes', 295 connection_limit=300, 296 description='', 297 ip_idle_timeout='500', 298 state='disabled', 299 tcp_idle_timeout='indefinite', 300 traffic_group='traffic-group-1', 301 udp_idle_timeout='indefinite', 302 provider=dict( 303 server='localhost', 304 password='password', 305 user='admin' 306 ) 307 )) 308 current = ApiParameters(params=load_fixture('load_ltm_snat_translation_default.json')) 309 310 module = AnsibleModule( 311 argument_spec=self.spec.argument_spec, 312 supports_check_mode=self.spec.supports_check_mode, 313 required_if=self.spec.required_if 314 ) 315 mm = ModuleManager(module=module) 316 317 # Override methods to force specific logic in the module to happen 318 mm.read_current_from_device = Mock(return_value=current) 319 mm.update_on_device = Mock(return_value=True) 320 mm.exists = Mock(return_value=True) 321 mm.create_on_device = Mock(return_value=True) 322 323 results = mm.exec_module() 324 325 assert results['changed'] is True 326 assert results['arp'] == 'yes' 327 assert results['connection_limit'] == 300 328 assert results['description'] == '' 329 assert results['ip_idle_timeout'] == '500' 330 assert results['tcp_idle_timeout'] == 'indefinite' 331 assert results['udp_idle_timeout'] == 'indefinite' 332