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_configsync_action import ( 20 Parameters, ModuleManager, ArgumentSpec 21) 22from ansible_collections.f5networks.f5_modules.tests.unit.compat import unittest 23from ansible_collections.f5networks.f5_modules.tests.unit.compat.mock import Mock, patch 24from ansible_collections.f5networks.f5_modules.tests.unit.modules.utils import set_module_args 25 26fixture_path = os.path.join(os.path.dirname(__file__), 'fixtures') 27fixture_data = {} 28 29 30def load_fixture(name): 31 path = os.path.join(fixture_path, name) 32 33 if path in fixture_data: 34 return fixture_data[path] 35 36 with open(path) as f: 37 data = f.read() 38 39 try: 40 data = json.loads(data) 41 except Exception: 42 pass 43 44 fixture_data[path] = data 45 return data 46 47 48class TestParameters(unittest.TestCase): 49 def test_module_parameters(self): 50 args = dict( 51 sync_device_to_group=True, 52 sync_group_to_device=True, 53 overwrite_config=True, 54 device_group="foo" 55 ) 56 p = Parameters(params=args) 57 assert p.sync_device_to_group is True 58 assert p.sync_group_to_device is True 59 assert p.overwrite_config is True 60 assert p.device_group == 'foo' 61 62 def test_module_parameters_yes_no(self): 63 args = dict( 64 sync_device_to_group='yes', 65 sync_group_to_device='no', 66 overwrite_config='yes', 67 device_group="foo" 68 ) 69 p = Parameters(params=args) 70 assert p.sync_device_to_group == 'yes' 71 assert p.sync_group_to_device == 'no' 72 assert p.overwrite_config == 'yes' 73 assert p.device_group == 'foo' 74 75 76class TestManager(unittest.TestCase): 77 78 def setUp(self): 79 self.spec = ArgumentSpec() 80 self.patcher1 = patch('time.sleep') 81 self.patcher1.start() 82 self.p2 = patch('ansible_collections.f5networks.f5_modules.plugins.modules.bigip_configsync_action.tmos_version') 83 self.p3 = patch('ansible_collections.f5networks.f5_modules.plugins.modules.bigip_configsync_action.send_teem') 84 self.m2 = self.p2.start() 85 self.m2.return_value = '14.1.0' 86 self.m3 = self.p3.start() 87 self.m3.return_value = True 88 89 def tearDown(self): 90 self.patcher1.stop() 91 self.p2.stop() 92 self.p3.stop() 93 94 def test_update_agent_status_traps(self, *args): 95 set_module_args(dict( 96 sync_device_to_group='yes', 97 device_group="foo", 98 provider=dict( 99 server='localhost', 100 password='password', 101 user='admin' 102 ) 103 )) 104 105 module = AnsibleModule( 106 argument_spec=self.spec.argument_spec, 107 supports_check_mode=self.spec.supports_check_mode, 108 mutually_exclusive=self.spec.mutually_exclusive, 109 required_one_of=self.spec.required_one_of 110 ) 111 mm = ModuleManager(module=module) 112 113 # Override methods to force specific logic in the module to happen 114 mm._device_group_exists = Mock(return_value=True) 115 mm._sync_to_group_required = Mock(return_value=False) 116 mm.execute_on_device = Mock(return_value=True) 117 mm.read_current_from_device = Mock(return_value=None) 118 119 mm._get_status_from_resource = Mock() 120 mm._get_status_from_resource.side_effect = [ 121 'Changes Pending', 'Awaiting Initial Sync', 'In Sync' 122 ] 123 124 results = mm.exec_module() 125 126 assert results['changed'] is True 127