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