cinder/cinder/tests/unit/test_scality.py

# Copyright (c) 2015 Scality
# 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 the Scality SOFS Volume Driver.
"""
import errno
import os

import mock
from oslo_utils import imageutils
from six.moves import urllib

from cinder import context
from cinder import exception
from cinder import test
from cinder.tests.unit import fake_snapshot
from cinder.tests.unit import fake_volume
from cinder.volume import configuration as conf
import cinder.volume.drivers.scality as driver

_FAKE_VOLUME_ID = 'a79d463e-1fd5-11e5-a6ff-5b81bfee8544'
_FAKE_VOLUME_NAME = 'volume-a79d463e-1fd5-11e5-a6ff-5b81bfee8544'

_FAKE_SNAPSHOT_ID = 'ae3d6da2-1fd5-11e5-967f-1b8cf3b401ab'
_FAKE_SNAPSHOT_NAME = 'snapshot-ae3d6da2-1fd5-11e5-967f-1b8cf3b401ab'

_FAKE_BACKUP = {'id': '914849d2-2585-11e5-be54-d70ca0c343d6',
                'volume_id': _FAKE_VOLUME_ID}

_FAKE_MNT_POINT = '/tmp'
_FAKE_SOFS_CONFIG = '/etc/sfused.conf'
_FAKE_VOLUME_DIR = 'cinder/volumes'
_FAKE_VOL_BASEDIR = os.path.join(_FAKE_MNT_POINT, _FAKE_VOLUME_DIR, '00')
_FAKE_VOL_PATH = os.path.join(_FAKE_VOL_BASEDIR, _FAKE_VOLUME_NAME)
_FAKE_SNAP_PATH = os.path.join(_FAKE_VOL_BASEDIR, _FAKE_SNAPSHOT_NAME)

_FAKE_MOUNTS_TABLE = [['tmpfs /dev/shm\n'],
                      ['fuse ' + _FAKE_MNT_POINT + '\n']]


class ScalityDriverTestCase(test.TestCase):
    """Test case for the Scality driver."""

    def setUp(self):
        super(ScalityDriverTestCase, self).setUp()

        self.cfg = mock.Mock(spec=conf.Configuration)
        self.cfg.scality_sofs_mount_point = _FAKE_MNT_POINT
        self.cfg.scality_sofs_config = _FAKE_SOFS_CONFIG
        self.cfg.scality_sofs_volume_dir = _FAKE_VOLUME_DIR

        self.drv = driver.ScalityDriver(configuration=self.cfg)
        self.drv.db = mock.Mock()

        self.context = context.get_admin_context()

    def _simple_volume(self, **kwargs):
        updates = {'display_name': _FAKE_VOLUME_NAME,
                   'id': _FAKE_VOLUME_ID,
                   'provider_location': 'fake_share'}
        updates.update(kwargs)

        return fake_volume.fake_volume_obj(self.context, **updates)

    def _simple_snapshot(self, **kwargs):
        updates = {'id': _FAKE_SNAPSHOT_ID,
                   'display_name': _FAKE_SNAPSHOT_NAME,
                   'status': 'available',
                   'provider_location': None,
                   'volume_size': 1}

        updates.update(kwargs)
        snapshot = fake_snapshot.fake_snapshot_obj(self.context, **updates)
        volume = self._simple_volume()
        snapshot.volume = volume

        return snapshot

    @mock.patch.object(driver.urllib.request, 'urlopen')
    @mock.patch('os.access')
    def test_check_for_setup_error(self, mock_os_access, mock_urlopen):
        self.drv.check_for_setup_error()

        mock_urlopen.assert_called_once_with('file://%s' % _FAKE_SOFS_CONFIG,
                                             timeout=5)
        mock_os_access.assert_called_once_with('/sbin/mount.sofs', os.X_OK)

    def test_check_for_setup_error_with_no_sofs_config(self):
        self.cfg.scality_sofs_config = ''

        self.drv = driver.ScalityDriver(configuration=self.cfg)

        self.assertRaises(exception.VolumeBackendAPIException,
                          self.drv.check_for_setup_error)
        exec_patcher = mock.patch.object(self.drv, '_execute',
                                         mock.MagicMock())
        exec_patcher.start()
        self.addCleanup(exec_patcher.stop)

    @mock.patch.object(driver.urllib.request, 'urlopen')
    def test_check_for_setup_error_with_urlerror(self, mock_urlopen):
        # Add a Unicode char to be sure that the exception is properly
        # handled even if it contains Unicode chars
        mock_urlopen.side_effect = urllib.error.URLError(u'\u9535')
        self.assertRaises(exception.VolumeBackendAPIException,
                          self.drv.check_for_setup_error)

    @mock.patch.object(driver.urllib.request, 'urlopen')
    def test_check_for_setup_error_with_httperror(self, mock_urlopen):
        mock_urlopen.side_effect = urllib.error.HTTPError(*[None] * 5)
        self.assertRaises(exception.VolumeBackendAPIException,
                          self.drv.check_for_setup_error)

    @mock.patch.object(driver.urllib.request, 'urlopen', mock.Mock())
    @mock.patch('os.access')
    def test_check_for_setup_error_with_no_mountsofs(self, mock_os_access):
        mock_os_access.return_value = False
        self.assertRaises(exception.VolumeBackendAPIException,
                          self.drv.check_for_setup_error)
        mock_os_access.assert_called_once_with('/sbin/mount.sofs', os.X_OK)

    def test_load_shares_config(self):
        self.assertEqual({}, self.drv.shares)
        self.drv._load_shares_config()
        self.assertEqual({_FAKE_VOLUME_DIR: None}, self.drv.shares)

    def test_get_mount_point_for_share(self):
        self.assertEqual(_FAKE_VOL_BASEDIR,
                         self.drv._get_mount_point_for_share())

    @mock.patch("cinder.volume.utils.read_proc_mounts")
    @mock.patch("oslo_concurrency.processutils.execute")
    def test_ensure_share_mounted_when_mount_failed(self, mock_execute,
                                                    mock_read_proc_mounts):
        mock_read_proc_mounts.return_value = ['tmpfs /dev/shm\n']
        self.assertRaises(exception.VolumeBackendAPIException,
                          self.drv._ensure_share_mounted)
        self.assertEqual(2, mock_read_proc_mounts.call_count)
        self.assertEqual(1, mock_execute.call_count)

    @mock.patch("cinder.volume.utils.read_proc_mounts")
    @mock.patch("oslo_concurrency.processutils.execute")
    @mock.patch("oslo_utils.fileutils.ensure_tree")
    @mock.patch("os.symlink")
    def test_ensure_shares_mounted(self, mock_symlink, mock_ensure_tree,
                                   mock_execute, mock_read_proc_mounts):
        self.assertEqual([], self.drv._mounted_shares)

        mock_read_proc_mounts.side_effect = _FAKE_MOUNTS_TABLE

        self.drv._ensure_shares_mounted()

        self.assertEqual([_FAKE_VOLUME_DIR], self.drv._mounted_shares)
        self.assertEqual(2, mock_read_proc_mounts.call_count)
        mock_symlink.assert_called_once_with('.', _FAKE_VOL_BASEDIR)
        self.assertEqual(2, mock_ensure_tree.call_count)
        self.assertEqual(1, mock_execute.call_count)
        expected_args = ('mount', '-t', 'sofs', _FAKE_SOFS_CONFIG,
                         _FAKE_MNT_POINT)
        self.assertEqual(expected_args, mock_execute.call_args[0])

    @mock.patch("cinder.volume.utils.read_proc_mounts")
    @mock.patch("oslo_concurrency.processutils.execute")
    @mock.patch("oslo_utils.fileutils.ensure_tree", mock.Mock())
    @mock.patch("os.symlink", mock.Mock())
    def test_ensure_shares_mounted_when_sofs_mounted(self, mock_execute,
                                                     mock_read_proc_mounts):
        mock_read_proc_mounts.return_value = _FAKE_MOUNTS_TABLE[1]

        self.drv._ensure_shares_mounted()

        # Because SOFS is mounted from the beginning, we shouldn't read
        # /proc/mounts more than once.
        mock_read_proc_mounts.assert_called_once_with()
        self.assertFalse(mock_execute.called)

    def test_find_share_when_no_shares_mounted(self):
        self.assertRaises(exception.RemoteFSNoSharesMounted,
                          self.drv._find_share, 'ignored')

    @mock.patch("cinder.volume.utils.read_proc_mounts")
    @mock.patch("oslo_concurrency.processutils.execute")
    @mock.patch("oslo_utils.fileutils.ensure_tree")
    @mock.patch("os.symlink")
    def test_find_share(self, mock_symlink, mock_ensure_tree, mock_execute,
                        mock_read_proc_mounts):
        mock_read_proc_mounts.side_effect = _FAKE_MOUNTS_TABLE

        self.drv._ensure_shares_mounted()

        self.assertEqual(_FAKE_VOLUME_DIR, self.drv._find_share('ignored'))
        self.assertEqual(2, mock_read_proc_mounts.call_count)
        self.assertEqual(1, mock_execute.call_count)

        expected_args = ('mount', '-t', 'sofs', _FAKE_SOFS_CONFIG,
                         _FAKE_MNT_POINT)
        self.assertEqual(expected_args, mock_execute.call_args[0])

        mock_symlink.assert_called_once_with('.', _FAKE_VOL_BASEDIR)

        self.assertEqual(mock_ensure_tree.call_args_list,
                         [mock.call(_FAKE_MNT_POINT),
                          mock.call(os.path.join(_FAKE_MNT_POINT,
                                                 _FAKE_VOLUME_DIR))])

    def test_get_volume_stats(self):
        with mock.patch.object(self.cfg, 'safe_get') as mock_safe_get:
            mock_safe_get.return_value = 'fake_backend_name'
            stats = self.drv.get_volume_stats()
        self.assertEqual(self.drv.VERSION, stats['driver_version'])
        self.assertEqual(mock_safe_get.return_value,
                         stats['volume_backend_name'])
        mock_safe_get.assert_called_once_with('volume_backend_name')

    @mock.patch("cinder.image.image_utils.qemu_img_info")
    def test_initialize_connection(self, mock_qemu_img_info):
        info = imageutils.QemuImgInfo()
        info.file_format = 'raw'
        info.image = _FAKE_VOLUME_NAME
        mock_qemu_img_info.return_value = info

        volume = self._simple_volume()
        with mock.patch.object(self.drv, 'get_active_image_from_info') as \
                mock_get_active_image_from_info:

            mock_get_active_image_from_info.return_value = _FAKE_VOLUME_NAME
            conn_info = self.drv.initialize_connection(volume, None)

        expected_conn_info = {
            'driver_volume_type': driver.ScalityDriver.driver_volume_type,
            'mount_point_base': _FAKE_MNT_POINT,
            'data': {
                'export': volume.provider_location,
                'name': volume.name,
                'sofs_path': 'cinder/volumes/00/' + volume.name,
                'format': 'raw'
            }
        }
        self.assertEqual(expected_conn_info, conn_info)
        mock_get_active_image_from_info.assert_called_once_with(volume)
        mock_qemu_img_info.assert_called_once_with(_FAKE_VOL_PATH)

    @mock.patch("cinder.image.image_utils.resize_image")
    @mock.patch("cinder.image.image_utils.qemu_img_info")
    def test_extend_volume(self, mock_qemu_img_info, mock_resize_image):
        info = imageutils.QemuImgInfo()
        info.file_format = 'raw'
        mock_qemu_img_info.return_value = info

        self.drv.extend_volume(self._simple_volume(), 2)

        mock_qemu_img_info.assert_called_once_with(_FAKE_VOL_PATH)

        mock_resize_image.assert_called_once_with(_FAKE_VOL_PATH, 2)

    @mock.patch("cinder.image.image_utils.qemu_img_info")
    def test_extend_volume_with_invalid_format(self, mock_qemu_img_info):
        info = imageutils.QemuImgInfo()
        info.file_format = 'vmdk'
        mock_qemu_img_info.return_value = info

        self.assertRaises(exception.InvalidVolume,
                          self.drv.extend_volume, self._simple_volume(), 2)

    @mock.patch("cinder.image.image_utils.resize_image")
    @mock.patch("cinder.image.image_utils.convert_image")
    def test_copy_volume_from_snapshot_with_ioerror(self, mock_convert_image,
                                                    mock_resize_image):
        with mock.patch.object(self.drv, '_read_info_file') as \
                mock_read_info_file, \
                mock.patch.object(self.drv, '_set_rw_permissions_for_all') as \
                mock_set_rw_permissions:
            mock_read_info_file.side_effect = IOError(errno.ENOENT, '')
            self.drv._copy_volume_from_snapshot(self._simple_snapshot(),
                                                self._simple_volume(), 1)

        mock_read_info_file.assert_called_once_with("%s.info" % _FAKE_VOL_PATH)
        mock_convert_image.assert_called_once_with(_FAKE_SNAP_PATH,
                                                   _FAKE_VOL_PATH, 'raw',
                                                   run_as_root=True)
        mock_set_rw_permissions.assert_called_once_with(_FAKE_VOL_PATH)
        mock_resize_image.assert_called_once_with(_FAKE_VOL_PATH, 1)

    @mock.patch("cinder.image.image_utils.resize_image")
    @mock.patch("cinder.image.image_utils.convert_image")
    @mock.patch("cinder.image.image_utils.qemu_img_info")
    def test_copy_volume_from_snapshot(self, mock_qemu_img_info,
                                       mock_convert_image, mock_resize_image):

        new_volume = self._simple_volume(
            display_name='volume-3fa63b02-1fe5-11e5-b492-abf97a8fb23b',
            id='3fa63b02-1fe5-11e5-b492-abf97a8fb23b',
            provider_location='fake_share')
        new_vol_path = os.path.join(_FAKE_VOL_BASEDIR, new_volume.name)

        info = imageutils.QemuImgInfo()
        info.file_format = 'raw'
        info.backing_file = _FAKE_VOL_PATH
        mock_qemu_img_info.return_value = info

        with mock.patch.object(self.drv, '_read_info_file') as \
                mock_read_info_file, \
                mock.patch.object(self.drv, '_set_rw_permissions_for_all') as \
                mock_set_rw_permissions:
            self.drv._copy_volume_from_snapshot(self._simple_snapshot(),
                                                new_volume, 1)

        mock_read_info_file.assert_called_once_with("%s.info" % _FAKE_VOL_PATH)
        mock_convert_image.assert_called_once_with(_FAKE_VOL_PATH,
                                                   new_vol_path, 'raw',
                                                   run_as_root=True)
        mock_set_rw_permissions.assert_called_once_with(new_vol_path)
        mock_resize_image.assert_called_once_with(new_vol_path, 1)

    @mock.patch("cinder.image.image_utils.qemu_img_info")
    @mock.patch("cinder.utils.temporary_chown")
    @mock.patch("six.moves.builtins.open")
    def test_backup_volume(self, mock_open, mock_temporary_chown,
                           mock_qemu_img_info):
        """Backup a volume with no snapshots."""

        info = imageutils.QemuImgInfo()
        info.file_format = 'raw'
        mock_qemu_img_info.return_value = info

        backup = {'volume_id': _FAKE_VOLUME_ID}
        mock_backup_service = mock.MagicMock()
        self.drv.db.volume_get.return_value = self._simple_volume()

        self.drv.backup_volume(context, backup, mock_backup_service)

        mock_qemu_img_info.assert_called_once_with(_FAKE_VOL_PATH)
        mock_temporary_chown.assert_called_once_with(_FAKE_VOL_PATH)
        mock_open.assert_called_once_with(_FAKE_VOL_PATH)
        mock_backup_service.backup.assert_called_once_with(
            backup, mock_open().__enter__())

    @mock.patch("cinder.image.image_utils.qemu_img_info")
    def test_backup_volume_with_non_raw_volume(self, mock_qemu_img_info):

        info = imageutils.QemuImgInfo()
        info.file_format = 'qcow2'
        mock_qemu_img_info.return_value = info

        self.drv.db.volume_get.return_value = self._simple_volume()

        self.assertRaises(exception.InvalidVolume, self.drv.backup_volume,
                          context, _FAKE_BACKUP, mock.MagicMock())

        mock_qemu_img_info.assert_called_once_with(_FAKE_VOL_PATH)

    @mock.patch("cinder.image.image_utils.qemu_img_info")
    def test_backup_volume_with_backing_file(self, mock_qemu_img_info):

        info = imageutils.QemuImgInfo()
        info.file_format = 'raw'
        info.backing_file = 'fake.img'
        mock_qemu_img_info.return_value = info

        backup = {'volume_id': _FAKE_VOLUME_ID}
        self.drv.db.volume_get.return_value = self._simple_volume()

        self.assertRaises(exception.InvalidVolume, self.drv.backup_volume,
                          context, backup, mock.MagicMock())

        mock_qemu_img_info.assert_called_once_with(_FAKE_VOL_PATH)

    @mock.patch("cinder.utils.temporary_chown")
    @mock.patch("six.moves.builtins.open")
    def test_restore_bakup(self, mock_open, mock_temporary_chown):
        mock_backup_service = mock.MagicMock()

        self.drv.restore_backup(context, _FAKE_BACKUP, self._simple_volume(),
                                mock_backup_service)

        mock_temporary_chown.assert_called_once_with(_FAKE_VOL_PATH)
        mock_open.assert_called_once_with(_FAKE_VOL_PATH, 'wb')
        mock_backup_service.restore.assert_called_once_with(
            _FAKE_BACKUP, _FAKE_VOLUME_ID, mock_open().__enter__())