# (c) Copyright 2015 Hewlett-Packard Development Company, L.P.
#
#    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 builtins
import functools
import io
import time
from unittest import mock

from castellan.common import objects as castellan_objects
import ddt

from os_brick import exception
from os_brick.tests import base
from os_brick import utils


class WrongException(exception.BrickException):
    pass


@ddt.ddt
class TestUtils(base.TestCase):
    @ddt.data(('1024', 1024), ('junk', None), ('2048\n', 2048))
    @ddt.unpack
    def test_get_device_size(self, cmd_out, expected):
        mock_execute = mock.Mock()
        mock_execute._execute.return_value = (cmd_out, None)

        device = '/dev/fake'
        ret_size = utils.get_device_size(mock_execute, device)
        self.assertEqual(expected, ret_size)
        mock_execute._execute.assert_called_once_with(
            'blockdev', '--getsize64', device,
            run_as_root=True, root_helper=mock_execute._root_helper)

    @mock.patch.object(builtins, 'open')
    def test_get_nvme_host_id_file_available(self, mock_open):
        mock_open.return_value.__enter__.return_value.read.return_value = (
            'uuid\n')
        result = utils.get_nvme_host_id(mock.sentinel.uuid)
        mock_open.assert_called_once_with('/etc/nvme/hostid', 'r')
        self.assertEqual('uuid', result)

    @mock.patch.object(utils.priv_nvme, 'create_hostid')
    @mock.patch.object(builtins, 'open')
    def test_get_nvme_host_id_io_err(self, mock_open, mock_create):
        mock_create.return_value = mock.sentinel.uuid_return
        mock_open.side_effect = IOError()
        result = utils.get_nvme_host_id(mock.sentinel.uuid)
        mock_open.assert_called_once_with('/etc/nvme/hostid', 'r')
        mock_create.assert_called_once_with(mock.sentinel.uuid)
        self.assertEqual(mock.sentinel.uuid_return, result)

    @mock.patch('uuid.uuid4')
    @mock.patch.object(utils.priv_nvme, 'create_hostid')
    @mock.patch.object(builtins, 'open')
    def test_get_nvme_host_id_io_err_no_uuid(self, mock_open, mock_create,
                                             mock_uuid):
        mock_create.return_value = mock.sentinel.uuid_return
        mock_open.side_effect = IOError()
        result = utils.get_nvme_host_id(None)
        mock_open.assert_called_once_with('/etc/nvme/hostid', 'r')
        mock_create.assert_called_once_with(str(mock_uuid.return_value))
        self.assertEqual(mock.sentinel.uuid_return, result)

    @mock.patch.object(utils.priv_nvme, 'create_hostid')
    @mock.patch.object(builtins, 'open')
    def test_get_nvme_host_id_err(self, mock_open, mock_create):
        mock_open.side_effect = Exception()
        result = utils.get_nvme_host_id(None)
        mock_open.assert_called_once_with('/etc/nvme/hostid', 'r')
        mock_create.assert_not_called()
        self.assertIsNone(result)

    @ddt.data(('fake_info', True), (None, False))
    @ddt.unpack
    def test_check_valid_device(self, fake_info, expected):
        mock_execute = mock.Mock()
        mock_execute._execute.return_value = ('fake_out', fake_info)

        fake_path = '/dev/fake'
        is_valid = utils.check_valid_device(mock_execute, fake_path)
        self.assertEqual(expected, is_valid)
        mock_execute._execute.assert_called_once_with(
            'dd', 'if=/dev/fake', 'of=/dev/null', 'count=1',
            run_as_root=True, root_helper=mock_execute._root_helper)

    def test_check_valid_device_error(self):
        mock_execute = mock.Mock()
        p_exception = utils.processutils.ProcessExecutionError
        mock_execute._execute.side_effect = p_exception

        fake_path = '/dev/fake'
        is_valid = utils.check_valid_device(mock_execute, fake_path)
        self.assertEqual(False, is_valid)
        mock_execute._execute.assert_called_once_with(
            'dd', 'if=/dev/fake', 'of=/dev/null', 'count=1',
            run_as_root=True, root_helper=mock_execute._root_helper)

    @mock.patch('binascii.hexlify')
    @ddt.data(
        castellan_objects.passphrase.Passphrase(b'test-passphrase'),
        castellan_objects.symmetric_key.SymmetricKey('AES',
                                                     mock.sentinel.bitlength,
                                                     mock.sentinel.key),
        castellan_objects.opaque_data.OpaqueData(mock.sentinel.key),
        castellan_objects.private_key.PrivateKey('RSA',
                                                 mock.sentinel.bitlength,
                                                 mock.sentinel.key),
        castellan_objects.public_key.PublicKey('RSA',
                                               mock.sentinel.bitlength,
                                               mock.sentinel.key),
        castellan_objects.x_509.X509(mock.sentinel.key)
    )
    def test_get_passphrase_from_secret(self, secret, mock_hexlify):
        """Test proper passphrase processing of different secret types."""
        if secret.managed_type() == 'passphrase':
            passphrase = utils.get_passphrase_from_secret(secret)
            mock_hexlify.assert_not_called()
            self.assertEqual('test-passphrase', passphrase)
        else:
            hexlified_bytes = mock.MagicMock()
            hexlified_bytes.decode.return_value = mock.sentinel.passphrase
            mock_hexlify.return_value = hexlified_bytes
            passphrase = utils.get_passphrase_from_secret(secret)
            mock_hexlify.assert_called_once_with(mock.sentinel.key)
            self.assertEqual(mock.sentinel.passphrase, passphrase)


