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_member import ( 20 ModuleParameters, ApiParameters, NodeApiParameters, 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 pool='my-pool', 52 address='1.2.3.4', 53 fqdn='fqdn.foo.bar', 54 name='my-name', 55 port=2345, 56 connection_limit=100, 57 description='this is a description', 58 rate_limit=70, 59 ratio=20, 60 preserve_node=False, 61 priority_group=10, 62 state='present', 63 partition='Common', 64 fqdn_auto_populate=False, 65 reuse_nodes=False, 66 ) 67 68 p = ModuleParameters(params=args) 69 assert p.name == 'my-name' 70 71 def test_api_parameters(self): 72 args = load_fixture('load_net_node_with_fqdn.json') 73 p = ApiParameters(params=args) 74 assert p.state == 'present' 75 76 77class TestManager(unittest.TestCase): 78 def setUp(self): 79 self.spec = ArgumentSpec() 80 self.p2 = patch('ansible_collections.f5networks.f5_modules.plugins.modules.bigip_pool_member.tmos_version') 81 self.p3 = patch('ansible_collections.f5networks.f5_modules.plugins.modules.bigip_pool_member.send_teem') 82 self.m2 = self.p2.start() 83 self.m2.return_value = '14.1.0' 84 self.m3 = self.p3.start() 85 self.m3.return_value = True 86 87 def tearDown(self): 88 self.p2.stop() 89 self.p3.stop() 90 91 def test_create_reuse_node_with_name(self, *args): 92 # Configure the arguments that would be sent to the Ansible module 93 set_module_args(dict( 94 pool='my-pool', 95 fqdn='foo.bar.com', 96 port=2345, 97 state='present', 98 partition='Common', 99 reuse_nodes=True, 100 provider=dict( 101 password='password', 102 server='localhost', 103 user='admin' 104 ) 105 )) 106 107 current_node = NodeApiParameters(params=load_fixture('load_net_node_with_fqdn.json')) 108 module = AnsibleModule( 109 argument_spec=self.spec.argument_spec, 110 supports_check_mode=self.spec.supports_check_mode, 111 mutually_exclusive=self.spec.mutually_exclusive, 112 required_one_of=self.spec.required_one_of, 113 ) 114 mm = ModuleManager(module=module) 115 116 # Override methods to force specific logic in the module to happen 117 mm.exists = Mock(return_value=False) 118 mm.create_on_device = Mock(return_value=True) 119 mm.read_current_node_from_device = Mock(return_value=current_node) 120 121 results = mm.exec_module() 122 123 assert results['changed'] is True 124 assert results['fqdn_auto_populate'] is True 125 assert results['fqdn'] == 'foo.bar.com' 126 assert results['state'] == 'present' 127 128 def test_create_reuse_node_with_ipv4_address(self, *args): 129 # Configure the arguments that would be sent to the Ansible module 130 set_module_args(dict( 131 pool='my-pool', 132 address='7.3.67.8', 133 port=2345, 134 state='present', 135 partition='Common', 136 reuse_nodes=True, 137 provider=dict( 138 password='password', 139 server='localhost', 140 user='admin' 141 ) 142 )) 143 144 current_node = NodeApiParameters(params=load_fixture('load_net_node_with_ipv4_address.json')) 145 module = AnsibleModule( 146 argument_spec=self.spec.argument_spec, 147 supports_check_mode=self.spec.supports_check_mode, 148 mutually_exclusive=self.spec.mutually_exclusive, 149 required_one_of=self.spec.required_one_of, 150 ) 151 mm = ModuleManager(module=module) 152 153 # Override methods to force specific logic in the module to happen 154 mm.exists = Mock(return_value=False) 155 mm.create_on_device = Mock(return_value=True) 156 mm.read_current_node_from_device = Mock(return_value=current_node) 157 158 results = mm.exec_module() 159 160 assert results['changed'] is True 161 assert results['fqdn_auto_populate'] is False 162 assert results['address'] == '7.3.67.8' 163 assert results['state'] == 'present' 164 165 def test_create_reuse_node_with_fqdn_auto_populate(self, *args): 166 # Configure the arguments that would be sent to the Ansible module 167 set_module_args(dict( 168 pool='my-pool', 169 fqdn='foo.bar.com', 170 port=2345, 171 state='present', 172 partition='Common', 173 reuse_nodes=True, 174 fqdn_auto_populate=False, 175 provider=dict( 176 password='password', 177 server='localhost', 178 user='admin' 179 ) 180 )) 181 182 current_node = NodeApiParameters(params=load_fixture('load_net_node_with_fqdn.json')) 183 module = AnsibleModule( 184 argument_spec=self.spec.argument_spec, 185 supports_check_mode=self.spec.supports_check_mode, 186 mutually_exclusive=self.spec.mutually_exclusive, 187 required_one_of=self.spec.required_one_of, 188 ) 189 mm = ModuleManager(module=module) 190 191 # Override methods to force specific logic in the module to happen 192 mm.exists = Mock(return_value=False) 193 mm.create_on_device = Mock(return_value=True) 194 mm.read_current_node_from_device = Mock(return_value=current_node) 195 196 results = mm.exec_module() 197 198 assert results['changed'] is True 199 assert results['fqdn_auto_populate'] is True 200 assert results['fqdn'] == 'foo.bar.com' 201 assert results['state'] == 'present' 202 203 def test_create_aggregate_pool_members(self, *args): 204 set_module_args(dict( 205 pool='fake_pool', 206 aggregate=[ 207 dict( 208 name='my-name', 209 host="1.1.1.1", 210 port=1234, 211 state='present', 212 partition='Common', 213 reuse_nodes=True, 214 fqdn_auto_populate=False, 215 ), 216 dict( 217 name='my-name2', 218 fqdn='google.com', 219 port=2423, 220 state='present', 221 partition='Common', 222 fqdn_auto_populate=True, 223 reuse_nodes=True, 224 ) 225 ], 226 provider=dict( 227 password='password', 228 server='localhost', 229 user='admin' 230 ) 231 )) 232 233 module = AnsibleModule( 234 argument_spec=self.spec.argument_spec, 235 supports_check_mode=self.spec.supports_check_mode, 236 mutually_exclusive=self.spec.mutually_exclusive, 237 required_one_of=self.spec.required_one_of, 238 ) 239 240 mm = ModuleManager(module=module) 241 mm.create_on_device = Mock(return_value=True) 242 mm.exists = Mock(return_value=False) 243 244 results = mm.exec_module() 245 246 assert results['changed'] is True 247