1import json 2import os 3 4from units.compat.mock import mock_open 5from ansible.module_utils import basic 6from ansible.module_utils._text import to_native 7import ansible.module_utils.six 8from ansible.module_utils.six.moves import xmlrpc_client 9from ansible.modules.packaging.os import rhn_register 10 11import pytest 12 13 14SYSTEMID = """<?xml version="1.0"?> 15<params> 16<param> 17<value><struct> 18<member> 19<name>system_id</name> 20<value><string>ID-123456789</string></value> 21</member> 22</struct></value> 23</param> 24</params> 25""" 26 27 28def skipWhenAllModulesMissing(modules): 29 """Skip the decorated test unless one of modules is available.""" 30 for module in modules: 31 try: 32 __import__(module) 33 return False 34 except ImportError: 35 continue 36 37 return True 38 39 40orig_import = __import__ 41 42 43@pytest.fixture 44def import_libxml(mocker): 45 def mock_import(name, *args, **kwargs): 46 if name in ['libxml2', 'libxml']: 47 raise ImportError() 48 else: 49 return orig_import(name, *args, **kwargs) 50 51 if ansible.module_utils.six.PY3: 52 mocker.patch('builtins.__import__', side_effect=mock_import) 53 else: 54 mocker.patch('__builtin__.__import__', side_effect=mock_import) 55 56 57@pytest.fixture 58def patch_rhn(mocker): 59 load_config_return = { 60 'serverURL': 'https://xmlrpc.rhn.redhat.com/XMLRPC', 61 'systemIdPath': '/etc/sysconfig/rhn/systemid' 62 } 63 64 mocker.patch.object(rhn_register.Rhn, 'load_config', return_value=load_config_return) 65 mocker.patch.object(rhn_register, 'HAS_UP2DATE_CLIENT', mocker.PropertyMock(return_value=True)) 66 67 68@pytest.mark.skipif(skipWhenAllModulesMissing(['libxml2', 'libxml']), reason='none are available: libxml2, libxml') 69def test_systemid_with_requirements(capfd, mocker, patch_rhn): 70 """Check 'msg' and 'changed' results""" 71 72 mocker.patch.object(rhn_register.Rhn, 'enable') 73 mock_isfile = mocker.patch('os.path.isfile', return_value=True) 74 mocker.patch('ansible.modules.packaging.os.rhn_register.open', mock_open(read_data=SYSTEMID), create=True) 75 rhn = rhn_register.Rhn() 76 assert '123456789' == to_native(rhn.systemid) 77 78 79@pytest.mark.parametrize('patch_ansible_module', [{}], indirect=['patch_ansible_module']) 80@pytest.mark.usefixtures('patch_ansible_module') 81def test_systemid_requirements_missing(capfd, mocker, patch_rhn, import_libxml): 82 """Check that missing dependencies are detected""" 83 84 mocker.patch('os.path.isfile', return_value=True) 85 mocker.patch('ansible.modules.packaging.os.rhn_register.open', mock_open(read_data=SYSTEMID), create=True) 86 87 with pytest.raises(SystemExit): 88 rhn_register.main() 89 90 out, err = capfd.readouterr() 91 results = json.loads(out) 92 assert results['failed'] 93 assert 'Missing arguments' in results['msg'] 94 95 96@pytest.mark.parametrize('patch_ansible_module', [{}], indirect=['patch_ansible_module']) 97@pytest.mark.usefixtures('patch_ansible_module') 98def test_without_required_parameters(capfd, patch_rhn): 99 """Failure must occurs when all parameters are missing""" 100 101 with pytest.raises(SystemExit): 102 rhn_register.main() 103 out, err = capfd.readouterr() 104 results = json.loads(out) 105 assert results['failed'] 106 assert 'Missing arguments' in results['msg'] 107 108 109TESTED_MODULE = rhn_register.__name__ 110TEST_CASES = [ 111 [ 112 # Registering an unregistered host with channels 113 { 114 'channels': 'rhel-x86_64-server-6', 115 'username': 'user', 116 'password': 'pass', 117 }, 118 { 119 'calls': [ 120 ('auth.login', ['X' * 43]), 121 ('channel.software.listSystemChannels', 122 [[{'channel_name': 'Red Hat Enterprise Linux Server (v. 6 for 64-bit x86_64)', 'channel_label': 'rhel-x86_64-server-6'}]]), 123 ('channel.software.setSystemChannels', [1]), 124 ('auth.logout', [1]), 125 ], 126 'is_registered': False, 127 'is_registered.call_count': 1, 128 'enable.call_count': 1, 129 'systemid.call_count': 2, 130 'changed': True, 131 'msg': "System successfully registered to 'rhn.redhat.com'.", 132 'run_command.call_count': 1, 133 'run_command.call_args': '/usr/sbin/rhnreg_ks', 134 'request_called': True, 135 'unlink.call_count': 0, 136 } 137 ], 138 [ 139 # Registering an unregistered host without channels 140 { 141 'activationkey': 'key', 142 'username': 'user', 143 'password': 'pass', 144 }, 145 { 146 'calls': [ 147 ], 148 'is_registered': False, 149 'is_registered.call_count': 1, 150 'enable.call_count': 1, 151 'systemid.call_count': 0, 152 'changed': True, 153 'msg': "System successfully registered to 'rhn.redhat.com'.", 154 'run_command.call_count': 1, 155 'run_command.call_args': '/usr/sbin/rhnreg_ks', 156 'request_called': False, 157 'unlink.call_count': 0, 158 } 159 ], 160 [ 161 # Register an host already registered, check that result is unchanged 162 { 163 'activationkey': 'key', 164 'username': 'user', 165 'password': 'pass', 166 }, 167 { 168 'calls': [ 169 ], 170 'is_registered': True, 171 'is_registered.call_count': 1, 172 'enable.call_count': 0, 173 'systemid.call_count': 0, 174 'changed': False, 175 'msg': 'System already registered.', 176 'run_command.call_count': 0, 177 'request_called': False, 178 'unlink.call_count': 0, 179 }, 180 ], 181 [ 182 # Unregister an host, check that result is changed 183 { 184 'activationkey': 'key', 185 'username': 'user', 186 'password': 'pass', 187 'state': 'absent', 188 }, 189 { 190 'calls': [ 191 ('auth.login', ['X' * 43]), 192 ('system.deleteSystems', [1]), 193 ('auth.logout', [1]), 194 ], 195 'is_registered': True, 196 'is_registered.call_count': 1, 197 'enable.call_count': 0, 198 'systemid.call_count': 1, 199 'changed': True, 200 'msg': 'System successfully unregistered from rhn.redhat.com.', 201 'run_command.call_count': 0, 202 'request_called': True, 203 'unlink.call_count': 1, 204 } 205 ], 206 [ 207 # Unregister a unregistered host (systemid missing) locally, check that result is unchanged 208 { 209 'activationkey': 'key', 210 'username': 'user', 211 'password': 'pass', 212 'state': 'absent', 213 }, 214 { 215 'calls': [], 216 'is_registered': False, 217 'is_registered.call_count': 1, 218 'enable.call_count': 0, 219 'systemid.call_count': 0, 220 'changed': False, 221 'msg': 'System already unregistered.', 222 'run_command.call_count': 0, 223 'request_called': False, 224 'unlink.call_count': 0, 225 } 226 227 ], 228 [ 229 # Unregister an unknown host (an host with a systemid available locally, check that result contains failed 230 { 231 'activationkey': 'key', 232 'username': 'user', 233 'password': 'pass', 234 'state': 'absent', 235 }, 236 { 237 'calls': [ 238 ('auth.login', ['X' * 43]), 239 ('system.deleteSystems', xmlrpc_client.Fault(1003, 'The following systems were NOT deleted: 123456789')), 240 ('auth.logout', [1]), 241 ], 242 'is_registered': True, 243 'is_registered.call_count': 1, 244 'enable.call_count': 0, 245 'systemid.call_count': 1, 246 'failed': True, 247 'msg': "Failed to unregister: <Fault 1003: 'The following systems were NOT deleted: 123456789'>", 248 'run_command.call_count': 0, 249 'request_called': True, 250 'unlink.call_count': 0, 251 } 252 ], 253] 254 255 256@pytest.mark.parametrize('patch_ansible_module, testcase', TEST_CASES, indirect=['patch_ansible_module']) 257@pytest.mark.usefixtures('patch_ansible_module') 258def test_register_parameters(mocker, capfd, mock_request, patch_rhn, testcase): 259 # successful execution, no output 260 mocker.patch.object(basic.AnsibleModule, 'run_command', return_value=(0, '', '')) 261 mock_is_registered = mocker.patch.object(rhn_register.Rhn, 'is_registered', mocker.PropertyMock(return_value=testcase['is_registered'])) 262 mocker.patch.object(rhn_register.Rhn, 'enable') 263 mock_systemid = mocker.patch.object(rhn_register.Rhn, 'systemid', mocker.PropertyMock(return_value=12345)) 264 mocker.patch('os.unlink', return_value=True) 265 266 with pytest.raises(SystemExit): 267 rhn_register.main() 268 269 assert basic.AnsibleModule.run_command.call_count == testcase['run_command.call_count'] 270 if basic.AnsibleModule.run_command.call_count: 271 assert basic.AnsibleModule.run_command.call_args[0][0][0] == testcase['run_command.call_args'] 272 273 assert mock_is_registered.call_count == testcase['is_registered.call_count'] 274 assert rhn_register.Rhn.enable.call_count == testcase['enable.call_count'] 275 assert mock_systemid.call_count == testcase['systemid.call_count'] 276 assert xmlrpc_client.Transport.request.called == testcase['request_called'] 277 assert os.unlink.call_count == testcase['unlink.call_count'] 278 279 out, err = capfd.readouterr() 280 results = json.loads(out) 281 assert results.get('changed') == testcase.get('changed') 282 assert results.get('failed') == testcase.get('failed') 283 assert results['msg'] == testcase['msg'] 284 assert not testcase['calls'] # all calls should have been consumed 285