class TestRetryDecorator(base.TestCase):

    def test_no_retry_required(self):
        self.counter = 0

        with mock.patch.object(utils, '_time_sleep') as mock_sleep:
            @utils.retry(exception.VolumeDeviceNotFound,
                         interval=2,
                         retries=3,
                         backoff_rate=2)
            def succeeds():
                self.counter += 1
                return 'success'

            ret = succeeds()
            self.assertFalse(mock_sleep.called)
            self.assertEqual('success', ret)
            self.assertEqual(1, self.counter)

    def test_retries_once(self):
        self.counter = 0
        interval = 2
        backoff_rate = 2
        retries = 3

        with mock.patch.object(utils, '_time_sleep') as mock_sleep:
            @utils.retry(exception.VolumeDeviceNotFound,
                         interval,
                         retries,
                         backoff_rate)
            def fails_once():
                self.counter += 1
                if self.counter < 2:
                    raise exception.VolumeDeviceNotFound(device='fake')
                else:
                    return 'success'

            ret = fails_once()
            self.assertEqual('success', ret)
            self.assertEqual(2, self.counter)
            self.assertEqual(1, mock_sleep.call_count)
            mock_sleep.assert_called_with(interval)

    def test_limit_is_reached(self):
        self.counter = 0
        retries = 3
        interval = 2
        backoff_rate = 4

        with mock.patch.object(utils, '_time_sleep') as mock_sleep:
            @utils.retry(exception.VolumeDeviceNotFound,
                         interval,
                         retries,
                         backoff_rate)
            def always_fails():
                self.counter += 1
                raise exception.VolumeDeviceNotFound(device='fake')

            self.assertRaises(exception.VolumeDeviceNotFound,
                              always_fails)
            self.assertEqual(retries, self.counter)

            expected_sleep_arg = []

            for i in range(retries):
                if i > 0:
                    interval *= (backoff_rate ** (i - 1))
                    expected_sleep_arg.append(float(interval))

            mock_sleep.assert_has_calls(
                list(map(mock.call, expected_sleep_arg)))

    def test_wrong_exception_no_retry(self):

        with mock.patch.object(utils, '_time_sleep') as mock_sleep:
            @utils.retry(exception.VolumeDeviceNotFound)
            def raise_unexpected_error():
                raise WrongException("wrong exception")

            self.assertRaises(WrongException, raise_unexpected_error)
            self.assertFalse(mock_sleep.called)

    @mock.patch('tenacity.nap.sleep')
    def test_retry_exit_code(self, sleep_mock):
        exit_code = 5
        exception = utils.processutils.ProcessExecutionError

        @utils.retry(retry=utils.retry_if_exit_code, retry_param=exit_code)
        def raise_retriable_exit_code():
            raise exception(exit_code=exit_code)
        self.assertRaises(exception, raise_retriable_exit_code)
        self.assertEqual(0, sleep_mock.call_count)

    @mock.patch('tenacity.nap.sleep')
    def test_retry_exit_code_non_retriable(self, sleep_mock):
        exit_code = 5
        exception = utils.processutils.ProcessExecutionError

        @utils.retry(retry=utils.retry_if_exit_code, retry_param=exit_code)
        def raise_non_retriable_exit_code():
            raise exception(exit_code=exit_code + 1)
        self.assertRaises(exception, raise_non_retriable_exit_code)
        sleep_mock.assert_not_called()


