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_tcp_echo import Parameters 21 from library.modules.bigip_monitor_tcp_echo import ModuleManager 22 from library.modules.bigip_monitor_tcp_echo 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_tcp_echo import Parameters 33 from ansible.modules.network.f5.bigip_monitor_tcp_echo import ModuleManager 34 from ansible.modules.network.f5.bigip_monitor_tcp_echo 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 ip='10.10.10.10', 73 interval=20, 74 timeout=30, 75 time_until_up=60, 76 partition='Common' 77 ) 78 79 p = Parameters(params=args) 80 assert p.name == 'foo' 81 assert p.parent == '/Common/parent' 82 assert p.ip == '10.10.10.10' 83 assert p.type == 'tcp_echo' 84 assert p.destination == '10.10.10.10' 85 assert p.interval == 20 86 assert p.timeout == 30 87 assert p.time_until_up == 60 88 89 def test_module_parameters_ints_as_strings(self): 90 args = dict( 91 name='foo', 92 parent='parent', 93 ip='10.10.10.10', 94 interval='20', 95 timeout='30', 96 time_until_up='60', 97 partition='Common' 98 ) 99 100 p = Parameters(params=args) 101 assert p.name == 'foo' 102 assert p.parent == '/Common/parent' 103 assert p.ip == '10.10.10.10' 104 assert p.type == 'tcp_echo' 105 assert p.destination == '10.10.10.10' 106 assert p.interval == 20 107 assert p.timeout == 30 108 assert p.time_until_up == 60 109 110 def test_api_parameters(self): 111 args = dict( 112 name='foo', 113 defaultsFrom='/Common/parent', 114 destination='10.10.10.10', 115 interval=20, 116 timeout=30, 117 timeUntilUp=60 118 ) 119 120 p = Parameters(params=args) 121 assert p.name == 'foo' 122 assert p.parent == '/Common/parent' 123 assert p.ip == '10.10.10.10' 124 assert p.type == 'tcp_echo' 125 assert p.destination == '10.10.10.10' 126 assert p.interval == 20 127 assert p.timeout == 30 128 assert p.time_until_up == 60 129 130 131class TestManagerEcho(unittest.TestCase): 132 133 def setUp(self): 134 self.spec = ArgumentSpec() 135 136 def test_create_monitor(self, *args): 137 set_module_args(dict( 138 name='foo', 139 ip='10.10.10.10', 140 interval=20, 141 timeout=30, 142 time_until_up=60, 143 provider=dict( 144 server='localhost', 145 password='password', 146 user='admin' 147 ) 148 )) 149 150 module = AnsibleModule( 151 argument_spec=self.spec.argument_spec, 152 supports_check_mode=self.spec.supports_check_mode 153 ) 154 155 # Override methods in the specific type of manager 156 mm = ModuleManager(module=module) 157 mm.exists = Mock(side_effect=[False, True]) 158 mm.create_on_device = Mock(return_value=True) 159 160 results = mm.exec_module() 161 162 assert results['changed'] is True 163 164 def test_create_monitor_idempotent(self, *args): 165 set_module_args(dict( 166 name='foo', 167 ip='10.10.10.10', 168 interval=20, 169 timeout=30, 170 time_until_up=60, 171 provider=dict( 172 server='localhost', 173 password='password', 174 user='admin' 175 ) 176 )) 177 178 current = Parameters(params=load_fixture('load_ltm_monitor_tcp_echo.json')) 179 module = AnsibleModule( 180 argument_spec=self.spec.argument_spec, 181 supports_check_mode=self.spec.supports_check_mode 182 ) 183 184 # Override methods in the specific type of manager 185 mm = ModuleManager(module=module) 186 mm.exists = Mock(return_value=True) 187 mm.read_current_from_device = Mock(return_value=current) 188 189 results = mm.exec_module() 190 191 assert results['changed'] is False 192 193 def test_update_interval(self, *args): 194 set_module_args(dict( 195 name='foo', 196 interval=10, 197 provider=dict( 198 server='localhost', 199 password='password', 200 user='admin' 201 ) 202 )) 203 204 current = Parameters(params=load_fixture('load_ltm_monitor_tcp_echo.json')) 205 module = AnsibleModule( 206 argument_spec=self.spec.argument_spec, 207 supports_check_mode=self.spec.supports_check_mode 208 ) 209 210 # Override methods in the specific type of manager 211 mm = ModuleManager(module=module) 212 mm.exists = Mock(return_value=True) 213 mm.read_current_from_device = Mock(return_value=current) 214 mm.update_on_device = Mock(return_value=True) 215 216 results = mm.exec_module() 217 218 assert results['changed'] is True 219 assert results['interval'] == 10 220 221 def test_update_interval_larger_than_existing_timeout(self, *args): 222 set_module_args(dict( 223 name='foo', 224 interval=30, 225 provider=dict( 226 server='localhost', 227 password='password', 228 user='admin' 229 ) 230 )) 231 232 current = Parameters(params=load_fixture('load_ltm_monitor_tcp_echo.json')) 233 module = AnsibleModule( 234 argument_spec=self.spec.argument_spec, 235 supports_check_mode=self.spec.supports_check_mode 236 ) 237 238 # Override methods in the specific type of manager 239 mm = ModuleManager(module=module) 240 mm.exists = Mock(return_value=True) 241 mm.read_current_from_device = Mock(return_value=current) 242 mm.update_on_device = Mock(return_value=True) 243 244 with pytest.raises(F5ModuleError) as ex: 245 mm.exec_module() 246 247 assert "must be less than" in str(ex.value) 248 249 def test_update_interval_larger_than_new_timeout(self, *args): 250 set_module_args(dict( 251 name='foo', 252 interval=10, 253 timeout=5, 254 provider=dict( 255 server='localhost', 256 password='password', 257 user='admin' 258 ) 259 )) 260 261 current = Parameters(params=load_fixture('load_ltm_monitor_tcp_echo.json')) 262 module = AnsibleModule( 263 argument_spec=self.spec.argument_spec, 264 supports_check_mode=self.spec.supports_check_mode 265 ) 266 267 # Override methods in the specific type of manager 268 mm = ModuleManager(module=module) 269 mm.exists = Mock(return_value=True) 270 mm.read_current_from_device = Mock(return_value=current) 271 mm.update_on_device = Mock(return_value=True) 272 273 with pytest.raises(F5ModuleError) as ex: 274 mm.exec_module() 275 276 assert "must be less than" in str(ex.value) 277 278 def test_update_timeout(self, *args): 279 set_module_args(dict( 280 name='foo', 281 timeout=300, 282 provider=dict( 283 server='localhost', 284 password='password', 285 user='admin' 286 ) 287 )) 288 289 current = Parameters(params=load_fixture('load_ltm_monitor_tcp_echo.json')) 290 module = AnsibleModule( 291 argument_spec=self.spec.argument_spec, 292 supports_check_mode=self.spec.supports_check_mode 293 ) 294 295 # Override methods in the specific type of manager 296 mm = ModuleManager(module=module) 297 mm.exists = Mock(return_value=True) 298 mm.read_current_from_device = Mock(return_value=current) 299 mm.update_on_device = Mock(return_value=True) 300 301 results = mm.exec_module() 302 assert results['changed'] is True 303 assert results['timeout'] == 300 304 305 def test_update_time_until_up(self, *args): 306 set_module_args(dict( 307 name='foo', 308 time_until_up=300, 309 provider=dict( 310 server='localhost', 311 password='password', 312 user='admin' 313 ) 314 )) 315 316 current = Parameters(params=load_fixture('load_ltm_monitor_tcp_echo.json')) 317 module = AnsibleModule( 318 argument_spec=self.spec.argument_spec, 319 supports_check_mode=self.spec.supports_check_mode 320 ) 321 322 # Override methods in the specific type of manager 323 mm = ModuleManager(module=module) 324 mm.exists = Mock(return_value=True) 325 mm.read_current_from_device = Mock(return_value=current) 326 mm.update_on_device = Mock(return_value=True) 327 328 results = mm.exec_module() 329 330 assert results['changed'] is True 331 assert results['time_until_up'] == 300 332