path: root/oslo_vmware/tests/test_api.py

# coding=utf-8
# Copyright (c) 2014 VMware, Inc.
# 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.

"""Unit tests for session management and API invocation classes."""

from datetime import datetime
from unittest import mock

from eventlet import greenthread
from oslo_context import context
import suds

from oslo_vmware import api
from oslo_vmware import exceptions
from oslo_vmware import pbm
from oslo_vmware.tests import base
from oslo_vmware import vim_util


class RetryDecoratorTest(base.TestCase):
    """Tests for retry decorator class."""

    def test_retry(self):
        result = "RESULT"

        @api.RetryDecorator()
        def func(*args, **kwargs):
            return result

        self.assertEqual(result, func())

        def func2(*args, **kwargs):
            return result

        retry = api.RetryDecorator()
        self.assertEqual(result, retry(func2)())
        self.assertTrue(retry._retry_count == 0)

    def test_retry_with_expected_exceptions(self):
        result = "RESULT"
        responses = [exceptions.VimSessionOverLoadException(None),
                     exceptions.VimSessionOverLoadException(None),
                     result]

        def func(*args, **kwargs):
            response = responses.pop(0)
            if isinstance(response, Exception):
                raise response
            return response

        sleep_time_incr = 0.01
        retry_count = 2
        retry = api.RetryDecorator(10, sleep_time_incr, 10,
                                   (exceptions.VimSessionOverLoadException,))
        self.assertEqual(result, retry(func)())
        self.assertTrue(retry._retry_count == retry_count)
        self.assertEqual(retry_count * sleep_time_incr, retry._sleep_time)

    def test_retry_with_max_retries(self):
        responses = [exceptions.VimSessionOverLoadException(None),
                     exceptions.VimSessionOverLoadException(None),
                     exceptions.VimSessionOverLoadException(None)]

        def func(*args, **kwargs):
            response = responses.pop(0)
            if isinstance(response, Exception):
                raise response
            return response

        retry = api.RetryDecorator(2, 0, 0,
                                   (exceptions.VimSessionOverLoadException,))
        self.assertRaises(exceptions.VimSessionOverLoadException, retry(func))
        self.assertTrue(retry._retry_count == 2)

    def test_retry_with_unexpected_exception(self):

        def func(*args, **kwargs):
            raise exceptions.VimException(None)

        retry = api.RetryDecorator()
        self.assertRaises(exceptions.VimException, retry(func))
        self.assertTrue(retry._retry_count == 0)


