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_remote_syslog import ( 20 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 remote_host='10.10.10.10', 53 remote_port=514, 54 local_ip='1.1.1.1' 55 ) 56 57 p = ModuleParameters(params=args) 58 assert p.remote_host == '10.10.10.10' 59 assert p.remote_port == 514 60 assert p.local_ip == '1.1.1.1' 61 62 63class TestManager(unittest.TestCase): 64 def setUp(self): 65 self.spec = ArgumentSpec() 66 self.p2 = patch('ansible_collections.f5networks.f5_modules.plugins.modules.bigip_remote_syslog.tmos_version') 67 self.p3 = patch('ansible_collections.f5networks.f5_modules.plugins.modules.bigip_remote_syslog.send_teem') 68 self.m2 = self.p2.start() 69 self.m2.return_value = '14.1.0' 70 self.m3 = self.p3.start() 71 self.m3.return_value = True 72 73 def tearDown(self): 74 self.p2.stop() 75 self.p3.stop() 76 77 def test_create_remote_syslog(self, *args): 78 set_module_args(dict( 79 remote_host='1.1.1.1', 80 provider=dict( 81 server='localhost', 82 password='password', 83 user='admin' 84 ) 85 )) 86 87 fixture = load_fixture('load_tm_sys_syslog_1.json') 88 current = fixture['remoteServers'] 89 90 module = AnsibleModule( 91 argument_spec=self.spec.argument_spec, 92 supports_check_mode=self.spec.supports_check_mode 93 ) 94 95 # Override methods in the specific type of manager 96 mm = ModuleManager(module=module) 97 mm.exists = Mock(side_effect=[False, True]) 98 mm.update_on_device = Mock(return_value=True) 99 mm.read_current_from_device = Mock(return_value=current) 100 101 results = mm.exec_module() 102 103 assert results['changed'] is True 104 105 def test_create_remote_syslog_idempotent(self, *args): 106 set_module_args(dict( 107 name='remotesyslog1', 108 remote_host='10.10.10.10', 109 provider=dict( 110 server='localhost', 111 password='password', 112 user='admin' 113 ) 114 )) 115 116 fixture = load_fixture('load_tm_sys_syslog_1.json') 117 current = fixture['remoteServers'] 118 119 module = AnsibleModule( 120 argument_spec=self.spec.argument_spec, 121 supports_check_mode=self.spec.supports_check_mode 122 ) 123 124 # Override methods in the specific type of manager 125 mm = ModuleManager(module=module) 126 mm.exists = Mock(return_value=True) 127 mm.read_current_from_device = Mock(return_value=current) 128 129 results = mm.exec_module() 130 131 assert results['changed'] is False 132 133 def test_update_remote_port(self, *args): 134 set_module_args(dict( 135 remote_host='10.10.10.10', 136 remote_port=800, 137 provider=dict( 138 server='localhost', 139 password='password', 140 user='admin' 141 ) 142 )) 143 144 fixture = load_fixture('load_tm_sys_syslog_1.json') 145 current = fixture['remoteServers'] 146 147 module = AnsibleModule( 148 argument_spec=self.spec.argument_spec, 149 supports_check_mode=self.spec.supports_check_mode 150 ) 151 152 # Override methods in the specific type of manager 153 mm = ModuleManager(module=module) 154 mm.exists = Mock(return_value=True) 155 mm.read_current_from_device = Mock(return_value=current) 156 mm.update_on_device = Mock(return_value=True) 157 158 results = mm.exec_module() 159 160 assert results['changed'] is True 161 assert results['remote_port'] == 800 162 163 def test_update_local_ip(self, *args): 164 set_module_args(dict( 165 remote_host='10.10.10.10', 166 local_ip='2.2.2.2', 167 provider=dict( 168 server='localhost', 169 password='password', 170 user='admin' 171 ) 172 )) 173 174 fixture = load_fixture('load_tm_sys_syslog_1.json') 175 current = fixture['remoteServers'] 176 177 module = AnsibleModule( 178 argument_spec=self.spec.argument_spec, 179 supports_check_mode=self.spec.supports_check_mode 180 ) 181 182 # Override methods in the specific type of manager 183 mm = ModuleManager(module=module) 184 mm.exists = Mock(return_value=True) 185 mm.read_current_from_device = Mock(return_value=current) 186 mm.update_on_device = Mock(return_value=True) 187 188 results = mm.exec_module() 189 190 assert results['changed'] is True 191 assert results['local_ip'] == '2.2.2.2' 192 193 def test_update_no_name_dupe_host(self, *args): 194 set_module_args(dict( 195 remote_host='10.10.10.10', 196 local_ip='2.2.2.2', 197 provider=dict( 198 server='localhost', 199 password='password', 200 user='admin' 201 ) 202 )) 203 204 fixture = load_fixture('load_tm_sys_syslog_2.json') 205 current = fixture['remoteServers'] 206 207 module = AnsibleModule( 208 argument_spec=self.spec.argument_spec, 209 supports_check_mode=self.spec.supports_check_mode 210 ) 211 212 # Override methods in the specific type of manager 213 mm = ModuleManager(module=module) 214 mm.exists = Mock(return_value=True) 215 mm.read_current_from_device = Mock(return_value=current) 216 mm.update_on_device = Mock(return_value=True) 217 218 with pytest.raises(F5ModuleError) as ex: 219 mm.exec_module() 220 221 assert "Multiple occurrences of hostname" in str(ex.value) 222