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_iapp_service import ( 20 Parameters, ApiParameters, ModuleParameters, 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 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 51 def test_module_parameters_keys(self): 52 args = load_fixture('create_iapp_service_parameters_f5_http.json') 53 p = ModuleParameters(params=args) 54 55 # Assert the top-level keys 56 assert p.name == 'http_example' 57 assert p.partition == 'Common' 58 assert p.template == '/Common/f5.http' 59 assert p.device_group is None 60 assert p.inheritedTrafficGroup == 'true' 61 assert p.inheritedDevicegroup == 'true' 62 assert p.traffic_group == '/Common/traffic-group-local-only' 63 64 def test_module_parameters_lists(self): 65 args = load_fixture('create_iapp_service_parameters_f5_http.json') 66 p = ModuleParameters(params=args) 67 68 assert 'lists' in p._values 69 70 assert p.lists[0]['name'] == 'irules__irules' 71 assert p.lists[0]['encrypted'] == 'no' 72 assert len(p.lists[0]['value']) == 1 73 assert p.lists[0]['value'][0] == '/Common/lgyft' 74 75 assert p.lists[1]['name'] == 'net__client_vlan' 76 assert p.lists[1]['encrypted'] == 'no' 77 assert len(p.lists[1]['value']) == 1 78 assert p.lists[1]['value'][0] == '/Common/net2' 79 80 def test_module_parameters_tables(self): 81 args = load_fixture('create_iapp_service_parameters_f5_http.json') 82 p = ModuleParameters(params=args) 83 84 assert 'tables' in p._values 85 86 assert 'columnNames' in p.tables[0] 87 assert len(p.tables[0]['columnNames']) == 1 88 assert p.tables[0]['columnNames'][0] == 'name' 89 90 assert 'name' in p.tables[0] 91 assert p.tables[0]['name'] == 'pool__hosts' 92 93 assert 'rows' in p.tables[0] 94 assert len(p.tables[0]['rows']) == 1 95 assert 'row' in p.tables[0]['rows'][0] 96 assert len(p.tables[0]['rows'][0]['row']) == 1 97 assert p.tables[0]['rows'][0]['row'][0] == 'demo.example.com' 98 99 assert len(p.tables[1]['rows']) == 2 100 assert 'row' in p.tables[0]['rows'][0] 101 assert len(p.tables[1]['rows'][0]['row']) == 2 102 assert p.tables[1]['rows'][0]['row'][0] == '10.1.1.1' 103 assert p.tables[1]['rows'][0]['row'][1] == '0' 104 assert p.tables[1]['rows'][1]['row'][0] == '10.1.1.2' 105 assert p.tables[1]['rows'][1]['row'][1] == '0' 106 107 def test_module_parameters_variables(self): 108 args = load_fixture('create_iapp_service_parameters_f5_http.json') 109 p = ModuleParameters(params=args) 110 111 assert 'variables' in p._values 112 assert len(p.variables) == 34 113 114 # Assert one configuration value 115 assert 'name' in p.variables[0] 116 assert 'value' in p.variables[0] 117 assert p.variables[0]['name'] == 'afm__dos_security_profile' 118 assert p.variables[0]['value'] == '/#do_not_use#' 119 120 # Assert a second configuration value 121 assert 'name' in p.variables[1] 122 assert 'value' in p.variables[1] 123 assert p.variables[1]['name'] == 'afm__policy' 124 assert p.variables[1]['value'] == '/#do_not_use#' 125 126 def test_module_strict_updates_from_top_level(self): 127 # Assumes the user did not provide any parameters 128 129 args = dict( 130 strict_updates=True 131 ) 132 p = ModuleParameters(params=args) 133 assert p.strict_updates == 'enabled' 134 135 args = dict( 136 strict_updates=False 137 ) 138 p = ModuleParameters(params=args) 139 assert p.strict_updates == 'disabled' 140 141 def test_module_strict_updates_override_from_top_level(self): 142 args = dict( 143 strict_updates=True, 144 parameters=dict( 145 strictUpdates='disabled' 146 ) 147 ) 148 p = ModuleParameters(params=args) 149 assert p.strict_updates == 'enabled' 150 151 args = dict( 152 strict_updates=False, 153 parameters=dict( 154 strictUpdates='enabled' 155 ) 156 ) 157 p = ModuleParameters(params=args) 158 assert p.strict_updates == 'disabled' 159 160 def test_module_strict_updates_only_parameters(self): 161 args = dict( 162 parameters=dict( 163 strictUpdates='disabled' 164 ) 165 ) 166 p = ModuleParameters(params=args) 167 assert p.strict_updates == 'disabled' 168 169 args = dict( 170 parameters=dict( 171 strictUpdates='enabled' 172 ) 173 ) 174 p = ModuleParameters(params=args) 175 assert p.strict_updates == 'enabled' 176 177 def test_api_strict_updates_from_top_level(self): 178 args = dict( 179 strictUpdates='enabled' 180 ) 181 p = ApiParameters(params=args) 182 assert p.strict_updates == 'enabled' 183 184 args = dict( 185 strictUpdates='disabled' 186 ) 187 p = ApiParameters(params=args) 188 assert p.strict_updates == 'disabled' 189 190 def test_api_parameters_variables(self): 191 args = dict( 192 variables=[ 193 dict( 194 name="client__http_compression", 195 encrypted="no", 196 value="/#create_new#" 197 ) 198 ] 199 ) 200 p = ApiParameters(params=args) 201 assert p.variables[0]['name'] == 'client__http_compression' 202 203 def test_api_parameters_tables(self): 204 args = dict( 205 tables=[ 206 { 207 "name": "pool__members", 208 "columnNames": [ 209 "addr", 210 "port", 211 "connection_limit" 212 ], 213 "rows": [ 214 { 215 "row": [ 216 "12.12.12.12", 217 "80", 218 "0" 219 ] 220 }, 221 { 222 "row": [ 223 "13.13.13.13", 224 "443", 225 10 226 ] 227 } 228 ] 229 } 230 ] 231 ) 232 p = ApiParameters(params=args) 233 assert p.tables[0]['name'] == 'pool__members' 234 assert p.tables[0]['columnNames'] == ['addr', 'port', 'connection_limit'] 235 assert len(p.tables[0]['rows']) == 2 236 assert 'row' in p.tables[0]['rows'][0] 237 assert 'row' in p.tables[0]['rows'][1] 238 assert p.tables[0]['rows'][0]['row'] == ['12.12.12.12', '80', '0'] 239 assert p.tables[0]['rows'][1]['row'] == ['13.13.13.13', '443', '10'] 240 241 def test_api_parameters_device_group(self): 242 args = dict( 243 deviceGroup='none' 244 ) 245 p = ApiParameters(params=args) 246 assert p.device_group is None 247 248 def test_api_parameters_inherited_traffic_group(self): 249 args = dict( 250 inheritedTrafficGroup='true' 251 ) 252 p = ApiParameters(params=args) 253 assert p.inheritedTrafficGroup == 'true' 254 255 def test_api_parameters_inherited_devicegroup(self): 256 args = dict( 257 inheritedDevicegroup='true' 258 ) 259 p = ApiParameters(params=args) 260 assert p.inheritedDevicegroup == 'true' 261 262 def test_api_parameters_traffic_group(self): 263 args = dict( 264 trafficGroup='/Common/traffic-group-local-only' 265 ) 266 p = ApiParameters(params=args) 267 assert p.traffic_group == '/Common/traffic-group-local-only' 268 269 def test_module_template_same_partition(self): 270 args = dict( 271 template='foo', 272 partition='bar' 273 ) 274 p = ModuleParameters(params=args) 275 assert p.template == '/bar/foo' 276 277 def test_module_template_same_partition_full_path(self): 278 args = dict( 279 template='/bar/foo', 280 partition='bar' 281 ) 282 p = ModuleParameters(params=args) 283 assert p.template == '/bar/foo' 284 285 def test_module_template_different_partition_full_path(self): 286 args = dict( 287 template='/Common/foo', 288 partition='bar' 289 ) 290 p = ModuleParameters(params=args) 291 assert p.template == '/Common/foo' 292 293 294class TestManager(unittest.TestCase): 295 296 def setUp(self): 297 self.spec = ArgumentSpec() 298 self.p2 = patch('ansible_collections.f5networks.f5_modules.plugins.modules.bigip_iapp_service.tmos_version') 299 self.p3 = patch('ansible_collections.f5networks.f5_modules.plugins.modules.bigip_iapp_service.send_teem') 300 self.m2 = self.p2.start() 301 self.m2.return_value = '14.1.0' 302 self.m3 = self.p3.start() 303 self.m3.return_value = True 304 305 def tearDown(self): 306 self.p2.stop() 307 self.p3.stop() 308 309 def test_create_service(self, *args): 310 parameters = load_fixture('create_iapp_service_parameters_f5_http.json') 311 set_module_args(dict( 312 name='foo', 313 template='f5.http', 314 parameters=parameters, 315 state='present', 316 provider=dict( 317 server='localhost', 318 password='password', 319 user='admin' 320 ) 321 )) 322 323 module = AnsibleModule( 324 argument_spec=self.spec.argument_spec, 325 supports_check_mode=self.spec.supports_check_mode 326 ) 327 mm = ModuleManager(module=module) 328 329 # Override methods to force specific logic in the module to happen 330 mm.exists = Mock(return_value=False) 331 mm.create_on_device = Mock(return_value=True) 332 mm.template_exists = Mock(return_value=True) 333 334 results = mm.exec_module() 335 assert results['changed'] is True 336 337 def test_update_agent_status_traps(self, *args): 338 parameters = load_fixture('update_iapp_service_parameters_f5_http.json') 339 set_module_args(dict( 340 name='foo', 341 template='f5.http', 342 parameters=parameters, 343 state='present', 344 provider=dict( 345 server='localhost', 346 password='password', 347 user='admin' 348 ) 349 )) 350 351 # Configure the parameters that would be returned by querying the 352 # remote device 353 parameters = load_fixture('create_iapp_service_parameters_f5_http.json') 354 current = Parameters(parameters) 355 356 module = AnsibleModule( 357 argument_spec=self.spec.argument_spec, 358 supports_check_mode=self.spec.supports_check_mode 359 ) 360 mm = ModuleManager(module=module) 361 362 # Override methods to force specific logic in the module to happen 363 mm.exists = Mock(return_value=True) 364 mm.update_on_device = Mock(return_value=True) 365 mm.read_current_from_device = Mock(return_value=current) 366 367 results = mm.exec_module() 368 assert results['changed'] is True 369