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_http 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 send='this is a send string', 56 receive='this is a receive string', 57 ip='10.10.10.10', 58 port=80, 59 interval=20, 60 timeout=30, 61 time_until_up=60, 62 partition='Common' 63 ) 64 65 p = Parameters(params=args) 66 assert p.name == 'foo' 67 assert p.parent == '/Common/parent' 68 assert p.send == 'this is a send string' 69 assert p.receive == 'this is a receive string' 70 assert p.ip == '10.10.10.10' 71 assert p.type == 'http' 72 assert p.port == 80 73 assert p.destination == '10.10.10.10:80' 74 assert p.interval == 20 75 assert p.timeout == 30 76 assert p.time_until_up == 60 77 78 def test_module_parameters_ints_as_strings(self): 79 args = dict( 80 name='foo', 81 parent='parent', 82 send='this is a send string', 83 receive='this is a receive string', 84 ip='10.10.10.10', 85 port='80', 86 interval='20', 87 timeout='30', 88 time_until_up='60', 89 partition='Common' 90 ) 91 92 p = Parameters(params=args) 93 assert p.name == 'foo' 94 assert p.parent == '/Common/parent' 95 assert p.send == 'this is a send string' 96 assert p.receive == 'this is a receive string' 97 assert p.ip == '10.10.10.10' 98 assert p.type == 'http' 99 assert p.port == 80 100 assert p.destination == '10.10.10.10:80' 101 assert p.interval == 20 102 assert p.timeout == 30 103 assert p.time_until_up == 60 104 105 def test_api_parameters(self): 106 args = dict( 107 name='foo', 108 defaultsFrom='/Common/parent', 109 send='this is a send string', 110 recv='this is a receive string', 111 destination='10.10.10.10:80', 112 interval=20, 113 timeout=30, 114 timeUntilUp=60 115 ) 116 117 p = Parameters(params=args) 118 assert p.name == 'foo' 119 assert p.parent == '/Common/parent' 120 assert p.send == 'this is a send string' 121 assert p.receive == 'this is a receive string' 122 assert p.ip == '10.10.10.10' 123 assert p.type == 'http' 124 assert p.port == 80 125 assert p.destination == '10.10.10.10:80' 126 assert p.interval == 20 127 assert p.timeout == 30 128 assert p.time_until_up == 60 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_monitor_http.tmos_version') 135 self.p3 = patch('ansible_collections.f5networks.f5_modules.plugins.modules.bigip_monitor_http.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_monitor(self, *args): 146 set_module_args(dict( 147 name='foo', 148 parent='parent', 149 send='this is a send string', 150 receive='this is a receive string', 151 ip='10.10.10.10', 152 port=80, 153 interval=20, 154 timeout=30, 155 time_until_up=60, 156 partition='Common', 157 provider=dict( 158 server='localhost', 159 password='password', 160 user='admin' 161 ) 162 )) 163 164 module = AnsibleModule( 165 argument_spec=self.spec.argument_spec, 166 supports_check_mode=self.spec.supports_check_mode 167 ) 168 169 # Override methods in the specific type of manager 170 mm = ModuleManager(module=module) 171 mm.exists = Mock(side_effect=[False, True]) 172 mm.create_on_device = Mock(return_value=True) 173 174 results = mm.exec_module() 175 176 assert results['changed'] is True 177 assert results['parent'] == '/Common/parent' 178 179 def test_create_monitor_idempotent(self, *args): 180 set_module_args(dict( 181 name='asdf', 182 parent='http', 183 send='GET /\\r\\n', 184 receive='hello world', 185 ip='1.1.1.1', 186 port=389, 187 interval=5, 188 timeout=16, 189 time_until_up=0, 190 partition='Common', 191 provider=dict( 192 server='localhost', 193 password='password', 194 user='admin' 195 ) 196 )) 197 198 current = Parameters(params=load_fixture('load_ltm_monitor_http.json')) 199 module = AnsibleModule( 200 argument_spec=self.spec.argument_spec, 201 supports_check_mode=self.spec.supports_check_mode 202 ) 203 204 # Override methods in the specific type of manager 205 mm = ModuleManager(module=module) 206 mm.exists = Mock(return_value=True) 207 mm.read_current_from_device = Mock(return_value=current) 208 209 results = mm.exec_module() 210 211 assert results['changed'] is False 212 213 def test_update_port(self, *args): 214 set_module_args(dict( 215 name='asdf', 216 port=800, 217 partition='Common', 218 provider=dict( 219 server='localhost', 220 password='password', 221 user='admin' 222 ) 223 )) 224 225 current = Parameters(params=load_fixture('load_ltm_monitor_http.json')) 226 module = AnsibleModule( 227 argument_spec=self.spec.argument_spec, 228 supports_check_mode=self.spec.supports_check_mode 229 ) 230 231 # Override methods in the specific type of manager 232 mm = ModuleManager(module=module) 233 mm.exists = Mock(return_value=True) 234 mm.read_current_from_device = Mock(return_value=current) 235 mm.update_on_device = Mock(return_value=True) 236 237 results = mm.exec_module() 238 239 assert results['changed'] is True 240 assert results['port'] == 800 241 242 def test_update_interval(self, *args): 243 set_module_args(dict( 244 name='foo', 245 interval=10, 246 partition='Common', 247 provider=dict( 248 server='localhost', 249 password='password', 250 user='admin' 251 ) 252 )) 253 254 current = Parameters(params=load_fixture('load_ltm_monitor_http.json')) 255 module = AnsibleModule( 256 argument_spec=self.spec.argument_spec, 257 supports_check_mode=self.spec.supports_check_mode 258 ) 259 260 # Override methods in the specific type of manager 261 mm = ModuleManager(module=module) 262 mm.exists = Mock(return_value=True) 263 mm.read_current_from_device = Mock(return_value=current) 264 mm.update_on_device = Mock(return_value=True) 265 266 results = mm.exec_module() 267 268 assert results['changed'] is True 269 assert results['interval'] == 10 270 271 def test_update_interval_larger_than_existing_timeout(self, *args): 272 set_module_args(dict( 273 name='asdf', 274 interval=30, 275 partition='Common', 276 provider=dict( 277 server='localhost', 278 password='password', 279 user='admin' 280 ) 281 )) 282 283 current = Parameters(params=load_fixture('load_ltm_monitor_http.json')) 284 module = AnsibleModule( 285 argument_spec=self.spec.argument_spec, 286 supports_check_mode=self.spec.supports_check_mode 287 ) 288 289 # Override methods in the specific type of manager 290 mm = ModuleManager(module=module) 291 mm.exists = Mock(return_value=True) 292 mm.read_current_from_device = Mock(return_value=current) 293 mm.update_on_device = Mock(return_value=True) 294 295 with pytest.raises(F5ModuleError) as ex: 296 mm.exec_module() 297 298 assert "must be less than" in str(ex.value) 299 300 def test_update_interval_larger_than_new_timeout(self, *args): 301 set_module_args(dict( 302 name='asdf', 303 interval=10, 304 timeout=5, 305 partition='Common', 306 provider=dict( 307 server='localhost', 308 password='password', 309 user='admin' 310 ) 311 )) 312 313 current = Parameters(params=load_fixture('load_ltm_monitor_http.json')) 314 module = AnsibleModule( 315 argument_spec=self.spec.argument_spec, 316 supports_check_mode=self.spec.supports_check_mode 317 ) 318 319 # Override methods in the specific type of manager 320 mm = ModuleManager(module=module) 321 mm.exists = Mock(return_value=True) 322 mm.read_current_from_device = Mock(return_value=current) 323 mm.update_on_device = Mock(return_value=True) 324 325 with pytest.raises(F5ModuleError) as ex: 326 mm.exec_module() 327 328 assert "must be less than" in str(ex.value) 329 330 def test_update_send(self, *args): 331 set_module_args(dict( 332 name='asdf', 333 send='this is another send string', 334 partition='Common', 335 provider=dict( 336 server='localhost', 337 password='password', 338 user='admin' 339 ) 340 )) 341 342 current = Parameters(params=load_fixture('load_ltm_monitor_http.json')) 343 module = AnsibleModule( 344 argument_spec=self.spec.argument_spec, 345 supports_check_mode=self.spec.supports_check_mode 346 ) 347 348 # Override methods in the specific type of manager 349 mm = ModuleManager(module=module) 350 mm.exists = Mock(return_value=True) 351 mm.read_current_from_device = Mock(return_value=current) 352 mm.update_on_device = Mock(return_value=True) 353 354 results = mm.exec_module() 355 356 assert results['changed'] is True 357 assert results['send'] == 'this is another send string' 358 359 def test_update_receive(self, *args): 360 set_module_args(dict( 361 name='asdf', 362 receive='this is another receive string', 363 partition='Common', 364 provider=dict( 365 server='localhost', 366 password='password', 367 user='admin' 368 ) 369 )) 370 371 current = Parameters(params=load_fixture('load_ltm_monitor_http.json')) 372 module = AnsibleModule( 373 argument_spec=self.spec.argument_spec, 374 supports_check_mode=self.spec.supports_check_mode 375 ) 376 377 # Override methods in the specific type of manager 378 mm = ModuleManager(module=module) 379 mm.exists = Mock(return_value=True) 380 mm.read_current_from_device = Mock(return_value=current) 381 mm.update_on_device = Mock(return_value=True) 382 383 results = mm.exec_module() 384 385 assert results['changed'] is True 386 assert results['receive'] == 'this is another receive string' 387 388 def test_update_timeout(self, *args): 389 set_module_args(dict( 390 name='asdf', 391 timeout=300, 392 partition='Common', 393 provider=dict( 394 server='localhost', 395 password='password', 396 user='admin' 397 ) 398 )) 399 400 current = Parameters(params=load_fixture('load_ltm_monitor_http.json')) 401 module = AnsibleModule( 402 argument_spec=self.spec.argument_spec, 403 supports_check_mode=self.spec.supports_check_mode 404 ) 405 406 # Override methods in the specific type of manager 407 mm = ModuleManager(module=module) 408 mm.exists = Mock(return_value=True) 409 mm.read_current_from_device = Mock(return_value=current) 410 mm.update_on_device = Mock(return_value=True) 411 412 results = mm.exec_module() 413 414 assert results['changed'] is True 415 assert results['timeout'] == 300 416 417 def test_update_time_until_up(self, *args): 418 set_module_args(dict( 419 name='asdf', 420 time_until_up=300, 421 partition='Common', 422 provider=dict( 423 server='localhost', 424 password='password', 425 user='admin' 426 ) 427 )) 428 429 current = Parameters(params=load_fixture('load_ltm_monitor_http.json')) 430 module = AnsibleModule( 431 argument_spec=self.spec.argument_spec, 432 supports_check_mode=self.spec.supports_check_mode 433 ) 434 435 # Override methods in the specific type of manager 436 mm = ModuleManager(module=module) 437 mm.exists = Mock(return_value=True) 438 mm.read_current_from_device = Mock(return_value=current) 439 mm.update_on_device = Mock(return_value=True) 440 441 results = mm.exec_module() 442 443 assert results['changed'] is True 444 assert results['time_until_up'] == 300 445