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_monitor_tcp_echo import ( 20 Parameters, 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 27 28fixture_path = os.path.join(os.path.dirname(__file__), 'fixtures') 29fixture_data = {} 30 31 32def load_fixture(name): 33 path = os.path.join(fixture_path, name) 34 35 if path in fixture_data: 36 return fixture_data[path] 37 38 with open(path) as f: 39 data = f.read() 40 41 try: 42 data = json.loads(data) 43 except Exception: 44 pass 45 46 fixture_data[path] = data 47 return data 48 49 50class TestParameters(unittest.TestCase): 51 def test_module_parameters(self): 52 args = dict( 53 name='foo', 54 parent='parent', 55 ip='10.10.10.10', 56 interval=20, 57 timeout=30, 58 time_until_up=60, 59 partition='Common' 60 ) 61 62 p = Parameters(params=args) 63 assert p.name == 'foo' 64 assert p.parent == '/Common/parent' 65 assert p.ip == '10.10.10.10' 66 assert p.type == 'tcp_echo' 67 assert p.destination == '10.10.10.10' 68 assert p.interval == 20 69 assert p.timeout == 30 70 assert p.time_until_up == 60 71 72 def test_module_parameters_ints_as_strings(self): 73 args = dict( 74 name='foo', 75 parent='parent', 76 ip='10.10.10.10', 77 interval='20', 78 timeout='30', 79 time_until_up='60', 80 partition='Common' 81 ) 82 83 p = Parameters(params=args) 84 assert p.name == 'foo' 85 assert p.parent == '/Common/parent' 86 assert p.ip == '10.10.10.10' 87 assert p.type == 'tcp_echo' 88 assert p.destination == '10.10.10.10' 89 assert p.interval == 20 90 assert p.timeout == 30 91 assert p.time_until_up == 60 92 93 def test_api_parameters(self): 94 args = dict( 95 name='foo', 96 defaultsFrom='/Common/parent', 97 destination='10.10.10.10', 98 interval=20, 99 timeout=30, 100 timeUntilUp=60 101 ) 102 103 p = Parameters(params=args) 104 assert p.name == 'foo' 105 assert p.parent == '/Common/parent' 106 assert p.ip == '10.10.10.10' 107 assert p.type == 'tcp_echo' 108 assert p.destination == '10.10.10.10' 109 assert p.interval == 20 110 assert p.timeout == 30 111 assert p.time_until_up == 60 112 113 114class TestManagerEcho(unittest.TestCase): 115 def setUp(self): 116 self.spec = ArgumentSpec() 117 self.p2 = patch('ansible_collections.f5networks.f5_modules.plugins.modules.bigip_monitor_tcp_echo.tmos_version') 118 self.p3 = patch('ansible_collections.f5networks.f5_modules.plugins.modules.bigip_monitor_tcp_echo.send_teem') 119 self.m2 = self.p2.start() 120 self.m2.return_value = '14.1.0' 121 self.m3 = self.p3.start() 122 self.m3.return_value = True 123 124 def tearDown(self): 125 self.p2.stop() 126 self.p3.stop() 127 128 def test_create_monitor(self, *args): 129 set_module_args(dict( 130 name='foo', 131 ip='10.10.10.10', 132 interval=20, 133 timeout=30, 134 time_until_up=60, 135 provider=dict( 136 server='localhost', 137 password='password', 138 user='admin' 139 ) 140 )) 141 142 module = AnsibleModule( 143 argument_spec=self.spec.argument_spec, 144 supports_check_mode=self.spec.supports_check_mode 145 ) 146 147 # Override methods in the specific type of manager 148 mm = ModuleManager(module=module) 149 mm.exists = Mock(side_effect=[False, True]) 150 mm.create_on_device = Mock(return_value=True) 151 152 results = mm.exec_module() 153 154 assert results['changed'] is True 155 156 def test_create_monitor_idempotent(self, *args): 157 set_module_args(dict( 158 name='foo', 159 ip='10.10.10.10', 160 interval=20, 161 timeout=30, 162 time_until_up=60, 163 provider=dict( 164 server='localhost', 165 password='password', 166 user='admin' 167 ) 168 )) 169 170 current = Parameters(params=load_fixture('load_ltm_monitor_tcp_echo.json')) 171 module = AnsibleModule( 172 argument_spec=self.spec.argument_spec, 173 supports_check_mode=self.spec.supports_check_mode 174 ) 175 176 # Override methods in the specific type of manager 177 mm = ModuleManager(module=module) 178 mm.exists = Mock(return_value=True) 179 mm.read_current_from_device = Mock(return_value=current) 180 181 results = mm.exec_module() 182 183 assert results['changed'] is False 184 185 def test_update_interval(self, *args): 186 set_module_args(dict( 187 name='foo', 188 interval=10, 189 provider=dict( 190 server='localhost', 191 password='password', 192 user='admin' 193 ) 194 )) 195 196 current = Parameters(params=load_fixture('load_ltm_monitor_tcp_echo.json')) 197 module = AnsibleModule( 198 argument_spec=self.spec.argument_spec, 199 supports_check_mode=self.spec.supports_check_mode 200 ) 201 202 # Override methods in the specific type of manager 203 mm = ModuleManager(module=module) 204 mm.exists = Mock(return_value=True) 205 mm.read_current_from_device = Mock(return_value=current) 206 mm.update_on_device = Mock(return_value=True) 207 208 results = mm.exec_module() 209 210 assert results['changed'] is True 211 assert results['interval'] == 10 212 213 def test_update_interval_larger_than_existing_timeout(self, *args): 214 set_module_args(dict( 215 name='foo', 216 interval=30, 217 provider=dict( 218 server='localhost', 219 password='password', 220 user='admin' 221 ) 222 )) 223 224 current = Parameters(params=load_fixture('load_ltm_monitor_tcp_echo.json')) 225 module = AnsibleModule( 226 argument_spec=self.spec.argument_spec, 227 supports_check_mode=self.spec.supports_check_mode 228 ) 229 230 # Override methods in the specific type of manager 231 mm = ModuleManager(module=module) 232 mm.exists = Mock(return_value=True) 233 mm.read_current_from_device = Mock(return_value=current) 234 mm.update_on_device = Mock(return_value=True) 235 236 with pytest.raises(F5ModuleError) as ex: 237 mm.exec_module() 238 239 assert "must be less than" in str(ex.value) 240 241 def test_update_interval_larger_than_new_timeout(self, *args): 242 set_module_args(dict( 243 name='foo', 244 interval=10, 245 timeout=5, 246 provider=dict( 247 server='localhost', 248 password='password', 249 user='admin' 250 ) 251 )) 252 253 current = Parameters(params=load_fixture('load_ltm_monitor_tcp_echo.json')) 254 module = AnsibleModule( 255 argument_spec=self.spec.argument_spec, 256 supports_check_mode=self.spec.supports_check_mode 257 ) 258 259 # Override methods in the specific type of manager 260 mm = ModuleManager(module=module) 261 mm.exists = Mock(return_value=True) 262 mm.read_current_from_device = Mock(return_value=current) 263 mm.update_on_device = Mock(return_value=True) 264 265 with pytest.raises(F5ModuleError) as ex: 266 mm.exec_module() 267 268 assert "must be less than" in str(ex.value) 269 270 def test_update_timeout(self, *args): 271 set_module_args(dict( 272 name='foo', 273 timeout=300, 274 provider=dict( 275 server='localhost', 276 password='password', 277 user='admin' 278 ) 279 )) 280 281 current = Parameters(params=load_fixture('load_ltm_monitor_tcp_echo.json')) 282 module = AnsibleModule( 283 argument_spec=self.spec.argument_spec, 284 supports_check_mode=self.spec.supports_check_mode 285 ) 286 287 # Override methods in the specific type of manager 288 mm = ModuleManager(module=module) 289 mm.exists = Mock(return_value=True) 290 mm.read_current_from_device = Mock(return_value=current) 291 mm.update_on_device = Mock(return_value=True) 292 293 results = mm.exec_module() 294 assert results['changed'] is True 295 assert results['timeout'] == 300 296 297 def test_update_time_until_up(self, *args): 298 set_module_args(dict( 299 name='foo', 300 time_until_up=300, 301 provider=dict( 302 server='localhost', 303 password='password', 304 user='admin' 305 ) 306 )) 307 308 current = Parameters(params=load_fixture('load_ltm_monitor_tcp_echo.json')) 309 module = AnsibleModule( 310 argument_spec=self.spec.argument_spec, 311 supports_check_mode=self.spec.supports_check_mode 312 ) 313 314 # Override methods in the specific type of manager 315 mm = ModuleManager(module=module) 316 mm.exists = Mock(return_value=True) 317 mm.read_current_from_device = Mock(return_value=current) 318 mm.update_on_device = Mock(return_value=True) 319 320 results = mm.exec_module() 321 322 assert results['changed'] is True 323 assert results['time_until_up'] == 300 324