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_device_connectivity import ( 20 ApiParameters, ModuleParameters, 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 27fixture_path = os.path.join(os.path.dirname(__file__), 'fixtures') 28fixture_data = {} 29 30 31def load_fixture(name): 32 path = os.path.join(fixture_path, name) 33 34 if path in fixture_data: 35 return fixture_data[path] 36 37 with open(path) as f: 38 data = f.read() 39 40 try: 41 data = json.loads(data) 42 except Exception: 43 pass 44 45 fixture_data[path] = data 46 return data 47 48 49class TestParameters(unittest.TestCase): 50 def test_module_parameters(self): 51 args = dict( 52 multicast_port='1010', 53 multicast_address='10.10.10.10', 54 multicast_interface='eth0', 55 failover_multicast=True, 56 unicast_failover=[ 57 dict( 58 address='20.20.20.20', 59 port='1234' 60 ) 61 ], 62 mirror_primary_address='1.2.3.4', 63 mirror_secondary_address='5.6.7.8', 64 config_sync_ip='4.3.2.1', 65 state='present', 66 ) 67 p = ModuleParameters(params=args) 68 assert p.multicast_port == 1010 69 assert p.multicast_address == '10.10.10.10' 70 assert p.multicast_interface == 'eth0' 71 assert p.failover_multicast is True 72 assert p.mirror_primary_address == '1.2.3.4' 73 assert p.mirror_secondary_address == '5.6.7.8' 74 assert p.config_sync_ip == '4.3.2.1' 75 assert len(p.unicast_failover) == 1 76 assert 'effectiveIp' in p.unicast_failover[0] 77 assert 'effectivePort' in p.unicast_failover[0] 78 assert 'port' in p.unicast_failover[0] 79 assert 'ip' in p.unicast_failover[0] 80 assert p.unicast_failover[0]['effectiveIp'] == '20.20.20.20' 81 assert p.unicast_failover[0]['ip'] == '20.20.20.20' 82 assert p.unicast_failover[0]['port'] == 1234 83 assert p.unicast_failover[0]['effectivePort'] == 1234 84 85 def test_api_parameters(self): 86 params = load_fixture('load_tm_cm_device.json') 87 p = ApiParameters(params=params) 88 assert p.multicast_port == 62960 89 assert p.multicast_address == '224.0.0.245' 90 assert p.multicast_interface == 'eth0' 91 assert p.mirror_primary_address == '10.2.2.2' 92 assert p.mirror_secondary_address == '10.2.3.2' 93 assert p.config_sync_ip == '10.2.2.2' 94 assert len(p.unicast_failover) == 2 95 assert 'effectiveIp' in p.unicast_failover[0] 96 assert 'effectivePort' in p.unicast_failover[0] 97 assert 'port' in p.unicast_failover[0] 98 assert 'ip' in p.unicast_failover[0] 99 assert p.unicast_failover[0]['effectiveIp'] == 'management-ip' 100 assert p.unicast_failover[0]['ip'] == 'management-ip' 101 assert p.unicast_failover[0]['port'] == 1026 102 assert p.unicast_failover[0]['effectivePort'] == 1026 103 104 105class TestManager(unittest.TestCase): 106 107 def setUp(self): 108 self.spec = ArgumentSpec() 109 self.p2 = patch('ansible_collections.f5networks.f5_modules.plugins.modules.bigip_device_connectivity.tmos_version') 110 self.p3 = patch('ansible_collections.f5networks.f5_modules.plugins.modules.bigip_device_connectivity.send_teem') 111 self.m2 = self.p2.start() 112 self.m2.return_value = '14.1.0' 113 self.m3 = self.p3.start() 114 self.m3.return_value = True 115 116 def tearDown(self): 117 self.p2.stop() 118 self.p3.stop() 119 120 def test_update_settings(self, *args): 121 set_module_args(dict( 122 config_sync_ip="10.1.30.1", 123 mirror_primary_address="10.1.30.1", 124 unicast_failover=[ 125 dict( 126 address="10.1.30.1" 127 ) 128 ], 129 provider=dict( 130 server='localhost', 131 password='password', 132 user='admin' 133 ) 134 )) 135 136 # Configure the parameters that would be returned by querying the 137 # remote device 138 current = ApiParameters(params=load_fixture('load_tm_cm_device_default.json')) 139 140 module = AnsibleModule( 141 argument_spec=self.spec.argument_spec, 142 supports_check_mode=self.spec.supports_check_mode, 143 required_together=self.spec.required_together 144 ) 145 mm = ModuleManager(module=module) 146 147 # Override methods to force specific logic in the module to happen 148 mm.update_on_device = Mock(return_value=True) 149 mm.read_current_from_device = Mock(return_value=current) 150 151 results = mm.exec_module() 152 153 assert results['changed'] is True 154 assert results['config_sync_ip'] == '10.1.30.1' 155 assert results['mirror_primary_address'] == '10.1.30.1' 156 assert len(results.keys()) == 4 157 158 def test_set_primary_mirror_address_none(self, *args): 159 set_module_args(dict( 160 mirror_primary_address="none", 161 provider=dict( 162 server='localhost', 163 password='password', 164 user='admin' 165 ) 166 )) 167 168 # Configure the parameters that would be returned by querying the 169 # remote device 170 current = ApiParameters(params=load_fixture('load_tm_cm_device.json')) 171 172 module = AnsibleModule( 173 argument_spec=self.spec.argument_spec, 174 supports_check_mode=self.spec.supports_check_mode, 175 required_together=self.spec.required_together 176 ) 177 mm = ModuleManager(module=module) 178 179 # Override methods to force specific logic in the module to happen 180 mm.update_on_device = Mock(return_value=True) 181 mm.read_current_from_device = Mock(return_value=current) 182 183 results = mm.exec_module() 184 185 assert results['changed'] is True 186 assert results['mirror_primary_address'] == 'none' 187 assert len(results.keys()) == 2 188 189 def test_set_secondary_mirror_address_none(self, *args): 190 set_module_args(dict( 191 mirror_secondary_address="none", 192 provider=dict( 193 server='localhost', 194 password='password', 195 user='admin' 196 ) 197 )) 198 199 # Configure the parameters that would be returned by querying the 200 # remote device 201 current = ApiParameters(params=load_fixture('load_tm_cm_device.json')) 202 203 module = AnsibleModule( 204 argument_spec=self.spec.argument_spec, 205 supports_check_mode=self.spec.supports_check_mode, 206 required_together=self.spec.required_together 207 ) 208 mm = ModuleManager(module=module) 209 210 # Override methods to force specific logic in the module to happen 211 mm.update_on_device = Mock(return_value=True) 212 mm.read_current_from_device = Mock(return_value=current) 213 214 results = mm.exec_module() 215 216 assert results['changed'] is True 217 assert results['mirror_secondary_address'] == 'none' 218 assert len(results.keys()) == 2 219 220 def test_set_multicast_address_none(self, *args): 221 set_module_args(dict( 222 multicast_address="none", 223 multicast_port=62960, 224 multicast_interface="eth0", 225 provider=dict( 226 server='localhost', 227 password='password', 228 user='admin' 229 ) 230 )) 231 232 # Configure the parameters that would be returned by querying the 233 # remote device 234 current = ApiParameters(params=load_fixture('load_tm_cm_device.json')) 235 236 module = AnsibleModule( 237 argument_spec=self.spec.argument_spec, 238 supports_check_mode=self.spec.supports_check_mode, 239 required_together=self.spec.required_together 240 ) 241 mm = ModuleManager(module=module) 242 243 # Override methods to force specific logic in the module to happen 244 mm.update_on_device = Mock(return_value=True) 245 mm.read_current_from_device = Mock(return_value=current) 246 247 results = mm.exec_module() 248 249 assert results['changed'] is True 250 assert results['multicast_address'] == 'none' 251 assert len(results.keys()) == 2 252 253 def test_set_multicast_port_negative(self, *args): 254 set_module_args(dict( 255 multicast_port=-1, 256 multicast_address="224.0.0.245", 257 multicast_interface="eth0", 258 provider=dict( 259 server='localhost', 260 password='password', 261 user='admin' 262 ) 263 )) 264 265 # Configure the parameters that would be returned by querying the 266 # remote device 267 current = ApiParameters(params=load_fixture('load_tm_cm_device.json')) 268 269 module = AnsibleModule( 270 argument_spec=self.spec.argument_spec, 271 supports_check_mode=self.spec.supports_check_mode, 272 required_together=self.spec.required_together 273 ) 274 mm = ModuleManager(module=module) 275 276 # Override methods to force specific logic in the module to happen 277 mm.update_on_device = Mock(return_value=True) 278 mm.read_current_from_device = Mock(return_value=current) 279 280 with pytest.raises(F5ModuleError) as ex: 281 mm.exec_module() 282 283 assert 'must be between' in str(ex.value) 284 285 def test_set_multicast_address(self, *args): 286 set_module_args(dict( 287 multicast_address="10.1.1.1", 288 multicast_port=62960, 289 multicast_interface="eth0", 290 provider=dict( 291 server='localhost', 292 password='password', 293 user='admin' 294 ) 295 )) 296 297 # Configure the parameters that would be returned by querying the 298 # remote device 299 current = ApiParameters(params=load_fixture('load_tm_cm_device.json')) 300 301 module = AnsibleModule( 302 argument_spec=self.spec.argument_spec, 303 supports_check_mode=self.spec.supports_check_mode, 304 required_together=self.spec.required_together 305 ) 306 mm = ModuleManager(module=module) 307 308 # Override methods to force specific logic in the module to happen 309 mm.update_on_device = Mock(return_value=True) 310 mm.read_current_from_device = Mock(return_value=current) 311 312 results = mm.exec_module() 313 314 assert results['changed'] is True 315 assert results['multicast_address'] == '10.1.1.1' 316 assert len(results.keys()) == 2 317 318 def test_unset_unicast_failover(self, *args): 319 set_module_args(dict( 320 unicast_failover=[], 321 provider=dict( 322 server='localhost', 323 password='password', 324 user='admin' 325 ) 326 )) 327 328 # Configure the parameters that would be returned by querying the 329 # remote device 330 current = ApiParameters(params=load_fixture('load_tm_cm_device.json')) 331 332 module = AnsibleModule( 333 argument_spec=self.spec.argument_spec, 334 supports_check_mode=self.spec.supports_check_mode, 335 required_together=self.spec.required_together 336 ) 337 mm = ModuleManager(module=module) 338 339 # Override methods to force specific logic in the module to happen 340 mm.update_on_device = Mock(return_value=True) 341 mm.read_current_from_device = Mock(return_value=current) 342 343 results = mm.exec_module() 344 345 assert results['changed'] is True 346 assert results['unicast_failover'] == 'none' 347 assert len(results.keys()) == 2 348 349 def test_unset_config_sync_ip(self, *args): 350 set_module_args(dict( 351 config_sync_ip="none", 352 provider=dict( 353 server='localhost', 354 password='password', 355 user='admin' 356 ) 357 )) 358 359 # Configure the parameters that would be returned by querying the 360 # remote device 361 current = ApiParameters(params=load_fixture('load_tm_cm_device.json')) 362 363 module = AnsibleModule( 364 argument_spec=self.spec.argument_spec, 365 supports_check_mode=self.spec.supports_check_mode, 366 required_together=self.spec.required_together 367 ) 368 mm = ModuleManager(module=module) 369 370 # Override methods to force specific logic in the module to happen 371 mm.update_on_device = Mock(return_value=True) 372 mm.read_current_from_device = Mock(return_value=current) 373 374 results = mm.exec_module() 375 376 assert results['changed'] is True 377 assert results['config_sync_ip'] == 'none' 378 assert len(results.keys()) == 2 379