class LogTracingTestCase(base.TestCase):
    """Test out the log tracing."""

    def test_utils_trace_method_default_logger(self):
        mock_log = self.mock_object(utils, 'LOG')

        @utils.trace
        def _trace_test_method_custom_logger(*args, **kwargs):
            return 'OK'

        result = _trace_test_method_custom_logger()

        self.assertEqual('OK', result)
        self.assertEqual(2, mock_log.debug.call_count)

    def test_utils_trace_method_inner_decorator(self):
        mock_logging = self.mock_object(utils, 'logging')
        mock_log = mock.Mock()
        mock_log.isEnabledFor = lambda x: True
        mock_logging.getLogger = mock.Mock(return_value=mock_log)

        def _test_decorator(f):
            def blah(*args, **kwargs):
                return f(*args, **kwargs)
            return blah

        @_test_decorator
        @utils.trace
        def _trace_test_method(*args, **kwargs):
            return 'OK'

        result = _trace_test_method(self)

        self.assertEqual('OK', result)
        self.assertEqual(2, mock_log.debug.call_count)
        # Ensure the correct function name was logged
        for call in mock_log.debug.call_args_list:
            self.assertIn('_trace_test_method', str(call))
            self.assertNotIn('blah', str(call))

    def test_utils_trace_method_outer_decorator(self):
        mock_logging = self.mock_object(utils, 'logging')
        mock_log = mock.Mock()
        mock_log.isEnabledFor = lambda x: True
        mock_logging.getLogger = mock.Mock(return_value=mock_log)

        def _test_decorator(f):
            def blah(*args, **kwargs):
                return f(*args, **kwargs)
            return blah

        @utils.trace
        @_test_decorator
        def _trace_test_method(*args, **kwargs):
            return 'OK'

        result = _trace_test_method(self)

        self.assertEqual('OK', result)
        self.assertEqual(2, mock_log.debug.call_count)
        # Ensure the incorrect function name was logged
        for call in mock_log.debug.call_args_list:
            self.assertNotIn('_trace_test_method', str(call))
            self.assertIn('blah', str(call))

    def test_utils_trace_method_outer_decorator_with_functools(self):
        mock_log = mock.Mock()
        mock_log.isEnabledFor = lambda x: True
        self.mock_object(utils.logging, 'getLogger', mock_log)
        mock_log = self.mock_object(utils, 'LOG')

        def _test_decorator(f):
            @functools.wraps(f)
            def wraps(*args, **kwargs):
                return f(*args, **kwargs)
            return wraps

        @utils.trace
        @_test_decorator
        def _trace_test_method(*args, **kwargs):
            return 'OK'

        result = _trace_test_method()

        self.assertEqual('OK', result)
        self.assertEqual(2, mock_log.debug.call_count)
        # Ensure the incorrect function name was logged
        for call in mock_log.debug.call_args_list:
            self.assertIn('_trace_test_method', str(call))
            self.assertNotIn('wraps', str(call))

    def test_utils_trace_method_with_exception(self):
        self.LOG = self.mock_object(utils, 'LOG')

        @utils.trace
        def _trace_test_method(*args, **kwargs):
            raise exception.VolumeDeviceNotFound('test message')

        self.assertRaises(exception.VolumeDeviceNotFound, _trace_test_method)

        exception_log = self.LOG.debug.call_args_list[1]
        self.assertIn('exception', str(exception_log))
        self.assertIn('test message', str(exception_log))

    def test_utils_trace_method_with_time(self):
        mock_logging = self.mock_object(utils, 'logging')
        mock_log = mock.Mock()
        mock_log.isEnabledFor = lambda x: True
        mock_logging.getLogger = mock.Mock(return_value=mock_log)

        mock_time = mock.Mock(side_effect=[3.1, 6])
        self.mock_object(time, 'time', mock_time)

        @utils.trace
        def _trace_test_method(*args, **kwargs):
            return 'OK'

        result = _trace_test_method(self)

        self.assertEqual('OK', result)
        return_log = mock_log.debug.call_args_list[1]
        self.assertIn('2900', str(return_log))

    def test_utils_trace_method_with_password_dict(self):
        mock_logging = self.mock_object(utils, 'logging')
        mock_log = mock.Mock()
        mock_log.isEnabledFor = lambda x: True
        mock_logging.getLogger = mock.Mock(return_value=mock_log)

        @utils.trace
        def _trace_test_method(*args, **kwargs):
            return {'something': 'test',
                    'password': 'Now you see me'}

        result = _trace_test_method(self)
        expected_unmasked_dict = {'something': 'test',
                                  'password': 'Now you see me'}

        self.assertEqual(expected_unmasked_dict, result)
        self.assertEqual(2, mock_log.debug.call_count)
        self.assertIn("'password': '***'",
                      str(mock_log.debug.call_args_list[1]))

    def test_utils_trace_method_with_password_str(self):
        mock_logging = self.mock_object(utils, 'logging')
        mock_log = mock.Mock()
        mock_log.isEnabledFor = lambda x: True
        mock_logging.getLogger = mock.Mock(return_value=mock_log)

        @utils.trace
        def _trace_test_method(*args, **kwargs):
            return "'adminPass': 'Now you see me'"

        result = _trace_test_method(self)
        expected_unmasked_str = "'adminPass': 'Now you see me'"

        self.assertEqual(expected_unmasked_str, result)
        self.assertEqual(2, mock_log.debug.call_count)
        self.assertIn("'adminPass': '***'",
                      str(mock_log.debug.call_args_list[1]))

    def test_utils_trace_method_with_password_in_formal_params(self):
        mock_logging = self.mock_object(utils, 'logging')
        mock_log = mock.Mock()
        mock_log.isEnabledFor = lambda x: True
        mock_logging.getLogger = mock.Mock(return_value=mock_log)

        @utils.trace
        def _trace_test_method(*args, **kwargs):
            self.assertEqual('verybadpass',
                             kwargs['connection']['data']['auth_password'])
            pass

        connector_properties = {
            'data': {
                'auth_password': 'verybadpass'
            }
        }
        _trace_test_method(self, connection=connector_properties)

        self.assertEqual(2, mock_log.debug.call_count)
        self.assertIn("'auth_password': '***'",
                      str(mock_log.debug.call_args_list[0]))


