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_monitor_tcp_half_open import ( 20 Parameters, ModuleManager, ArgumentSpec 21) 22from ansible_collections.f5networks.f5_modules.plugins.module_utils.common import F5ModuleError 23from ansible_collections.f5networks.f5_modules.tests.unit.compat import unittest 24from ansible_collections.f5networks.f5_modules.tests.unit.compat.mock import Mock, patch 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 TestParameters(unittest.TestCase): 51 def test_module_parameters(self): 52 args = dict( 53 name='foo', 54 parent='parent', 55 ip='10.10.10.10', 56 port=80, 57 interval=20, 58 timeout=30, 59 time_until_up=60, 60 partition='Common' 61 ) 62 63 p = Parameters(params=args) 64 assert p.name == 'foo' 65 assert p.parent == '/Common/parent' 66 assert p.ip == '10.10.10.10' 67 assert p.port == 80 68 assert p.type == 'tcp_half_open' 69 assert p.destination == '10.10.10.10:80' 70 assert p.interval == 20 71 assert p.timeout == 30 72 assert p.time_until_up == 60 73 74 def test_module_parameters_ints_as_strings(self): 75 args = dict( 76 name='foo', 77 parent='parent', 78 ip='10.10.10.10', 79 port=80, 80 interval='20', 81 timeout='30', 82 time_until_up='60', 83 partition='Common' 84 ) 85 86 p = Parameters(params=args) 87 assert p.name == 'foo' 88 assert p.parent == '/Common/parent' 89 assert p.ip == '10.10.10.10' 90 assert p.port == 80 91 assert p.type == 'tcp_half_open' 92 assert p.destination == '10.10.10.10:80' 93 assert p.interval == 20 94 assert p.timeout == 30 95 assert p.time_until_up == 60 96 97 def test_api_parameters(self): 98 args = dict( 99 name='foo', 100 defaultsFrom='/Common/parent', 101 destination='10.10.10.10:80', 102 interval=20, 103 timeout=30, 104 timeUntilUp=60 105 ) 106 107 p = Parameters(params=args) 108 assert p.name == 'foo' 109 assert p.parent == '/Common/parent' 110 assert p.ip == '10.10.10.10' 111 assert p.port == 80 112 assert p.type == 'tcp_half_open' 113 assert p.destination == '10.10.10.10:80' 114 assert p.interval == 20 115 assert p.timeout == 30 116 assert p.time_until_up == 60 117 118 119class TestManager(unittest.TestCase): 120 def setUp(self): 121 self.spec = ArgumentSpec() 122 self.p2 = patch('ansible_collections.f5networks.f5_modules.plugins.modules.bigip_monitor_tcp_half_open.tmos_version') 123 self.p3 = patch('ansible_collections.f5networks.f5_modules.plugins.modules.bigip_monitor_tcp_half_open.send_teem') 124 self.m2 = self.p2.start() 125 self.m2.return_value = '14.1.0' 126 self.m3 = self.p3.start() 127 self.m3.return_value = True 128 129 def tearDown(self): 130 self.p2.stop() 131 self.p3.stop() 132 133 def test_create_monitor(self, *args): 134 set_module_args(dict( 135 name='foo', 136 ip='10.10.10.10', 137 port=80, 138 interval=20, 139 timeout=30, 140 time_until_up=60, 141 provider=dict( 142 server='localhost', 143 password='password', 144 user='admin' 145 ) 146 )) 147 148 module = AnsibleModule( 149 argument_spec=self.spec.argument_spec, 150 supports_check_mode=self.spec.supports_check_mode 151 ) 152 153 # Override methods in the specific type of manager 154 mm = ModuleManager(module=module) 155 mm.exists = Mock(side_effect=[False, True]) 156 mm.create_on_device = Mock(return_value=True) 157 158 results = mm.exec_module() 159 160 assert results['changed'] is True 161 162 def test_create_monitor_idempotent(self, *args): 163 set_module_args(dict( 164 name='foo', 165 ip='10.10.10.10', 166 port=80, 167 interval=20, 168 timeout=30, 169 time_until_up=60, 170 provider=dict( 171 server='localhost', 172 password='password', 173 user='admin' 174 ) 175 )) 176 177 current = Parameters(params=load_fixture('load_ltm_monitor_tcp_half_open.json')) 178 module = AnsibleModule( 179 argument_spec=self.spec.argument_spec, 180 supports_check_mode=self.spec.supports_check_mode 181 ) 182 183 # Override methods in the specific type of manager 184 mm = ModuleManager(module=module) 185 mm.exists = Mock(return_value=True) 186 mm.read_current_from_device = Mock(return_value=current) 187 188 results = mm.exec_module() 189 190 assert results['changed'] is False 191 192 def test_update_interval(self, *args): 193 set_module_args(dict( 194 name='foo', 195 interval=10, 196 provider=dict( 197 server='localhost', 198 password='password', 199 user='admin' 200 ) 201 )) 202 203 current = Parameters(params=load_fixture('load_ltm_monitor_tcp_half_open.json')) 204 module = AnsibleModule( 205 argument_spec=self.spec.argument_spec, 206 supports_check_mode=self.spec.supports_check_mode 207 ) 208 209 # Override methods in the specific type of manager 210 mm = ModuleManager(module=module) 211 mm.exists = Mock(return_value=True) 212 mm.read_current_from_device = Mock(return_value=current) 213 mm.update_on_device = Mock(return_value=True) 214 215 results = mm.exec_module() 216 217 assert results['changed'] is True 218 assert results['interval'] == 10 219 220 def test_update_interval_larger_than_existing_timeout(self, *args): 221 set_module_args(dict( 222 name='foo', 223 interval=30, 224 provider=dict( 225 server='localhost', 226 password='password', 227 user='admin' 228 ) 229 )) 230 231 current = Parameters(params=load_fixture('load_ltm_monitor_tcp_half_open.json')) 232 module = AnsibleModule( 233 argument_spec=self.spec.argument_spec, 234 supports_check_mode=self.spec.supports_check_mode 235 ) 236 237 # Override methods in the specific type of manager 238 mm = ModuleManager(module=module) 239 mm.exists = Mock(return_value=True) 240 mm.read_current_from_device = Mock(return_value=current) 241 mm.update_on_device = Mock(return_value=True) 242 243 with pytest.raises(F5ModuleError) as ex: 244 mm.exec_module() 245 246 assert "must be less than" in str(ex.value) 247 248 def test_update_interval_larger_than_new_timeout(self, *args): 249 set_module_args(dict( 250 name='foo', 251 interval=10, 252 timeout=5, 253 provider=dict( 254 server='localhost', 255 password='password', 256 user='admin' 257 ) 258 )) 259 260 current = Parameters(params=load_fixture('load_ltm_monitor_tcp_half_open.json')) 261 module = AnsibleModule( 262 argument_spec=self.spec.argument_spec, 263 supports_check_mode=self.spec.supports_check_mode 264 ) 265 266 # Override methods in the specific type of manager 267 mm = ModuleManager(module=module) 268 mm.exists = Mock(return_value=True) 269 mm.read_current_from_device = Mock(return_value=current) 270 mm.update_on_device = Mock(return_value=True) 271 272 with pytest.raises(F5ModuleError) as ex: 273 mm.exec_module() 274 275 assert "must be less than" in str(ex.value) 276 277 def test_update_timeout(self, *args): 278 set_module_args(dict( 279 name='foo', 280 timeout=300, 281 provider=dict( 282 server='localhost', 283 password='password', 284 user='admin' 285 ) 286 )) 287 288 current = Parameters(params=load_fixture('load_ltm_monitor_tcp_half_open.json')) 289 module = AnsibleModule( 290 argument_spec=self.spec.argument_spec, 291 supports_check_mode=self.spec.supports_check_mode 292 ) 293 294 # Override methods in the specific type of manager 295 mm = ModuleManager(module=module) 296 mm.exists = Mock(return_value=True) 297 mm.read_current_from_device = Mock(return_value=current) 298 mm.update_on_device = Mock(return_value=True) 299 300 results = mm.exec_module() 301 assert results['changed'] is True 302 assert results['timeout'] == 300 303 304 def test_update_time_until_up(self, *args): 305 set_module_args(dict( 306 name='foo', 307 time_until_up=300, 308 provider=dict( 309 server='localhost', 310 password='password', 311 user='admin' 312 ) 313 )) 314 315 current = Parameters(params=load_fixture('load_ltm_monitor_tcp_half_open.json')) 316 module = AnsibleModule( 317 argument_spec=self.spec.argument_spec, 318 supports_check_mode=self.spec.supports_check_mode 319 ) 320 321 # Override methods in the specific type of manager 322 mm = ModuleManager(module=module) 323 mm.exists = Mock(return_value=True) 324 mm.read_current_from_device = Mock(return_value=current) 325 mm.update_on_device = Mock(return_value=True) 326 327 results = mm.exec_module() 328 329 assert results['changed'] is True 330 assert results['time_until_up'] == 300 331