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 19try: 20 from library.modules.bigip_monitor_udp import Parameters 21 from library.modules.bigip_monitor_udp import ModuleManager 22 from library.modules.bigip_monitor_udp import ArgumentSpec 23 24 from library.module_utils.network.f5.common import F5ModuleError 25 26 # In Ansible 2.8, Ansible changed import paths. 27 from test.units.compat import unittest 28 from test.units.compat.mock import Mock 29 30 from test.units.modules.utils import set_module_args 31except ImportError: 32 from ansible.modules.network.f5.bigip_monitor_udp import Parameters 33 from ansible.modules.network.f5.bigip_monitor_udp import ModuleManager 34 from ansible.modules.network.f5.bigip_monitor_udp import ArgumentSpec 35 36 from ansible.module_utils.network.f5.common import F5ModuleError 37 38 # Ansible 2.8 imports 39 from units.compat import unittest 40 from units.compat.mock import Mock 41 42 from units.modules.utils import set_module_args 43 44 45fixture_path = os.path.join(os.path.dirname(__file__), 'fixtures') 46fixture_data = {} 47 48 49def load_fixture(name): 50 path = os.path.join(fixture_path, name) 51 52 if path in fixture_data: 53 return fixture_data[path] 54 55 with open(path) as f: 56 data = f.read() 57 58 try: 59 data = json.loads(data) 60 except Exception: 61 pass 62 63 fixture_data[path] = data 64 return data 65 66 67class TestParameters(unittest.TestCase): 68 def test_module_parameters(self): 69 args = dict( 70 name='foo', 71 parent='parent', 72 send='this is a send string', 73 receive='this is a receive string', 74 ip='10.10.10.10', 75 port=80, 76 interval=20, 77 timeout=30, 78 time_until_up=60, 79 partition='Common' 80 ) 81 82 p = Parameters(params=args) 83 assert p.name == 'foo' 84 assert p.parent == '/Common/parent' 85 assert p.send == 'this is a send string' 86 assert p.receive == 'this is a receive string' 87 assert p.ip == '10.10.10.10' 88 assert p.type == 'udp' 89 assert p.port == 80 90 assert p.destination == '10.10.10.10:80' 91 assert p.interval == 20 92 assert p.timeout == 30 93 assert p.time_until_up == 60 94 95 def test_module_parameters_ints_as_strings(self): 96 args = dict( 97 name='foo', 98 parent='parent', 99 send='this is a send string', 100 receive='this is a receive string', 101 ip='10.10.10.10', 102 port='80', 103 interval='20', 104 timeout='30', 105 time_until_up='60', 106 partition='Common' 107 ) 108 109 p = Parameters(params=args) 110 assert p.name == 'foo' 111 assert p.parent == '/Common/parent' 112 assert p.send == 'this is a send string' 113 assert p.receive == 'this is a receive string' 114 assert p.ip == '10.10.10.10' 115 assert p.type == 'udp' 116 assert p.port == 80 117 assert p.destination == '10.10.10.10:80' 118 assert p.interval == 20 119 assert p.timeout == 30 120 assert p.time_until_up == 60 121 122 def test_api_parameters(self): 123 args = dict( 124 name='foo', 125 defaultsFrom='/Common/parent', 126 send='this is a send string', 127 recv='this is a receive string', 128 destination='10.10.10.10:80', 129 interval=20, 130 timeout=30, 131 timeUntilUp=60 132 ) 133 134 p = Parameters(params=args) 135 assert p.name == 'foo' 136 assert p.parent == '/Common/parent' 137 assert p.send == 'this is a send string' 138 assert p.receive == 'this is a receive string' 139 assert p.ip == '10.10.10.10' 140 assert p.type == 'udp' 141 assert p.port == 80 142 assert p.destination == '10.10.10.10:80' 143 assert p.interval == 20 144 assert p.timeout == 30 145 assert p.time_until_up == 60 146 147 148class TestManager(unittest.TestCase): 149 150 def setUp(self): 151 self.spec = ArgumentSpec() 152 153 def test_create_monitor(self, *args): 154 set_module_args(dict( 155 name='foo', 156 parent='parent', 157 send='this is a send string', 158 receive='this is a receive string', 159 ip='10.10.10.10', 160 port=80, 161 interval=20, 162 timeout=30, 163 time_until_up=60, 164 partition='Common', 165 provider=dict( 166 server='localhost', 167 password='password', 168 user='admin' 169 ) 170 )) 171 172 module = AnsibleModule( 173 argument_spec=self.spec.argument_spec, 174 supports_check_mode=self.spec.supports_check_mode 175 ) 176 177 # Override methods in the specific type of manager 178 mm = ModuleManager(module=module) 179 mm.exists = Mock(side_effect=[False, True]) 180 mm.create_on_device = Mock(return_value=True) 181 182 results = mm.exec_module() 183 184 assert results['changed'] is True 185 assert results['parent'] == '/Common/parent' 186 187 def test_create_monitor_idempotent(self, *args): 188 set_module_args(dict( 189 name='asdf', 190 parent='udp', 191 send='default send string', 192 receive='hello world', 193 ip='1.1.1.1', 194 port=389, 195 interval=5, 196 timeout=16, 197 time_until_up=0, 198 partition='Common', 199 provider=dict( 200 server='localhost', 201 password='password', 202 user='admin' 203 ) 204 )) 205 206 current = Parameters(params=load_fixture('load_ltm_monitor_udp.json')) 207 module = AnsibleModule( 208 argument_spec=self.spec.argument_spec, 209 supports_check_mode=self.spec.supports_check_mode 210 ) 211 212 # Override methods in the specific type of manager 213 mm = ModuleManager(module=module) 214 mm.exists = Mock(return_value=True) 215 mm.read_current_from_device = Mock(return_value=current) 216 217 results = mm.exec_module() 218 219 assert results['changed'] is False 220 221 def test_update_port(self, *args): 222 set_module_args(dict( 223 name='asdf', 224 port=800, 225 partition='Common', 226 provider=dict( 227 server='localhost', 228 password='password', 229 user='admin' 230 ) 231 )) 232 233 current = Parameters(params=load_fixture('load_ltm_monitor_udp.json')) 234 module = AnsibleModule( 235 argument_spec=self.spec.argument_spec, 236 supports_check_mode=self.spec.supports_check_mode 237 ) 238 239 # Override methods in the specific type of manager 240 mm = ModuleManager(module=module) 241 mm.exists = Mock(return_value=True) 242 mm.read_current_from_device = Mock(return_value=current) 243 mm.update_on_device = Mock(return_value=True) 244 245 results = mm.exec_module() 246 247 assert results['changed'] is True 248 assert results['port'] == 800 249 250 def test_update_interval(self, *args): 251 set_module_args(dict( 252 name='foo', 253 interval=10, 254 partition='Common', 255 provider=dict( 256 server='localhost', 257 password='password', 258 user='admin' 259 ) 260 )) 261 262 current = Parameters(params=load_fixture('load_ltm_monitor_udp.json')) 263 module = AnsibleModule( 264 argument_spec=self.spec.argument_spec, 265 supports_check_mode=self.spec.supports_check_mode 266 ) 267 268 # Override methods in the specific type of manager 269 mm = ModuleManager(module=module) 270 mm.exists = Mock(return_value=True) 271 mm.read_current_from_device = Mock(return_value=current) 272 mm.update_on_device = Mock(return_value=True) 273 274 results = mm.exec_module() 275 276 assert results['changed'] is True 277 assert results['interval'] == 10 278 279 def test_update_interval_larger_than_existing_timeout(self, *args): 280 set_module_args(dict( 281 name='asdf', 282 interval=30, 283 partition='Common', 284 provider=dict( 285 server='localhost', 286 password='password', 287 user='admin' 288 ) 289 )) 290 291 current = Parameters(params=load_fixture('load_ltm_monitor_udp.json')) 292 module = AnsibleModule( 293 argument_spec=self.spec.argument_spec, 294 supports_check_mode=self.spec.supports_check_mode 295 ) 296 297 # Override methods in the specific type of manager 298 mm = ModuleManager(module=module) 299 mm.exists = Mock(return_value=True) 300 mm.read_current_from_device = Mock(return_value=current) 301 mm.update_on_device = Mock(return_value=True) 302 303 with pytest.raises(F5ModuleError) as ex: 304 mm.exec_module() 305 306 assert "must be less than" in str(ex.value) 307 308 def test_update_interval_larger_than_new_timeout(self, *args): 309 set_module_args(dict( 310 name='asdf', 311 interval=10, 312 timeout=5, 313 partition='Common', 314 provider=dict( 315 server='localhost', 316 password='password', 317 user='admin' 318 ) 319 )) 320 321 current = Parameters(params=load_fixture('load_ltm_monitor_udp.json')) 322 module = AnsibleModule( 323 argument_spec=self.spec.argument_spec, 324 supports_check_mode=self.spec.supports_check_mode 325 ) 326 327 # Override methods in the specific type of manager 328 mm = ModuleManager(module=module) 329 mm.exists = Mock(return_value=True) 330 mm.read_current_from_device = Mock(return_value=current) 331 mm.update_on_device = Mock(return_value=True) 332 333 with pytest.raises(F5ModuleError) as ex: 334 mm.exec_module() 335 336 assert "must be less than" in str(ex.value) 337 338 def test_update_send(self, *args): 339 set_module_args(dict( 340 name='asdf', 341 send='this is another send string', 342 partition='Common', 343 provider=dict( 344 server='localhost', 345 password='password', 346 user='admin' 347 ) 348 )) 349 350 current = Parameters(params=load_fixture('load_ltm_monitor_udp.json')) 351 module = AnsibleModule( 352 argument_spec=self.spec.argument_spec, 353 supports_check_mode=self.spec.supports_check_mode 354 ) 355 356 # Override methods in the specific type of manager 357 mm = ModuleManager(module=module) 358 mm.exists = Mock(return_value=True) 359 mm.read_current_from_device = Mock(return_value=current) 360 mm.update_on_device = Mock(return_value=True) 361 362 results = mm.exec_module() 363 364 assert results['changed'] is True 365 assert results['send'] == 'this is another send string' 366 367 def test_update_receive(self, *args): 368 set_module_args(dict( 369 name='asdf', 370 receive='this is another receive string', 371 partition='Common', 372 provider=dict( 373 server='localhost', 374 password='password', 375 user='admin' 376 ) 377 )) 378 379 current = Parameters(params=load_fixture('load_ltm_monitor_udp.json')) 380 module = AnsibleModule( 381 argument_spec=self.spec.argument_spec, 382 supports_check_mode=self.spec.supports_check_mode 383 ) 384 385 # Override methods in the specific type of manager 386 mm = ModuleManager(module=module) 387 mm.exists = Mock(return_value=True) 388 mm.read_current_from_device = Mock(return_value=current) 389 mm.update_on_device = Mock(return_value=True) 390 391 results = mm.exec_module() 392 393 assert results['changed'] is True 394 assert results['receive'] == 'this is another receive string' 395 396 def test_update_timeout(self, *args): 397 set_module_args(dict( 398 name='asdf', 399 timeout=300, 400 partition='Common', 401 provider=dict( 402 server='localhost', 403 password='password', 404 user='admin' 405 ) 406 )) 407 408 current = Parameters(params=load_fixture('load_ltm_monitor_udp.json')) 409 module = AnsibleModule( 410 argument_spec=self.spec.argument_spec, 411 supports_check_mode=self.spec.supports_check_mode 412 ) 413 414 # Override methods in the specific type of manager 415 mm = ModuleManager(module=module) 416 mm.exists = Mock(return_value=True) 417 mm.read_current_from_device = Mock(return_value=current) 418 mm.update_on_device = Mock(return_value=True) 419 420 results = mm.exec_module() 421 422 assert results['changed'] is True 423 assert results['timeout'] == 300 424 425 def test_update_time_until_up(self, *args): 426 set_module_args(dict( 427 name='asdf', 428 time_until_up=300, 429 partition='Common', 430 provider=dict( 431 server='localhost', 432 password='password', 433 user='admin' 434 ) 435 )) 436 437 current = Parameters(params=load_fixture('load_ltm_monitor_udp.json')) 438 module = AnsibleModule( 439 argument_spec=self.spec.argument_spec, 440 supports_check_mode=self.spec.supports_check_mode 441 ) 442 443 # Override methods in the specific type of manager 444 mm = ModuleManager(module=module) 445 mm.exists = Mock(return_value=True) 446 mm.read_current_from_device = Mock(return_value=current) 447 mm.update_on_device = Mock(return_value=True) 448 449 results = mm.exec_module() 450 451 assert results['changed'] is True 452 assert results['time_until_up'] == 300 453