@ddt.ddt
class GetDevPathTestCase(base.TestCase):
    """Test the get_dev_path method."""
    @ddt.data({'con_props': {}, 'dev_info': {'path': '/dev/sda'}},
              {'con_props': {}, 'dev_info': {'path': b'/dev/sda'}},
              {'con_props': None, 'dev_info': {'path': '/dev/sda'}},
              {'con_props': None, 'dev_info': {'path': b'/dev/sda'}},
              {'con_props': {'device_path': b'/dev/sdb'},
               'dev_info': {'path': '/dev/sda'}},
              {'con_props': {'device_path': '/dev/sdb'},
               'dev_info': {'path': b'/dev/sda'}})
    @ddt.unpack
    def test_get_dev_path_device_info(self, con_props, dev_info):
        self.assertEqual('/dev/sda', utils.get_dev_path(con_props, dev_info))

    @ddt.data({'con_props': {'device_path': '/dev/sda'},
               'dev_info': {'path': None}},
              {'con_props': {'device_path': b'/dev/sda'},
               'dev_info': {'path': None}},
              {'con_props': {'device_path': '/dev/sda'},
               'dev_info': {'path': ''}},
              {'con_props': {'device_path': b'/dev/sda'},
               'dev_info': {'path': ''}},
              {'con_props': {'device_path': '/dev/sda'},
               'dev_info': {}},
              {'con_props': {'device_path': b'/dev/sda'},
               'dev_info': {}},
              {'con_props': {'device_path': '/dev/sda'},
               'dev_info': None},
              {'con_props': {'device_path': b'/dev/sda'},
               'dev_info': None})
    @ddt.unpack
    def test_get_dev_path_conn_props(self, con_props, dev_info):
        self.assertEqual('/dev/sda', utils.get_dev_path(con_props, dev_info))

    @ddt.data({'con_props': {'device_path': ''}, 'dev_info': {'path': None}},
              {'con_props': {'device_path': None}, 'dev_info': {'path': ''}},
              {'con_props': {}, 'dev_info': {}},
              {'con_props': {}, 'dev_info': None})
    @ddt.unpack
    def test_get_dev_path_no_path(self, con_props, dev_info):
        self.assertEqual('', utils.get_dev_path(con_props, dev_info))