class VMwareAPISessionTest(base.TestCase):
    """Tests for VMwareAPISession."""

    SERVER_IP = '10.1.2.3'
    PORT = 443
    USERNAME = 'admin'
    PASSWORD = 'password'  # nosec
    POOL_SIZE = 15

    def setUp(self):
        super(VMwareAPISessionTest, self).setUp()
        patcher = mock.patch('oslo_vmware.vim.Vim')
        self.addCleanup(patcher.stop)
        self.VimMock = patcher.start()
        self.VimMock.side_effect = lambda *args, **kw: mock.MagicMock()
        self.cert_mock = mock.Mock()

    def _create_api_session(self, _create_session, retry_count=10,
                            task_poll_interval=1):
        return api.VMwareAPISession(VMwareAPISessionTest.SERVER_IP,
                                    VMwareAPISessionTest.USERNAME,
                                    VMwareAPISessionTest.PASSWORD,
                                    retry_count,
                                    task_poll_interval,
                                    'https',
                                    _create_session,
                                    port=VMwareAPISessionTest.PORT,
                                    cacert=self.cert_mock,
                                    insecure=False,
                                    pool_size=VMwareAPISessionTest.POOL_SIZE)

    def test_vim(self):
        api_session = self._create_api_session(False)
        api_session.vim
        self.VimMock.assert_called_with(
            protocol=api_session._scheme,
            host=VMwareAPISessionTest.SERVER_IP,
            port=VMwareAPISessionTest.PORT,
            wsdl_url=api_session._vim_wsdl_loc,
            cacert=self.cert_mock,
            insecure=False,
            pool_maxsize=VMwareAPISessionTest.POOL_SIZE,
            connection_timeout=None,
            op_id_prefix='oslo.vmware')

    @mock.patch.object(pbm, 'Pbm')
    def test_pbm(self, pbm_mock):
        api_session = self._create_api_session(True)
        vim_obj = api_session.vim
        cookie = mock.Mock()
        vim_obj.get_http_cookie.return_value = cookie
        api_session._pbm_wsdl_loc = mock.Mock()

        pbm = mock.Mock()
        pbm_mock.return_value = pbm
        api_session._get_session_cookie = mock.Mock(return_value=cookie)

        self.assertEqual(pbm, api_session.pbm)
        pbm.set_soap_cookie.assert_called_once_with(cookie)

    def test_create_session(self):
        session = mock.Mock()
        session.key = "12345"
        api_session = self._create_api_session(False)
        cookie = mock.Mock()
        vim_obj = api_session.vim
        vim_obj.Login.return_value = session
        vim_obj.get_http_cookie.return_value = cookie

        pbm = mock.Mock()
        api_session._pbm = pbm

        api_session._create_session()
        session_manager = vim_obj.service_content.sessionManager
        vim_obj.Login.assert_called_once_with(
            session_manager, userName=VMwareAPISessionTest.USERNAME,
            password=VMwareAPISessionTest.PASSWORD, locale='en')
        self.assertFalse(vim_obj.TerminateSession.called)
        self.assertEqual(session.key, api_session._session_id)
        pbm.set_soap_cookie.assert_called_once_with(cookie)

    def test_create_session_with_existing_inactive_session(self):
        old_session_key = '12345'
        new_session_key = '67890'
        session = mock.Mock()
        session.key = new_session_key
        api_session = self._create_api_session(False)
        api_session._session_id = old_session_key
        api_session._session_username = api_session._server_username
        vim_obj = api_session.vim
        vim_obj.Login.return_value = session
        vim_obj.SessionIsActive.return_value = False

        api_session._create_session()
        session_manager = vim_obj.service_content.sessionManager
        vim_obj.SessionIsActive.assert_called_once_with(
            session_manager, sessionID=old_session_key,
            userName=VMwareAPISessionTest.USERNAME)
        vim_obj.Login.assert_called_once_with(
            session_manager, userName=VMwareAPISessionTest.USERNAME,
            password=VMwareAPISessionTest.PASSWORD, locale='en')
        self.assertEqual(new_session_key, api_session._session_id)

    def test_create_session_with_existing_active_session(self):
        old_session_key = '12345'
        api_session = self._create_api_session(False)
        api_session._session_id = old_session_key
        api_session._session_username = api_session._server_username
        vim_obj = api_session.vim
        vim_obj.SessionIsActive.return_value = True

        api_session._create_session()
        session_manager = vim_obj.service_content.sessionManager
        vim_obj.SessionIsActive.assert_called_once_with(
            session_manager, sessionID=old_session_key,
            userName=VMwareAPISessionTest.USERNAME)
        self.assertFalse(vim_obj.Login.called)
        self.assertEqual(old_session_key, api_session._session_id)

    def test_invoke_api(self):
        api_session = self._create_api_session(True)
        response = mock.Mock()

        def api(*args, **kwargs):
            return response

        module = mock.Mock()
        module.api = api
        ret = api_session.invoke_api(module, 'api')
        self.assertEqual(response, ret)

    def test_logout_with_exception(self):
        session = mock.Mock()
        session.key = "12345"
        api_session = self._create_api_session(False)
        vim_obj = api_session.vim
        vim_obj.Login.return_value = session
        vim_obj.Logout.side_effect = exceptions.VimFaultException([], None)
        api_session._create_session()
        api_session.logout()
        self.assertEqual("12345", api_session._session_id)

    def test_logout_no_session(self):
        api_session = self._create_api_session(False)
        vim_obj = api_session.vim
        api_session.logout()
        self.assertEqual(0, vim_obj.Logout.call_count)

    def test_logout_calls_vim_logout(self):
        session = mock.Mock()
        session.key = "12345"
        api_session = self._create_api_session(False)
        vim_obj = api_session.vim
        vim_obj.Login.return_value = session
        vim_obj.Logout.return_value = None

        api_session._create_session()
        session_manager = vim_obj.service_content.sessionManager
        vim_obj.Login.assert_called_once_with(
            session_manager, userName=VMwareAPISessionTest.USERNAME,
            password=VMwareAPISessionTest.PASSWORD, locale='en')
        api_session.logout()
        vim_obj.Logout.assert_called_once_with(
            session_manager)
        self.assertIsNone(api_session._session_id)

    def test_invoke_api_with_expected_exception(self):
        api_session = self._create_api_session(True)
        api_session._create_session = mock.Mock()
        vim_obj = api_session.vim
        vim_obj.SessionIsActive.return_value = False
        ret = mock.Mock()
        responses = [exceptions.VimConnectionException(None), ret]

        def api(*args, **kwargs):
            response = responses.pop(0)
            if isinstance(response, Exception):
                raise response
            return response

        module = mock.Mock()
        module.api = api
        with mock.patch.object(greenthread, 'sleep'):
            self.assertEqual(ret, api_session.invoke_api(module, 'api'))
        api_session._create_session.assert_called_once_with()

    def test_invoke_api_not_recreate_session(self):
        api_session = self._create_api_session(True)
        api_session._create_session = mock.Mock()
        vim_obj = api_session.vim
        vim_obj.SessionIsActive.return_value = True
        ret = mock.Mock()
        responses = [exceptions.VimConnectionException(None), ret]

        def api(*args, **kwargs):
            response = responses.pop(0)
            if isinstance(response, Exception):
                raise response
            return response

        module = mock.Mock()
        module.api = api
        with mock.patch.object(greenthread, 'sleep'):
            self.assertEqual(ret, api_session.invoke_api(module, 'api'))
        self.assertFalse(api_session._create_session.called)

    def test_invoke_api_with_vim_fault_exception(self):
        api_session = self._create_api_session(True)

        def api(*args, **kwargs):
            raise exceptions.VimFaultException([], None)

        module = mock.Mock()
        module.api = api
        self.assertRaises(exceptions.VimFaultException,
                          api_session.invoke_api,
                          module,
                          'api')

    def test_invoke_api_with_vim_fault_exception_details(self):
        api_session = self._create_api_session(True)
        fault_string = 'Invalid property.'
        fault_list = [exceptions.INVALID_PROPERTY]
        details = {u'name': suds.sax.text.Text(u'фира')}

        module = mock.Mock()
        module.api.side_effect = exceptions.VimFaultException(fault_list,
                                                              fault_string,
                                                              details=details)
        e = self.assertRaises(exceptions.InvalidPropertyException,
                              api_session.invoke_api,
                              module,
                              'api')
        details_str = u"{'name': 'фира'}"
        expected_str = "%s\nFaults: %s\nDetails: %s" % (fault_string,
                                                        fault_list,
                                                        details_str)
        self.assertEqual(expected_str, str(e))
        self.assertEqual(details, e.details)

    def test_invoke_api_with_empty_response(self):
        api_session = self._create_api_session(True)
        vim_obj = api_session.vim
        vim_obj.SessionIsActive.return_value = True

        def api(*args, **kwargs):
            raise exceptions.VimFaultException(
                [exceptions.NOT_AUTHENTICATED], None)

        module = mock.Mock()
        module.api = api
        ret = api_session.invoke_api(module, 'api')
        self.assertEqual([], ret)
        vim_obj.SessionIsActive.assert_called_once_with(
            vim_obj.service_content.sessionManager,
            sessionID=api_session._session_id,
            userName=api_session._session_username)

    def test_invoke_api_with_stale_session(self):
        api_session = self._create_api_session(True)
        api_session._create_session = mock.Mock()
        vim_obj = api_session.vim
        vim_obj.SessionIsActive.return_value = False
        result = mock.Mock()
        responses = [exceptions.VimFaultException(
            [exceptions.NOT_AUTHENTICATED], None), result]

        def api(*args, **kwargs):
            response = responses.pop(0)
            if isinstance(response, Exception):
                raise response
            return response

        module = mock.Mock()
        module.api = api
        with mock.patch.object(greenthread, 'sleep'):
            ret = api_session.invoke_api(module, 'api')
        self.assertEqual(result, ret)
        vim_obj.SessionIsActive.assert_called_once_with(
            vim_obj.service_content.sessionManager,
            sessionID=api_session._session_id,
            userName=api_session._session_username)
        api_session._create_session.assert_called_once_with()

    def test_invoke_api_with_unknown_fault(self):
        api_session = self._create_api_session(True)
        fault_list = ['NotAFile']

        module = mock.Mock()
        module.api.side_effect = exceptions.VimFaultException(fault_list,
                                                              'Not a file.')
        ex = self.assertRaises(exceptions.VimFaultException,
                               api_session.invoke_api,
                               module,
                               'api')
        self.assertEqual(fault_list, ex.fault_list)

    @mock.patch.object(context, 'get_current')
    def test_wait_for_task(self, mock_curr_ctx):
        ctx = mock.Mock()
        mock_curr_ctx.return_value = ctx
        api_session = self._create_api_session(True)
        task_info_list = [('queued', 0), ('running', 40), ('success', 100)]
        task_info_list_size = len(task_info_list)

        def invoke_api_side_effect(module, method, *args, **kwargs):
            (state, progress) = task_info_list.pop(0)
            task_info = mock.Mock()
            task_info.progress = progress
            task_info.queueTime = datetime(2016, 12, 6, 15, 29, 43, 79060)
            task_info.completeTime = datetime(2016, 12, 6, 15, 29, 50, 79060)
            task_info.state = state
            return task_info

        api_session.invoke_api = mock.Mock(side_effect=invoke_api_side_effect)
        task = mock.Mock()
        with mock.patch.object(greenthread, 'sleep'):
            ret = api_session.wait_for_task(task)
            self.assertEqual('success', ret.state)
            self.assertEqual(100, ret.progress)
        api_session.invoke_api.assert_called_with(vim_util,
                                                  'get_object_property',
                                                  api_session.vim, task,
                                                  'info',
                                                  skip_op_id=True)
        self.assertEqual(task_info_list_size,
                         api_session.invoke_api.call_count)
        mock_curr_ctx.assert_called_once()
        self.assertEqual(3, ctx.update_store.call_count)

    @mock.patch.object(context, 'get_current', return_value=None)
    def test_wait_for_task_no_ctx(self, mock_curr_ctx):
        api_session = self._create_api_session(True)
        task_info_list = [('queued', 0), ('running', 40), ('success', 100)]
        task_info_list_size = len(task_info_list)

        def invoke_api_side_effect(module, method, *args, **kwargs):
            (state, progress) = task_info_list.pop(0)
            task_info = mock.Mock()
            task_info.progress = progress
            task_info.queueTime = datetime(2016, 12, 6, 15, 29, 43, 79060)
            task_info.completeTime = datetime(2016, 12, 6, 15, 29, 50, 79060)
            task_info.state = state
            return task_info

        api_session.invoke_api = mock.Mock(side_effect=invoke_api_side_effect)
        task = mock.Mock()
        with mock.patch.object(greenthread, 'sleep'):
            ret = api_session.wait_for_task(task)
            self.assertEqual('success', ret.state)
            self.assertEqual(100, ret.progress)
        api_session.invoke_api.assert_called_with(vim_util,
                                                  'get_object_property',
                                                  api_session.vim, task,
                                                  'info',
                                                  skip_op_id=True)
        self.assertEqual(task_info_list_size,
                         api_session.invoke_api.call_count)
        mock_curr_ctx.assert_called_once()

    @mock.patch.object(context, 'get_current')
    def test_wait_for_task_with_error_state(self, mock_curr_ctx):
        api_session = self._create_api_session(True)
        task_info_list = [('queued', 0), ('running', 40), ('error', -1)]
        task_info_list_size = len(task_info_list)

        def invoke_api_side_effect(module, method, *args, **kwargs):
            (state, progress) = task_info_list.pop(0)
            task_info = mock.Mock()
            task_info.progress = progress
            task_info.state = state
            return task_info

        api_session.invoke_api = mock.Mock(side_effect=invoke_api_side_effect)
        task = mock.Mock()
        with mock.patch.object(greenthread, 'sleep'):
            self.assertRaises(exceptions.VimFaultException,
                              api_session.wait_for_task,
                              task)
        api_session.invoke_api.assert_called_with(vim_util,
                                                  'get_object_property',
                                                  api_session.vim, task,
                                                  'info',
                                                  skip_op_id=True)
        self.assertEqual(task_info_list_size,
                         api_session.invoke_api.call_count)
        mock_curr_ctx.assert_called_once()

    @mock.patch.object(context, 'get_current')
    def test_wait_for_task_with_invoke_api_exception(self, mock_curr_ctx):
        api_session = self._create_api_session(True)
        api_session.invoke_api = mock.Mock(
            side_effect=exceptions.VimException(None))
        task = mock.Mock()
        with mock.patch.object(greenthread, 'sleep'):
            self.assertRaises(exceptions.VimException,
                              api_session.wait_for_task,
                              task)
        api_session.invoke_api.assert_called_once_with(vim_util,
                                                       'get_object_property',
                                                       api_session.vim, task,
                                                       'info',
                                                       skip_op_id=True)
        mock_curr_ctx.assert_called_once()

    def test_wait_for_lease_ready(self):
        api_session = self._create_api_session(True)
        lease_states = ['initializing', 'ready']
        num_states = len(lease_states)

        def invoke_api_side_effect(module, method, *args, **kwargs):
            return lease_states.pop(0)

        api_session.invoke_api = mock.Mock(side_effect=invoke_api_side_effect)
        lease = mock.Mock()
        with mock.patch.object(greenthread, 'sleep'):
            api_session.wait_for_lease_ready(lease)
        api_session.invoke_api.assert_called_with(vim_util,
                                                  'get_object_property',
                                                  api_session.vim, lease,
                                                  'state',
                                                  skip_op_id=True)
        self.assertEqual(num_states, api_session.invoke_api.call_count)

    def test_wait_for_lease_ready_with_error_state(self):
        api_session = self._create_api_session(True)
        responses = ['initializing', 'error', 'error_msg']

        def invoke_api_side_effect(module, method, *args, **kwargs):
            return responses.pop(0)

        api_session.invoke_api = mock.Mock(side_effect=invoke_api_side_effect)
        lease = mock.Mock()
        with mock.patch.object(greenthread, 'sleep'):
            self.assertRaises(exceptions.VimException,
                              api_session.wait_for_lease_ready,
                              lease)
        exp_calls = [mock.call(vim_util, 'get_object_property',
                               api_session.vim, lease, 'state',
                               skip_op_id=True)] * 2
        exp_calls.append(mock.call(vim_util, 'get_object_property',
                                   api_session.vim, lease, 'error'))
        self.assertEqual(exp_calls, api_session.invoke_api.call_args_list)

    def test_wait_for_lease_ready_with_unknown_state(self):
        api_session = self._create_api_session(True)

        def invoke_api_side_effect(module, method, *args, **kwargs):
            return 'unknown'

        api_session.invoke_api = mock.Mock(side_effect=invoke_api_side_effect)
        lease = mock.Mock()
        self.assertRaises(exceptions.VimException,
                          api_session.wait_for_lease_ready,
                          lease)
        api_session.invoke_api.assert_called_once_with(vim_util,
                                                       'get_object_property',
                                                       api_session.vim,
                                                       lease, 'state',
                                                       skip_op_id=True)

    def test_wait_for_lease_ready_with_invoke_api_exception(self):
        api_session = self._create_api_session(True)
        api_session.invoke_api = mock.Mock(
            side_effect=exceptions.VimException(None))
        lease = mock.Mock()
        self.assertRaises(exceptions.VimException,
                          api_session.wait_for_lease_ready,
                          lease)
        api_session.invoke_api.assert_called_once_with(
            vim_util, 'get_object_property', api_session.vim, lease,
            'state', skip_op_id=True)

    def _poll_task_well_known_exceptions(self, fault,
                                         expected_exception):
        api_session = self._create_api_session(False)

        def fake_invoke_api(self, module, method, *args, **kwargs):
            task_info = mock.Mock()
            task_info.progress = -1
            task_info.state = 'error'
            error = mock.Mock()
            error.localizedMessage = "Error message"
            error_fault = mock.Mock()
            error_fault.__class__.__name__ = fault
            error.fault = error_fault
            task_info.error = error
            return task_info

        with (
            mock.patch.object(api_session, 'invoke_api', fake_invoke_api)
        ):
            fake_task = vim_util.get_moref('Task', 'task-1')
            ctx = mock.Mock()
            self.assertRaises(expected_exception,
                              api_session._poll_task,
                              fake_task,
                              ctx)

    def test_poll_task_well_known_exceptions(self):
        for k, v in exceptions._fault_classes_registry.items():
            self._poll_task_well_known_exceptions(k, v)

    def test_poll_task_unknown_exception(self):
        _unknown_exceptions = {
            'NotAFile': exceptions.VimFaultException,
            'RuntimeFault': exceptions.VimFaultException
        }

        for k, v in _unknown_exceptions.items():
            self._poll_task_well_known_exceptions(k, v)

    def test_update_pbm_wsdl_loc(self):
        session = mock.Mock()
        session.key = "12345"
        api_session = self._create_api_session(False)
        self.assertIsNone(api_session._pbm_wsdl_loc)
        api_session.pbm_wsdl_loc_set('fake_wsdl')
        self.assertEqual('fake_wsdl', api_session._pbm_wsdl_loc)