1# -*- coding: utf-8 -*- 2# 3# Copyright: (c) 2018, 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 18from ansible_collections.f5networks.f5_modules.plugins.modules.bigip_asm_policy_manage import ( 19 V1ModuleParameters, V1Manager, ModuleManager, ArgumentSpec 20) 21from ansible_collections.f5networks.f5_modules.plugins.module_utils.common import F5ModuleError 22from ansible_collections.f5networks.f5_modules.tests.unit.modules.utils import set_module_args 23from ansible_collections.f5networks.f5_modules.tests.unit.compat import unittest 24from ansible_collections.f5networks.f5_modules.tests.unit.compat.mock import ( 25 Mock, patch 26) 27 28 29fixture_path = os.path.join(os.path.dirname(__file__), 'fixtures') 30fixture_data = {} 31 32 33def load_fixture(name): 34 path = os.path.join(fixture_path, name) 35 36 if path in fixture_data: 37 return fixture_data[path] 38 39 with open(path) as f: 40 data = f.read() 41 42 try: 43 data = json.loads(data) 44 except Exception: 45 pass 46 47 fixture_data[path] = data 48 return data 49 50 51class TestParameters(unittest.TestCase): 52 def test_module_parameters_template(self): 53 args = dict( 54 name='fake_policy', 55 state='present', 56 template='LotusDomino 6.5 (http)' 57 ) 58 59 p = V1ModuleParameters(params=args) 60 assert p.name == 'fake_policy' 61 assert p.state == 'present' 62 assert p.template == 'POLICY_TEMPLATE_LOTUSDOMINO_6_5_HTTP' 63 64 65class TestManager(unittest.TestCase): 66 def setUp(self): 67 self.spec = ArgumentSpec() 68 self.policy = os.path.join(fixture_path, 'fake_policy.xml') 69 self.patcher1 = patch('time.sleep') 70 self.patcher1.start() 71 self.p1 = patch('ansible_collections.f5networks.f5_modules.plugins.modules.bigip_asm_policy_manage.module_provisioned') 72 self.m1 = self.p1.start() 73 self.m1.return_value = True 74 self.p2 = patch('ansible_collections.f5networks.f5_modules.plugins.modules.bigip_asm_policy_manage.tmos_version') 75 self.p3 = patch('ansible_collections.f5networks.f5_modules.plugins.modules.bigip_asm_policy_manage.send_teem') 76 self.m2 = self.p2.start() 77 self.m2.return_value = '14.1.0' 78 self.m3 = self.p3.start() 79 self.m3.return_value = True 80 81 def tearDown(self): 82 self.patcher1.stop() 83 self.p1.stop() 84 self.p2.stop() 85 self.p3.stop() 86 87 def test_activate_create_from_template(self, *args): 88 set_module_args(dict( 89 name='fake_policy', 90 template='OWA Exchange 2007 (https)', 91 state='present', 92 active='yes', 93 provider=dict( 94 server='localhost', 95 password='password', 96 user='admin' 97 ) 98 )) 99 100 current = V1ModuleParameters(params=load_fixture('load_asm_policy_inactive.json')) 101 module = AnsibleModule( 102 argument_spec=self.spec.argument_spec, 103 supports_check_mode=self.spec.supports_check_mode 104 ) 105 106 v1 = V1Manager(module=module) 107 v1.exists = Mock(return_value=False) 108 v1.import_to_device = Mock(return_value=True) 109 v1.wait_for_task = Mock(side_effect=[True, True]) 110 v1.read_current_from_device = Mock(return_value=current) 111 v1.apply_on_device = Mock(return_value=True) 112 v1.create_from_template_on_device = Mock(return_value=True) 113 v1._file_is_missing = Mock(return_value=False) 114 115 # Override methods to force specific logic in the module to happen 116 mm = ModuleManager(module=module) 117 mm.version_is_less_than_13 = Mock(return_value=False) 118 mm.get_manager = Mock(return_value=v1) 119 120 results = mm.exec_module() 121 122 assert results['changed'] is True 123 assert results['name'] == 'fake_policy' 124 assert results['template'] == 'OWA Exchange 2007 (https)' 125 assert results['active'] == 'yes' 126 127 def test_activate_create_by_name(self, *args): 128 set_module_args(dict( 129 name='fake_policy', 130 state='present', 131 active='yes', 132 provider=dict( 133 server='localhost', 134 password='password', 135 user='admin' 136 ) 137 )) 138 139 current = V1ModuleParameters(params=load_fixture('load_asm_policy_inactive.json')) 140 module = AnsibleModule( 141 argument_spec=self.spec.argument_spec, 142 supports_check_mode=self.spec.supports_check_mode 143 ) 144 145 v1 = V1Manager(module=module) 146 v1.exists = Mock(return_value=False) 147 v1.import_to_device = Mock(return_value=True) 148 v1.wait_for_task = Mock(side_effect=[True, True]) 149 v1.create_on_device = Mock(return_value=True) 150 v1.create_blank = Mock(return_value=True) 151 v1.read_current_from_device = Mock(return_value=current) 152 v1.apply_on_device = Mock(return_value=True) 153 v1._file_is_missing = Mock(return_value=False) 154 155 # Override methods to force specific logic in the module to happen 156 mm = ModuleManager(module=module) 157 mm.version_is_less_than_13 = Mock(return_value=False) 158 mm.get_manager = Mock(return_value=v1) 159 160 results = mm.exec_module() 161 162 assert results['changed'] is True 163 assert results['name'] == 'fake_policy' 164 assert results['active'] == 'yes' 165 166 def test_activate_policy_exists_inactive(self, *args): 167 set_module_args(dict( 168 name='fake_policy', 169 state='present', 170 active='yes', 171 provider=dict( 172 server='localhost', 173 password='password', 174 user='admin' 175 ) 176 )) 177 178 current = V1ModuleParameters(params=load_fixture('load_asm_policy_inactive.json')) 179 module = AnsibleModule( 180 argument_spec=self.spec.argument_spec, 181 supports_check_mode=self.spec.supports_check_mode 182 ) 183 184 v1 = V1Manager(module=module) 185 v1.exists = Mock(return_value=True) 186 v1.update_on_device = Mock(return_value=True) 187 v1.wait_for_task = Mock(side_effect=[True, True]) 188 v1.read_current_from_device = Mock(return_value=current) 189 v1.apply_on_device = Mock(return_value=True) 190 191 # Override methods to force specific logic in the module to happen 192 mm = ModuleManager(module=module) 193 mm.version_is_less_than_13 = Mock(return_value=False) 194 mm.get_manager = Mock(return_value=v1) 195 196 results = mm.exec_module() 197 198 assert results['changed'] is True 199 assert results['active'] == 'yes' 200 201 def test_activate_policy_exists_active(self, *args): 202 set_module_args(dict( 203 name='fake_policy', 204 state='present', 205 active='yes', 206 provider=dict( 207 server='localhost', 208 password='password', 209 user='admin' 210 ) 211 )) 212 213 current = V1ModuleParameters(params=load_fixture('load_asm_policy_active.json')) 214 module = AnsibleModule( 215 argument_spec=self.spec.argument_spec, 216 supports_check_mode=self.spec.supports_check_mode 217 ) 218 219 # Override methods to force specific logic in the module to happen 220 v1 = V1Manager(module=module) 221 v1.exists = Mock(return_value=True) 222 v1.read_current_from_device = Mock(return_value=current) 223 224 # Override methods to force specific logic in the module to happen 225 mm = ModuleManager(module=module) 226 mm.version_is_less_than_13 = Mock(return_value=False) 227 mm.get_manager = Mock(return_value=v1) 228 229 results = mm.exec_module() 230 231 assert results['changed'] is False 232 233 def test_deactivate_policy_exists_active(self, *args): 234 set_module_args(dict( 235 name='fake_policy', 236 state='present', 237 active='no', 238 provider=dict( 239 server='localhost', 240 password='password', 241 user='admin' 242 ) 243 )) 244 245 current = V1ModuleParameters(params=load_fixture('load_asm_policy_active.json')) 246 module = AnsibleModule( 247 argument_spec=self.spec.argument_spec, 248 supports_check_mode=self.spec.supports_check_mode 249 ) 250 251 # Override methods to force specific logic in the module to happen 252 v1 = V1Manager(module=module) 253 v1.exists = Mock(return_value=True) 254 v1.read_current_from_device = Mock(return_value=current) 255 v1.update_on_device = Mock(return_value=True) 256 257 # Override methods to force specific logic in the module to happen 258 mm = ModuleManager(module=module) 259 mm.version_is_less_than_13 = Mock(return_value=False) 260 mm.get_manager = Mock(return_value=v1) 261 262 results = mm.exec_module() 263 264 assert results['changed'] is True 265 266 def test_deactivate_policy_exists_inactive(self, *args): 267 set_module_args(dict( 268 name='fake_policy', 269 state='present', 270 active='no', 271 provider=dict( 272 server='localhost', 273 password='password', 274 user='admin' 275 ) 276 )) 277 278 current = V1ModuleParameters(params=load_fixture('load_asm_policy_inactive.json')) 279 module = AnsibleModule( 280 argument_spec=self.spec.argument_spec, 281 supports_check_mode=self.spec.supports_check_mode 282 ) 283 284 # Override methods to force specific logic in the module to happen 285 v1 = V1Manager(module=module) 286 v1.exists = Mock(return_value=True) 287 v1.read_current_from_device = Mock(return_value=current) 288 289 # Override methods to force specific logic in the module to happen 290 mm = ModuleManager(module=module) 291 mm.version_is_less_than_13 = Mock(return_value=False) 292 mm.get_manager = Mock(return_value=v1) 293 294 results = mm.exec_module() 295 296 assert results['changed'] is False 297 298 def test_create_from_template(self, *args): 299 set_module_args(dict( 300 name='fake_policy', 301 template='LotusDomino 6.5 (http)', 302 state='present', 303 provider=dict( 304 server='localhost', 305 password='password', 306 user='admin' 307 ) 308 )) 309 310 current = V1ModuleParameters(params=load_fixture('load_asm_policy_inactive.json')) 311 module = AnsibleModule( 312 argument_spec=self.spec.argument_spec, 313 supports_check_mode=self.spec.supports_check_mode 314 ) 315 316 # Override methods to force specific logic in the module to happen 317 v1 = V1Manager(module=module) 318 v1.exists = Mock(return_value=False) 319 v1.create_from_template_on_device = Mock(return_value=True) 320 v1.wait_for_task = Mock(side_effect=[True, True]) 321 v1.read_current_from_device = Mock(return_value=current) 322 v1._file_is_missing = Mock(return_value=False) 323 324 # Override methods to force specific logic in the module to happen 325 mm = ModuleManager(module=module) 326 mm.version_is_less_than_13 = Mock(return_value=False) 327 mm.get_manager = Mock(return_value=v1) 328 329 results = mm.exec_module() 330 331 assert results['changed'] is True 332 assert results['name'] == 'fake_policy' 333 assert results['template'] == 'LotusDomino 6.5 (http)' 334 335 def test_create_by_name(self, *args): 336 set_module_args(dict( 337 name='fake_policy', 338 state='present', 339 provider=dict( 340 server='localhost', 341 password='password', 342 user='admin' 343 ) 344 )) 345 346 current = V1ModuleParameters(params=load_fixture('load_asm_policy_inactive.json')) 347 module = AnsibleModule( 348 argument_spec=self.spec.argument_spec, 349 supports_check_mode=self.spec.supports_check_mode 350 ) 351 352 v1 = V1Manager(module=module) 353 v1.exists = Mock(return_value=False) 354 v1.import_to_device = Mock(return_value=True) 355 v1.wait_for_task = Mock(side_effect=[True, True]) 356 v1.create_on_device = Mock(return_value=True) 357 v1.create_blank = Mock(return_value=True) 358 v1.read_current_from_device = Mock(return_value=current) 359 v1.apply_on_device = Mock(return_value=True) 360 v1._file_is_missing = Mock(return_value=False) 361 362 # Override methods to force specific logic in the module to happen 363 mm = ModuleManager(module=module) 364 mm.version_is_less_than_13 = Mock(return_value=False) 365 mm.get_manager = Mock(return_value=v1) 366 367 results = mm.exec_module() 368 369 assert results['changed'] is True 370 assert results['name'] == 'fake_policy' 371 372 def test_delete_policy(self, *args): 373 set_module_args(dict( 374 name='fake_policy', 375 state='absent', 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 ) 387 388 # Override methods to force specific logic in the module to happen 389 v1 = V1Manager(module=module) 390 v1.exists = Mock(side_effect=[True, False]) 391 v1.remove_from_device = Mock(return_value=True) 392 393 # Override methods to force specific logic in the module to happen 394 mm = ModuleManager(module=module) 395 mm.version_is_less_than_13 = Mock(return_value=False) 396 mm.get_manager = Mock(return_value=v1) 397 398 results = mm.exec_module() 399 400 assert results['changed'] is True 401 402 def test_activate_policy_raises(self, *args): 403 set_module_args(dict( 404 name='fake_policy', 405 state='present', 406 active='yes', 407 provider=dict( 408 server='localhost', 409 password='password', 410 user='admin' 411 ) 412 )) 413 414 current = V1ModuleParameters(params=load_fixture('load_asm_policy_inactive.json')) 415 module = AnsibleModule( 416 argument_spec=self.spec.argument_spec, 417 supports_check_mode=self.spec.supports_check_mode 418 ) 419 420 msg = 'Apply policy task failed.' 421 # Override methods to force specific logic in the module to happen 422 v1 = V1Manager(module=module) 423 v1.exists = Mock(return_value=True) 424 v1.wait_for_task = Mock(return_value=False) 425 v1.update_on_device = Mock(return_value=True) 426 v1.read_current_from_device = Mock(return_value=current) 427 v1.apply_on_device = Mock(return_value=True) 428 429 # Override methods to force specific logic in the module to happen 430 mm = ModuleManager(module=module) 431 mm.version_is_less_than_13 = Mock(return_value=False) 432 mm.get_manager = Mock(return_value=v1) 433 434 with pytest.raises(F5ModuleError) as err: 435 mm.exec_module() 436 assert str(err.value) == msg 437 438 def test_create_policy_raises(self, *args): 439 set_module_args(dict( 440 name='fake_policy', 441 state='present', 442 provider=dict( 443 server='localhost', 444 password='password', 445 user='admin' 446 ) 447 )) 448 449 module = AnsibleModule( 450 argument_spec=self.spec.argument_spec, 451 supports_check_mode=self.spec.supports_check_mode 452 ) 453 454 msg = 'Failed to create ASM policy: fake_policy' 455 # Override methods to force specific logic in the module to happen 456 v1 = V1Manager(module=module) 457 v1.exists = Mock(return_value=False) 458 v1.create_on_device = Mock(return_value=False) 459 v1._file_is_missing = Mock(return_value=False) 460 461 # Override methods to force specific logic in the module to happen 462 mm = ModuleManager(module=module) 463 mm.version_is_less_than_13 = Mock(return_value=False) 464 mm.get_manager = Mock(return_value=v1) 465 466 with pytest.raises(F5ModuleError) as err: 467 mm.exec_module() 468 assert str(err.value) == msg 469 470 def test_delete_policy_raises(self, *args): 471 set_module_args(dict( 472 name='fake_policy', 473 state='absent', 474 provider=dict( 475 server='localhost', 476 password='password', 477 user='admin' 478 ) 479 )) 480 481 module = AnsibleModule( 482 argument_spec=self.spec.argument_spec, 483 supports_check_mode=self.spec.supports_check_mode 484 ) 485 msg = 'Failed to delete ASM policy: fake_policy' 486 # Override methods to force specific logic in the module to happen 487 v1 = V1Manager(module=module) 488 v1.exists = Mock(side_effect=[True, True]) 489 v1.remove_from_device = Mock(return_value=True) 490 491 # Override methods to force specific logic in the module to happen 492 mm = ModuleManager(module=module) 493 mm.version_is_less_than_13 = Mock(return_value=False) 494 mm.get_manager = Mock(return_value=v1) 495 496 with pytest.raises(F5ModuleError) as err: 497 mm.exec_module() 498 assert str(err.value) == msg 499