@ddt.ddt
class ConnectionPropertiesDecoratorsTestCase(base.TestCase):
    def test__symlink_name_from_device_path(self):
        """Get symlink for non replicated device."""
        dev_name = '/dev/nvme0n1'
        res = utils._symlink_name_from_device_path(dev_name)
        self.assertEqual('/dev/disk/by-id/os-brick+dev+nvme0n1', res)

    def test__symlink_name_from_device_path_raid(self):
        """Get symlink for replicated device."""
        dev_name = '/dev/md/alias'
        res = utils._symlink_name_from_device_path(dev_name)
        self.assertEqual('/dev/disk/by-id/os-brick+dev+md+alias', res)

    def test__device_path_from_symlink(self):
        """Get device name for non replicated symlink."""
        symlink = '/dev/disk/by-id/os-brick+dev+nvme0n1'
        res = utils._device_path_from_symlink(symlink)
        self.assertEqual('/dev/nvme0n1', res)

    def test__device_path_from_symlink_raid(self):
        """Get device name for replicated symlink."""
        symlink = '/dev/disk/by-id/os-brick+dev+md+alias'
        res = utils._device_path_from_symlink(symlink)
        self.assertEqual('/dev/md/alias', res)

    def test__device_path_from_symlink_file_handle(self):
        """Get device name for a file handle (eg: RBD)."""
        handle = io.StringIO()
        res = utils._device_path_from_symlink(handle)
        self.assertEqual(handle, res)

    @ddt.data(({}, {'type': 'block', 'path': '/dev/sda'}),
              ({'encrypted': False}, {'type': 'block', 'path': '/dev/sda'}),
              ({'encrypted': False}, {'type': 'block', 'path': b'/dev/sda'}),
              ({'encrypted': True}, {'type': 'block', 'path': io.StringIO()}))
    @ddt.unpack
    @mock.patch('os_brick.utils._symlink_name_from_device_path')
    @mock.patch('os.path.realpath')
    @mock.patch('os_brick.privileged.rootwrap.link_root')
    def test_connect_volume_prepare_result_non_encrypted(
            self, conn_props, result, mock_link, mock_path, mock_get_symlink):
        """Test decorator for non encrypted devices or non host devices."""
        testing_self = mock.Mock()
        testing_self.connect_volume.return_value = result
        func = utils.connect_volume_prepare_result(testing_self.connect_volume)

        res = func(testing_self, conn_props)
        self.assertEqual(testing_self.connect_volume.return_value, res)

        testing_self.connect_volume.assert_called_once_with(testing_self,
                                                            conn_props)
        mock_path.assert_not_called()
        mock_get_symlink.assert_not_called()
        mock_link.assert_not_called()

    @ddt.data('/dev/md/alias', b'/dev/md/alias')
    @mock.patch('os_brick.utils._symlink_name_from_device_path')
    @mock.patch('os.path.realpath')
    @mock.patch('os_brick.privileged.rootwrap.link_root')
    def test_connect_volume_prepare_result_encrypted(
            self, connector_path, mock_link, mock_path, mock_get_symlink):
        """Test decorator for encrypted device."""
        real_device = '/dev/md-6'
        expected_symlink = '/dev/disk/by-id/os-brick_dev_md_alias'
        mock_path.return_value = real_device
        mock_get_symlink.return_value = expected_symlink
        testing_self = mock.Mock()
        testing_self.connect_volume.return_value = {'type': 'block',
                                                    'path': connector_path}
        conn_props = {'encrypted': True}
        func = utils.connect_volume_prepare_result(testing_self.connect_volume)

        res = func(testing_self, conn_props)
        self.assertEqual({'type': 'block', 'path': expected_symlink}, res)

        testing_self.connect_volume.assert_called_once_with(testing_self,
                                                            conn_props)
        expected_connector_path = utils.convert_str(connector_path)
        mock_get_symlink.assert_called_once_with(expected_connector_path)
        mock_link.assert_called_once_with(real_device, expected_symlink,
                                          force=True)

    @ddt.data({}, {'encrypted': False}, {'encrypted': True})
    @mock.patch('os_brick.utils._symlink_name_from_device_path')
    @mock.patch('os.path.realpath')
    @mock.patch('os_brick.privileged.rootwrap.link_root')
    def test_connect_volume_prepare_result_connect_fail(
            self, conn_props, mock_link, mock_path, mock_get_symlink):
        """Test decorator when decorated function fails."""
        testing_self = mock.Mock()
        testing_self.connect_volume.side_effect = ValueError

        func = utils.connect_volume_prepare_result(testing_self.connect_volume)
        self.assertRaises(ValueError, func, testing_self, conn_props)
        mock_link.assert_not_called()
        mock_path.assert_not_called()
        mock_get_symlink.assert_not_called()

    @mock.patch('os_brick.utils._symlink_name_from_device_path')
    @mock.patch('os.path.realpath')
    @mock.patch('os_brick.privileged.rootwrap.link_root')
    def test_connect_volume_prepare_result_symlink_fail(
            self, mock_link, mock_path, mock_get_symlink):
        """Test decorator for encrypted device failing on the symlink."""
        real_device = '/dev/md-6'
        connector_path = '/dev/md/alias'
        expected_symlink = '/dev/disk/by-id/os-brick_dev_md_alias'
        mock_path.return_value = real_device
        mock_get_symlink.return_value = expected_symlink
        testing_self = mock.Mock()
        connect_result = {'type': 'block', 'path': connector_path}
        mock_link.side_effect = ValueError

        testing_self.connect_volume.return_value = connect_result
        conn_props = {'encrypted': True}
        func = utils.connect_volume_prepare_result(testing_self.connect_volume)

        self.assertRaises(ValueError, func, testing_self, conn_props)

        testing_self.connect_volume.assert_called_once_with(testing_self,
                                                            conn_props)
        mock_get_symlink.assert_called_once_with(connector_path)
        mock_link.assert_called_once_with(real_device, expected_symlink,
                                          force=True)
        testing_self.disconnect_volume.assert_called_once_with(
            connect_result, force=True, ignore_errors=True)

    @ddt.data(({'device_path': '/dev/md/alias'}, {}),
              ({'device_path': '/dev/md/alias', 'encrypted': False}, None),
              ({'device_path': '/dev/md/alias'}, {'path': '/dev/md/alias'}),
              ({'device_path': '/dev/md/alias', 'encrypted': False},
               {'path': '/dev/md/alias'}),
              ({'device_path': io.StringIO(), 'encrypted': True}, None),
              ({'device_path': '/dev/disk/by-id/wwn-...', 'encrypted': True},
               None))
    @ddt.unpack
    @mock.patch('os_brick.utils._device_path_from_symlink')
    @mock.patch('os_brick.privileged.rootwrap.unlink_root')
    def test_connect_volume_undo_prepare_result_non_custom_link(
            outer_self, conn_props, dev_info, mock_unlink, mock_dev_path):

        class Test(object):
            @utils.connect_volume_undo_prepare_result(unlink_after=True)
            def disconnect_volume(self, connection_properties, device_info,
                                  force=False, ignore_errors=False):
                outer_self.assertEqual(conn_props, connection_properties)
                outer_self.assertEqual(dev_info, device_info)
                return 'disconnect_volume'

            @utils.connect_volume_undo_prepare_result
            def extend_volume(self, connection_properties):
                outer_self.assertEqual(conn_props, connection_properties)
                return 'extend_volume'

        path = conn_props['device_path']
        mock_dev_path.return_value = path

        t = Test()

        res = t.disconnect_volume(conn_props, dev_info)
        outer_self.assertEqual('disconnect_volume', res)

        res = t.extend_volume(conn_props)
        outer_self.assertEqual('extend_volume', res)

        if conn_props.get('encrypted'):
            outer_self.assertEqual(2, mock_dev_path.call_count)
            mock_dev_path.assert_has_calls((mock.call(path), mock.call(path)))
        else:
            mock_dev_path.assert_not_called()
        mock_unlink.assert_not_called()

    @mock.patch('os_brick.utils._device_path_from_symlink')
    @mock.patch('os_brick.privileged.rootwrap.unlink_root')
    def test_connect_volume_undo_prepare_result_encrypted_disconnect(
            outer_self, mock_unlink, mock_dev_path):
        connector_path = '/dev/md/alias'
        mock_dev_path.return_value = connector_path
        symlink_path = '/dev/disk/by-id/os-brick_dev_md_alias'
        mock_unlink.side_effect = ValueError

        class Test(object):
            @utils.connect_volume_undo_prepare_result(unlink_after=True)
            def disconnect_volume(self, connection_properties, device_info,
                                  force=False, ignore_errors=False):
                outer_self.assertEqual(connector_path,
                                       connection_properties['device_path'])
                outer_self.assertEqual(connector_path,
                                       device_info['path'])
                return 'disconnect_volume'

        conn_props = {'target_portal': '198.72.124.185:3260',
                      'target_iqn': 'iqn.2010-10.org.openstack:volume-uuid',
                      'target_lun': 0,
                      'encrypted': True,
                      'device_path': symlink_path}
        dev_info = {'type': 'block', 'path': symlink_path}

        t = Test()
        res = t.disconnect_volume(conn_props, dev_info)

        outer_self.assertEqual('disconnect_volume', res)
        mock_dev_path.assert_called_once_with(symlink_path)
        mock_unlink.assert_called_once_with(symlink_path)

    @mock.patch('os_brick.utils._device_path_from_symlink')
    @mock.patch('os_brick.privileged.rootwrap.unlink_root')
    def test_connect_volume_undo_prepare_result_encrypted_extend(
            outer_self, mock_unlink, mock_dev_path):
        connector_path = '/dev/md/alias'
        mock_dev_path.return_value = connector_path
        symlink_path = '/dev/disk/by-id/os-brick_dev_md_alias'
        mock_unlink.side_effect = ValueError

        class Test(object):
            @utils.connect_volume_undo_prepare_result
            def extend_volume(self, connection_properties):
                outer_self.assertEqual(connector_path,
                                       connection_properties['device_path'])
                return 'extend_volume'

        conn_props = {'target_portal': '198.72.124.185:3260',
                      'target_iqn': 'iqn.2010-10.org.openstack:volume-uuid',
                      'target_lun': 0,
                      'encrypted': True,
                      'device_path': symlink_path}

        t = Test()
        res = t.extend_volume(conn_props)

        outer_self.assertEqual('extend_volume', res)
        mock_dev_path.assert_called_once_with(symlink_path)
        mock_unlink.assert_not_called()


@ddt.ddt
class AnyTestCase(base.TestCase):
    @ddt.data('hola', 1, None, {'a': 1}, {1, 2}, False)
    def test_equal(self, what):
        self.assertEqual(what, utils.ANY)
        self.assertEqual(utils.ANY, what)

    @ddt.data('hola', 1, None, {'a': 1}, {1, 2}, False)
    def test_different(self, what):
        self.assertFalse(what != utils.ANY)  # noqa
        self.assertFalse(utils.ANY != what)  # noqa
        self.assertFalse(utils.ANY > what)  # noqa
        self.assertFalse(utils.ANY < what)  # noqa
        self.assertFalse(utils.ANY <= what)  # noqa
        self.assertFalse(utils.ANY >= what)  # noqa