cinder/cinder/tests/test_vmware_api.py

# 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 eventlet import greenthread
import mock

from cinder import test
from cinder.volume.drivers.vmware import api
from cinder.volume.drivers.vmware import error_util


class RetryTest(test.TestCase):
    """Tests for retry decorator class."""

    def test_retry(self):
        result = "RESULT"

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

        self.assertEqual(result, func())

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

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

    def test_retry_with_expected_exceptions(self):
        result = "RESULT"
        responses = [error_util.SessionOverLoadException(None),
                     error_util.SessionOverLoadException(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.Retry(10, sleep_time_incr, 10,
                          (error_util.SessionOverLoadException,))
        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 = [error_util.SessionOverLoadException(None),
                     error_util.SessionOverLoadException(None),
                     error_util.SessionOverLoadException(None)]

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

        retry = api.Retry(2, 0, 0, (error_util.SessionOverLoadException,))
        self.assertRaises(error_util.SessionOverLoadException, retry(func))
        self.assertTrue(retry._retry_count == 2)

    def test_retry_with_unexpected_exception(self):

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

        retry = api.Retry()
        self.assertRaises(error_util.VimException, retry(func))
        self.assertTrue(retry._retry_count == 0)


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

    SERVER_IP = '10.1.2.3'
    USERNAME = 'admin'
    PASSWORD = 'password'

    def setUp(self):
        super(VMwareAPISessionTest, self).setUp()
        patcher = mock.patch('cinder.volume.drivers.vmware.vim.Vim')
        self.addCleanup(patcher.stop)
        self.VimMock = patcher.start()
        self.VimMock.side_effect = lambda *args, **kw: 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)

    def test_create_session(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
        pbm_client = mock.Mock()
        api_session._pbm = pbm_client

        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)
        self.assertFalse(vim_obj.TerminateSession.called)
        self.assertEqual(session.key, api_session._session_id)
        pbm_client.set_cookie.assert_called_once_with()

    def test_create_session_with_existing_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
        vim_obj = api_session.vim
        vim_obj.Login.return_value = session

        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)
        vim_obj.TerminateSession.assert_called_once_with(
            session_manager, sessionId=[old_session_key])
        self.assertEqual(new_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_invoke_api_with_expected_exception(self):
        api_session = self._create_api_session(True)
        ret = mock.Mock()
        responses = [error_util.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'))

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

        def api(*args, **kwargs):
            raise error_util.VimFaultException([], "error")

        module = mock.Mock()
        module.api = api
        self.assertRaises(error_util.VimFaultException,
                          lambda: api_session.invoke_api(module, 'api'))

    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 error_util.VimFaultException(
                [error_util.NOT_AUTHENTICATED], "error")

        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 = [error_util.VimFaultException(
            [error_util.NOT_AUTHENTICATED], "error"), result]

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

        module = mock.Mock()
        module.api = api
        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_session_is_active_error(self):
        api_session = self._create_api_session(True)
        api_session.create_session = mock.Mock()
        vim_obj = api_session.vim
        vim_obj.SessionIsActive.side_effect = error_util.VimFaultException(
            None, None)
        result = mock.Mock()
        responses = [error_util.VimFaultException(
            [error_util.NOT_AUTHENTICATED], "error"), result]

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

        module = mock.Mock()
        module.api = api
        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()