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_pool import ( 20 ApiParameters, ModuleParameters, 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 monitor_type='m_of_n', 54 monitors=['/Common/Fake', '/Common/Fake2'], 55 quorum=1, 56 slow_ramp_time=200, 57 reselect_tries=5, 58 service_down_action='drop' 59 ) 60 61 p = ModuleParameters(params=args) 62 assert p.monitor_type == 'm_of_n' 63 assert p.quorum == 1 64 assert p.monitors == 'min 1 of { /Common/Fake /Common/Fake2 }' 65 assert p.slow_ramp_time == 200 66 assert p.reselect_tries == 5 67 assert p.service_down_action == 'drop' 68 69 def test_api_parameters(self): 70 args = dict( 71 monitor="/Common/Fake and /Common/Fake2 ", 72 slowRampTime=200, 73 reselectTries=5, 74 serviceDownAction='drop' 75 ) 76 77 p = ApiParameters(params=args) 78 assert p.monitors == '/Common/Fake and /Common/Fake2' 79 assert p.slow_ramp_time == 200 80 assert p.reselect_tries == 5 81 assert p.service_down_action == 'drop' 82 83 def test_unknown_module_lb_method(self): 84 args = dict( 85 lb_method='obscure_hyphenated_fake_method', 86 ) 87 with pytest.raises(F5ModuleError): 88 p = ModuleParameters(params=args) 89 assert p.lb_method == 'foo' 90 91 def test_unknown_api_lb_method(self): 92 args = dict( 93 loadBalancingMode='obscure_hypenated_fake_method' 94 ) 95 with pytest.raises(F5ModuleError): 96 p = ApiParameters(params=args) 97 assert p.lb_method == 'foo' 98 99 100class TestManager(unittest.TestCase): 101 def setUp(self): 102 self.spec = ArgumentSpec() 103 self.p2 = patch('ansible_collections.f5networks.f5_modules.plugins.modules.bigip_pool.tmos_version') 104 self.p3 = patch('ansible_collections.f5networks.f5_modules.plugins.modules.bigip_pool.send_teem') 105 self.m2 = self.p2.start() 106 self.m2.return_value = '14.1.0' 107 self.m3 = self.p3.start() 108 self.m3.return_value = True 109 110 def tearDown(self): 111 self.p2.stop() 112 self.p3.stop() 113 114 def test_create_pool(self, *args): 115 set_module_args(dict( 116 pool='fake_pool', 117 description='fakepool', 118 service_down_action='drop', 119 lb_method='round-robin', 120 partition='Common', 121 slow_ramp_time=10, 122 reselect_tries=1, 123 provider=dict( 124 server='localhost', 125 password='password', 126 user='admin' 127 ) 128 )) 129 130 module = AnsibleModule( 131 argument_spec=self.spec.argument_spec, 132 supports_check_mode=self.spec.supports_check_mode, 133 mutually_exclusive=self.spec.mutually_exclusive, 134 required_one_of=self.spec.required_one_of 135 ) 136 137 mm = ModuleManager(module=module) 138 mm.create_on_device = Mock(return_value=True) 139 mm.exists = Mock(return_value=False) 140 141 results = mm.exec_module() 142 143 assert results['changed'] is True 144 assert results['name'] == 'fake_pool' 145 assert results['description'] == 'fakepool' 146 assert results['service_down_action'] == 'drop' 147 assert results['lb_method'] == 'round-robin' 148 assert results['slow_ramp_time'] == 10 149 assert results['reselect_tries'] == 1 150 151 def test_create_pool_monitor_type_missing(self, *args): 152 set_module_args(dict( 153 pool='fake_pool', 154 lb_method='round-robin', 155 partition='Common', 156 monitors=['/Common/tcp', '/Common/http'], 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 mutually_exclusive=self.spec.mutually_exclusive, 168 required_one_of=self.spec.required_one_of 169 ) 170 171 mm = ModuleManager(module=module) 172 mm.create_on_device = Mock(return_value=True) 173 mm.exists = Mock(return_value=False) 174 175 results = mm.exec_module() 176 177 assert results['changed'] is True 178 assert results['name'] == 'fake_pool' 179 assert results['monitors'] == ['/Common/http', '/Common/tcp'] 180 assert results['monitor_type'] == 'and_list' 181 182 def test_create_pool_monitors_missing(self, *args): 183 set_module_args(dict( 184 pool='fake_pool', 185 lb_method='round-robin', 186 partition='Common', 187 monitor_type='and_list', 188 provider=dict( 189 server='localhost', 190 password='password', 191 user='admin' 192 ) 193 )) 194 195 module = AnsibleModule( 196 argument_spec=self.spec.argument_spec, 197 supports_check_mode=self.spec.supports_check_mode, 198 mutually_exclusive=self.spec.mutually_exclusive, 199 required_one_of=self.spec.required_one_of 200 ) 201 202 mm = ModuleManager(module=module) 203 mm.create_on_device = Mock(return_value=True) 204 mm.exists = Mock(return_value=False) 205 206 msg = "The 'monitors' parameter cannot be empty when " \ 207 "'monitor_type' parameter is specified" 208 with pytest.raises(F5ModuleError) as err: 209 mm.exec_module() 210 211 assert str(err.value) == msg 212 213 def test_create_pool_quorum_missing(self, *args): 214 set_module_args(dict( 215 pool='fake_pool', 216 lb_method='round-robin', 217 partition='Common', 218 monitor_type='m_of_n', 219 monitors=['/Common/tcp', '/Common/http'], 220 provider=dict( 221 server='localhost', 222 password='password', 223 user='admin' 224 ) 225 )) 226 227 module = AnsibleModule( 228 argument_spec=self.spec.argument_spec, 229 supports_check_mode=self.spec.supports_check_mode, 230 mutually_exclusive=self.spec.mutually_exclusive, 231 required_one_of=self.spec.required_one_of 232 ) 233 234 mm = ModuleManager(module=module) 235 mm.create_on_device = Mock(return_value=True) 236 mm.exists = Mock(return_value=False) 237 238 msg = "Quorum value must be specified with monitor_type 'm_of_n'." 239 with pytest.raises(F5ModuleError) as err: 240 mm.exec_module() 241 242 assert str(err.value) == msg 243 244 def test_create_pool_monitor_and_list(self, *args): 245 set_module_args(dict( 246 pool='fake_pool', 247 partition='Common', 248 monitor_type='and_list', 249 monitors=['/Common/tcp', '/Common/http'], 250 provider=dict( 251 server='localhost', 252 password='password', 253 user='admin' 254 ) 255 )) 256 257 module = AnsibleModule( 258 argument_spec=self.spec.argument_spec, 259 supports_check_mode=self.spec.supports_check_mode, 260 mutually_exclusive=self.spec.mutually_exclusive, 261 required_one_of=self.spec.required_one_of 262 ) 263 264 mm = ModuleManager(module=module) 265 mm.create_on_device = Mock(return_value=True) 266 mm.exists = Mock(return_value=False) 267 268 results = mm.exec_module() 269 270 assert results['changed'] is True 271 assert results['name'] == 'fake_pool' 272 assert results['monitors'] == ['/Common/http', '/Common/tcp'] 273 assert results['monitor_type'] == 'and_list' 274 275 def test_create_pool_monitor_m_of_n(self, *args): 276 set_module_args(dict( 277 pool='fake_pool', 278 partition='Common', 279 monitor_type='m_of_n', 280 quorum=1, 281 monitors=['/Common/tcp', '/Common/http'], 282 provider=dict( 283 server='localhost', 284 password='password', 285 user='admin' 286 ) 287 )) 288 289 module = AnsibleModule( 290 argument_spec=self.spec.argument_spec, 291 supports_check_mode=self.spec.supports_check_mode, 292 mutually_exclusive=self.spec.mutually_exclusive, 293 required_one_of=self.spec.required_one_of 294 ) 295 296 mm = ModuleManager(module=module) 297 mm.create_on_device = Mock(return_value=True) 298 mm.exists = Mock(return_value=False) 299 300 results = mm.exec_module() 301 302 assert results['changed'] is True 303 assert results['name'] == 'fake_pool' 304 assert results['monitors'] == ['/Common/http', '/Common/tcp'] 305 assert results['monitor_type'] == 'm_of_n' 306 307 def test_update_monitors(self, *args): 308 set_module_args(dict( 309 name='test_pool', 310 partition='Common', 311 monitor_type='and_list', 312 monitors=['/Common/http', '/Common/tcp'], 313 provider=dict( 314 server='localhost', 315 password='password', 316 user='admin' 317 ) 318 )) 319 320 module = AnsibleModule( 321 argument_spec=self.spec.argument_spec, 322 supports_check_mode=self.spec.supports_check_mode, 323 mutually_exclusive=self.spec.mutually_exclusive, 324 required_one_of=self.spec.required_one_of 325 ) 326 327 mm = ModuleManager(module=module) 328 329 current = ApiParameters(params=load_fixture('load_ltm_pool.json')) 330 331 mm.update_on_device = Mock(return_value=True) 332 mm.exists = Mock(return_value=True) 333 mm.read_current_from_device = Mock(return_value=current) 334 335 results = mm.exec_module() 336 337 assert results['changed'] is True 338 assert results['monitor_type'] == 'and_list' 339 340 def test_create_pool_monitor_and_list_no_partition(self, *args): 341 set_module_args(dict( 342 pool='fake_pool', 343 monitor_type='and_list', 344 monitors=['tcp', 'http'], 345 provider=dict( 346 server='localhost', 347 password='password', 348 user='admin' 349 ) 350 )) 351 352 module = AnsibleModule( 353 argument_spec=self.spec.argument_spec, 354 supports_check_mode=self.spec.supports_check_mode, 355 mutually_exclusive=self.spec.mutually_exclusive, 356 required_one_of=self.spec.required_one_of 357 ) 358 359 mm = ModuleManager(module=module) 360 mm.create_on_device = Mock(return_value=True) 361 mm.exists = Mock(return_value=False) 362 363 results = mm.exec_module() 364 365 assert results['changed'] is True 366 assert results['name'] == 'fake_pool' 367 assert results['monitors'] == ['/Common/http', '/Common/tcp'] 368 assert results['monitor_type'] == 'and_list' 369 370 def test_create_pool_monitor_m_of_n_no_partition(self, *args): 371 set_module_args(dict( 372 pool='fake_pool', 373 monitor_type='m_of_n', 374 quorum=1, 375 monitors=['tcp', 'http'], 376 provider=dict( 377 server='localhost', 378 password='password', 379 user='admin' 380 ) 381 )) 382 383 module = AnsibleModule( 384 argument_spec=self.spec.argument_spec, 385 supports_check_mode=self.spec.supports_check_mode, 386 mutually_exclusive=self.spec.mutually_exclusive, 387 required_one_of=self.spec.required_one_of 388 ) 389 390 mm = ModuleManager(module=module) 391 mm.create_on_device = Mock(return_value=True) 392 mm.exists = Mock(return_value=False) 393 394 results = mm.exec_module() 395 396 assert results['changed'] is True 397 assert results['name'] == 'fake_pool' 398 assert results['monitors'] == ['/Common/http', '/Common/tcp'] 399 assert results['monitor_type'] == 'm_of_n' 400 401 def test_create_pool_monitor_and_list_custom_partition(self, *args): 402 set_module_args(dict( 403 pool='fake_pool', 404 partition='Testing', 405 monitor_type='and_list', 406 monitors=['tcp', 'http'], 407 provider=dict( 408 server='localhost', 409 password='password', 410 user='admin' 411 ) 412 )) 413 414 module = AnsibleModule( 415 argument_spec=self.spec.argument_spec, 416 supports_check_mode=self.spec.supports_check_mode, 417 mutually_exclusive=self.spec.mutually_exclusive, 418 required_one_of=self.spec.required_one_of 419 ) 420 421 mm = ModuleManager(module=module) 422 mm.create_on_device = Mock(return_value=True) 423 mm.exists = Mock(return_value=False) 424 425 results = mm.exec_module() 426 427 assert results['changed'] is True 428 assert results['name'] == 'fake_pool' 429 assert results['monitors'] == ['/Testing/http', '/Testing/tcp'] 430 assert results['monitor_type'] == 'and_list' 431 432 def test_create_pool_monitor_m_of_n_custom_partition(self, *args): 433 set_module_args(dict( 434 pool='fake_pool', 435 partition='Testing', 436 monitor_type='m_of_n', 437 quorum=1, 438 monitors=['tcp', 'http'], 439 provider=dict( 440 server='localhost', 441 password='password', 442 user='admin' 443 ) 444 )) 445 446 module = AnsibleModule( 447 argument_spec=self.spec.argument_spec, 448 supports_check_mode=self.spec.supports_check_mode, 449 mutually_exclusive=self.spec.mutually_exclusive, 450 required_one_of=self.spec.required_one_of 451 ) 452 453 mm = ModuleManager(module=module) 454 mm.create_on_device = Mock(return_value=True) 455 mm.exists = Mock(return_value=False) 456 457 results = mm.exec_module() 458 459 assert results['changed'] is True 460 assert results['name'] == 'fake_pool' 461 assert results['monitors'] == ['/Testing/http', '/Testing/tcp'] 462 assert results['monitor_type'] == 'm_of_n' 463 464 def test_create_pool_with_metadata(self, *args): 465 set_module_args(dict( 466 pool='fake_pool', 467 metadata=dict(ansible='2.4'), 468 provider=dict( 469 server='localhost', 470 password='password', 471 user='admin' 472 ) 473 )) 474 475 module = AnsibleModule( 476 argument_spec=self.spec.argument_spec, 477 supports_check_mode=self.spec.supports_check_mode, 478 mutually_exclusive=self.spec.mutually_exclusive, 479 required_one_of=self.spec.required_one_of 480 ) 481 482 mm = ModuleManager(module=module) 483 mm.create_on_device = Mock(return_value=True) 484 mm.exists = Mock(return_value=False) 485 486 results = mm.exec_module() 487 488 assert results['changed'] is True 489 assert results['name'] == 'fake_pool' 490 assert 'metadata' in results 491 assert 'ansible' in results['metadata'] 492 assert results['metadata']['ansible'] == '2.4' 493 494 def test_create_aggregate_pools(self, *args): 495 set_module_args(dict( 496 aggregate=[ 497 dict( 498 pool='fake_pool', 499 description='fakepool', 500 service_down_action='drop', 501 lb_method='round-robin', 502 partition='Common', 503 slow_ramp_time=10, 504 reselect_tries=1, 505 ), 506 dict( 507 pool='fake_pool2', 508 description='fakepool2', 509 service_down_action='drop', 510 lb_method='predictive-node', 511 partition='Common', 512 slow_ramp_time=110, 513 reselect_tries=2, 514 ) 515 ], 516 provider=dict( 517 server='localhost', 518 password='password', 519 user='admin' 520 ) 521 )) 522 523 module = AnsibleModule( 524 argument_spec=self.spec.argument_spec, 525 supports_check_mode=self.spec.supports_check_mode, 526 mutually_exclusive=self.spec.mutually_exclusive, 527 required_one_of=self.spec.required_one_of 528 ) 529 530 mm = ModuleManager(module=module) 531 mm.create_on_device = Mock(return_value=True) 532 mm.exists = Mock(return_value=False) 533 534 results = mm.exec_module() 535 536 assert results['changed'] is True 537