# Copyright 2014, 2017 IBM Corp. # # All Rights Reserved. # # Licensed under the Apache License, Version 2.0 (the "License"); you may # not use this file except in compliance with the License. You may obtain # a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the # License for the specific language governing permissions and limitations # under the License. import copy import gc import mock import os import requests.models as req_mod import requests.structures as req_struct import subunit import sys import testtools import pypowervm.adapter as adp import pypowervm.exceptions as pvmex import pypowervm.tests.lib as testlib import pypowervm.tests.test_fixtures as fx _logon_response_password = testlib.file2b("logon.xml") _logon_response_file = testlib.file2b("logon_file.xml") class TestSession(subunit.IsolatedTestCase, testtools.TestCase): """Test cases to test the Session classes and methods.""" @mock.patch('time.sleep') @mock.patch('lxml.etree.fromstring', new=mock.Mock()) @mock.patch('pypowervm.adapter.Session._get_auth_tok_from_file', new=mock.Mock()) def test_Session(self, mock_sleep): """Ensure Session can be instantiated, and test logon retries.""" # Passing in 0.0.0.0 will raise a ConnectionError or SSLError, but only # if it gets past all the __init__ setup since _logon is the last # statement. self.assertRaises((pvmex.ConnectionError, pvmex.SSLError), adp.Session, '0.0.0.0', 'uid', 'pwd') mock_sleep.assert_not_called() # Now set up a retry self.assertRaises((pvmex.ConnectionError, pvmex.SSLError), adp.Session, '0.0.0.0', 'uid', 'pwd', conn_tries=5) # 5 tries = 4 sleeps mock_sleep.assert_has_calls([mock.call(2)] * 4) # Ensure 404 on the logon URI also retries mock_sleep.reset_mock() with mock.patch('requests.Session.request') as mock_rq: mock_rq.side_effect = [mock.Mock(status_code=404), mock.Mock(status_code=204)] adp.Session(conn_tries=5) # Only retried once, after the 404 mock_sleep.assert_called_once_with(2) @mock.patch('pypowervm.adapter.Session._logon', new=mock.Mock()) @mock.patch('pypowervm.adapter._EventListener._get_events') def test_session_init(self, mock_get_evts): """Ensure proper parameter handling in the Session initializer.""" mock_get_evts.return_value = {'general': 'init'}, [], [] logfx = self.useFixture(fx.LoggingFx()) # No params - local, file-based, http. sess = adp.Session() self.assertTrue(sess.use_file_auth) self.assertIsNone(sess.password) self.assertTrue(sess.username.startswith('pypowervm_')) self.assertEqual('localhost', sess.host) self.assertEqual('http', sess.protocol) self.assertEqual(12080, sess.port) self.assertEqual('http://localhost:12080', sess.dest) self.assertEqual(1200, sess.timeout) self.assertEqual('/etc/ssl/certs/', sess.certpath) self.assertEqual('.crt', sess.certext) # localhost + http is okay self.assertEqual(0, logfx.patchers['warning'].mock.call_count) # Verify unique session names sess2 = adp.Session() self.assertNotEqual(sess.username, sess2.username) # Ensure proper protocol, port, and certpath defaulting when remote sess = adp.Session(host='host', username='user', password='pass') self.assertFalse(sess.use_file_auth) self.assertIsNotNone(sess.password) self.assertEqual('user', sess.username) self.assertEqual('host', sess.host) self.assertEqual('https', sess.protocol) self.assertEqual(12443, sess.port) self.assertEqual('https://host:12443', sess.dest) self.assertEqual(1200, sess.timeout) self.assertEqual('/etc/ssl/certs/', sess.certpath) self.assertEqual('.crt', sess.certext) # non-localhost + (implied) https is okay self.assertEqual(0, logfx.patchers['warning'].mock.call_count) @mock.patch('pypowervm.adapter.Session._logon', new=mock.Mock()) @mock.patch('pypowervm.adapter._EventListener._get_events') @mock.patch('imp.load_source') def test_session_ext_cfg(self, mock_load, mock_get_evts): """Test Session init with external config from env var.""" mock_get_evts.return_value = {'general': 'init'}, [], [] with mock.patch.dict(os.environ, {'PYPOWERVM_SESSION_CONFIG': 'path'}): sess = adp.Session() mock_load.assert_called_once_with('sesscfg', 'path') mock_load.return_value.session_config.assert_called_once_with(sess) @mock.patch('pypowervm.adapter.Session._logon') def test_session_init_remote_http(self, mock_logon): # Proper port defaulting and warning emitted when remote + http with self.assertLogs(adp.__name__, 'WARNING'): sess = adp.Session(host='host', protocol='http') self.assertEqual(12080, sess.port) @mock.patch.object(adp.Session, '_logon') @mock.patch.object(adp.Session, '_logoff') def test_session_clone(self, mock_logoff, mock_logon): sess = adp.Session() # Ensure the id that created the object is recorded. self.assertTrue(hasattr(sess, '_init_by')) # Create a shallow clone and ensure the _init_by does not match the id sess_clone = copy.copy(sess) self.assertTrue(hasattr(sess_clone, '_init_by')) self.assertNotEqual(sess._init_by, id(sess_clone)) # Now test what happens when the clone is garbage collected. self.assertFalse(mock_logoff.called) sess_clone = None gc.collect() # The clone was not logged off self.assertFalse(mock_logoff.called) # Deep copy (properly) raises TypeError in py2 and py3 >= 3.6. pyver = sys.version_info if pyver.major == 2 or (pyver.major == 3 and pyver.minor >= 6): # Ensure deep copies raise an exception. self.assertRaises(TypeError, copy.deepcopy, sess) else: # Or if works, it is not logged off sess_deepclone = copy.deepcopy(sess) # Make pep8 happy, use the clone self.assertIsNotNone(sess_deepclone) sess_deepclone = None gc.collect() # The clone was not logged off self.assertFalse(mock_logoff.called) sess = None gc.collect() # The original session was logged off self.assertTrue(mock_logoff.called) @mock.patch('pypowervm.util.validate_certificate') @mock.patch('requests.Session') def test_logon(self, mock_session, mock_validate_cert): """Ensure a Session can be created and log on to PowerVM.""" # Init test data host = '0.0.0.0' user = 'user' pwd = 'pwd' auditmemento = 'audit' # Create a Response object, that will serve as a mock return value my_response = req_mod.Response() my_response.status_code = 200 my_response.reason = 'OK' dict_headers = {'content-length': '576', 'x-powered-by': 'Servlet/3.0', 'set-cookie': 'JSESSIONID=0000a41BnJsGTNQvBGERA3wR1nj:' '759878cb-4f9a-4b05-a09a-3357abfea3b4; P' 'ath=/; Secure; HttpOnly, CCFWSESSION=E4' 'C0FFBE9130431DBF1864171ECC6A6E; Path=/;' ' Secure; HttpOnly', 'expires': 'Thu, 01 Dec 1994 16:00:00 GMT', 'x-transaction-id': 'XT10000073', 'cache-control': 'no-cache="set-cookie, set-cookie2"', 'date': 'Wed, 23 Jul 2014 21:51:10 GMT', 'content-type': 'application/vnd.ibm.powervm.web+xml; ' 'type=LogonResponse'} my_response.headers = req_struct.CaseInsensitiveDict(dict_headers) my_response._content = _logon_response_password # Mock out the method and class we are not currently testing session = mock_session.return_value session.request.return_value = my_response # Run the actual test result = adp.Session(host, user, pwd, auditmemento=auditmemento) # Verify the result self.assertTrue(result._logged_in) self.assertEqual('PUIoR6x0kP6fQqA7qZ8sLZQJ8MLx9JHfLCYzT4oGFSE2WaGIhaFX' 'IyQYvbqdKNS8QagjBpPi9NP7YR_h61SOJ3krS_RvKAp-oCf2p8x8' 'uvQrrDv-dUzc17IT5DkR7_jv2qc8iUD7DJ6Rw53a17rY0p63KqPg' '9oUGd6Bn3fNDLiEwaBR4WICftVxUFj-tfWMOyZZY2hWEtN2K8ScX' 'vyFMe-w3SleyRbGnlR34jb0A99s=', result._sessToken) self.assertEqual(1, mock_validate_cert.call_count) # No X-MC-Type header => 'HMC' is assumed. self.assertEqual('HMC', result.mc_type) # Now test file-based authentication and X-MC-Type my_response._content = _logon_response_file # Local/HMC is bad self.assertRaises(pvmex.Error, adp.Session) my_response.headers['X-MC-Type'] = 'PVM' result = adp.Session() # Verify the result. self.assertTrue(result._logged_in) # Token read from token_file, as indicated by logon_file.xml response. self.assertEqual('file-based-auth-token', result._sessToken) # validate_certificate should not have been called again self.assertEqual(1, mock_validate_cert.call_count) self.assertEqual('PVM', result.mc_type)