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_snmp import ( 20 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 def test_module_parameters(self): 51 args = dict( 52 agent_status_traps='enabled', 53 agent_authentication_traps='enabled', 54 contact='Alice@foo.org', 55 device_warning_traps='enabled', 56 location='Lunar orbit', 57 ) 58 p = ModuleParameters(params=args) 59 assert p.agent_status_traps == 'enabled' 60 assert p.agent_authentication_traps == 'enabled' 61 assert p.device_warning_traps == 'enabled' 62 assert p.location == 'Lunar orbit' 63 assert p.contact == 'Alice@foo.org' 64 65 def test_module_parameters_disabled(self): 66 args = dict( 67 agent_status_traps='disabled', 68 agent_authentication_traps='disabled', 69 device_warning_traps='disabled', 70 ) 71 p = ModuleParameters(params=args) 72 assert p.agent_status_traps == 'disabled' 73 assert p.agent_authentication_traps == 'disabled' 74 assert p.device_warning_traps == 'disabled' 75 76 def test_api_parameters(self): 77 args = dict( 78 agentTrap='enabled', 79 authTrap='enabled', 80 bigipTraps='enabled', 81 sysLocation='Lunar orbit', 82 sysContact='Alice@foo.org', 83 ) 84 p = ApiParameters(params=args) 85 assert p.agent_status_traps == 'enabled' 86 assert p.agent_authentication_traps == 'enabled' 87 assert p.device_warning_traps == 'enabled' 88 assert p.location == 'Lunar orbit' 89 assert p.contact == 'Alice@foo.org' 90 91 def test_api_parameters_disabled(self): 92 args = dict( 93 agentTrap='disabled', 94 authTrap='disabled', 95 bigipTraps='disabled', 96 ) 97 p = ApiParameters(params=args) 98 assert p.agent_status_traps == 'disabled' 99 assert p.agent_authentication_traps == 'disabled' 100 assert p.device_warning_traps == 'disabled' 101 102 103class TestManager(unittest.TestCase): 104 def setUp(self): 105 self.spec = ArgumentSpec() 106 self.p2 = patch('ansible_collections.f5networks.f5_modules.plugins.modules.bigip_snmp.tmos_version') 107 self.p3 = patch('ansible_collections.f5networks.f5_modules.plugins.modules.bigip_snmp.send_teem') 108 self.m2 = self.p2.start() 109 self.m2.return_value = '14.1.0' 110 self.m3 = self.p3.start() 111 self.m3.return_value = True 112 113 def tearDown(self): 114 self.p2.stop() 115 self.p3.stop() 116 117 def test_update_agent_status_traps(self, *args): 118 set_module_args(dict( 119 agent_status_traps='enabled', 120 provider=dict( 121 server='localhost', 122 password='password', 123 user='admin' 124 ) 125 )) 126 127 # Configure the parameters that would be returned by querying the 128 # remote device 129 current = ApiParameters( 130 params=dict( 131 agent_status_traps='disabled' 132 ) 133 ) 134 135 module = AnsibleModule( 136 argument_spec=self.spec.argument_spec, 137 supports_check_mode=self.spec.supports_check_mode 138 ) 139 mm = ModuleManager(module=module) 140 141 # Override methods to force specific logic in the module to happen 142 mm.update_on_device = Mock(return_value=True) 143 mm.read_current_from_device = Mock(return_value=current) 144 145 results = mm.exec_module() 146 147 assert results['changed'] is True 148 assert results['agent_status_traps'] == 'enabled' 149 150 def test_update_allowed_addresses(self, *args): 151 set_module_args(dict( 152 allowed_addresses=[ 153 '127.0.0.0/8', 154 '10.10.10.10', 155 'foo', 156 'baz.foo.com' 157 ], 158 provider=dict( 159 server='localhost', 160 password='password', 161 user='admin' 162 ) 163 )) 164 165 # Configure the parameters that would be returned by querying the 166 # remote device 167 current = ApiParameters( 168 params=dict( 169 allowed_addresses=['127.0.0.0/8'] 170 ) 171 ) 172 173 module = AnsibleModule( 174 argument_spec=self.spec.argument_spec, 175 supports_check_mode=self.spec.supports_check_mode 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 len(results['allowed_addresses']) == 4 187 assert results['allowed_addresses'] == [ 188 '10.10.10.10', '127.0.0.0/8', 'baz.foo.com', 'foo' 189 ] 190 191 def test_update_allowed_addresses_default(self, *args): 192 set_module_args(dict( 193 allowed_addresses=[ 194 'default' 195 ], 196 provider=dict( 197 server='localhost', 198 password='password', 199 user='admin' 200 ) 201 )) 202 203 # Configure the parameters that would be returned by querying the 204 # remote device 205 current = ApiParameters( 206 params=dict( 207 allowed_addresses=['10.0.0.0'] 208 ) 209 ) 210 211 module = AnsibleModule( 212 argument_spec=self.spec.argument_spec, 213 supports_check_mode=self.spec.supports_check_mode 214 ) 215 mm = ModuleManager(module=module) 216 217 # Override methods to force specific logic in the module to happen 218 mm.update_on_device = Mock(return_value=True) 219 mm.read_current_from_device = Mock(return_value=current) 220 221 results = mm.exec_module() 222 223 assert results['changed'] is True 224 assert len(results['allowed_addresses']) == 1 225 assert results['allowed_addresses'] == ['127.0.0.0/8'] 226 227 def test_update_allowed_addresses_empty(self, *args): 228 set_module_args(dict( 229 allowed_addresses=[''], 230 provider=dict( 231 server='localhost', 232 password='password', 233 user='admin' 234 ) 235 )) 236 237 # Configure the parameters that would be returned by querying the 238 # remote device 239 current = ApiParameters( 240 params=dict( 241 allowed_addresses=['10.0.0.0'] 242 ) 243 ) 244 245 module = AnsibleModule( 246 argument_spec=self.spec.argument_spec, 247 supports_check_mode=self.spec.supports_check_mode 248 ) 249 mm = ModuleManager(module=module) 250 251 # Override methods to force specific logic in the module to happen 252 mm.update_on_device = Mock(return_value=True) 253 mm.read_current_from_device = Mock(return_value=current) 254 255 results = mm.exec_module() 256 257 assert results['changed'] is True 258 assert len(results['allowed_addresses']) == 1 259 assert results['allowed_addresses'] == ['127.0.0.0/8'] 260