"Fossies" - the Fresh Open Source Software Archive

Member "cinder-17.1.0/cinder/tests/unit/volume/drivers/test_nfs.py" (8 Mar 2021, 71539 Bytes) of package /linux/misc/openstack/cinder-17.1.0.tar.gz:


As a special service "Fossies" has tried to format the requested source page into HTML format using (guessed) Python source code syntax highlighting (style: standard) with prefixed line numbers. Alternatively you can here view or download the uninterpreted source code file. See also the latest Fossies "Diffs" side-by-side code changes report for "test_nfs.py": 17.0.1_vs_17.1.0.

    1 # Copyright (c) 2012 NetApp, Inc.
    2 # All Rights Reserved.
    3 #
    4 #    Licensed under the Apache License, Version 2.0 (the "License"); you may
    5 #    not use this file except in compliance with the License. You may obtain
    6 #    a copy of the License at
    7 #
    8 #         http://www.apache.org/licenses/LICENSE-2.0
    9 #
   10 #    Unless required by applicable law or agreed to in writing, software
   11 #    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
   12 #    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
   13 #    License for the specific language governing permissions and limitations
   14 #    under the License.
   15 """Unit tests for the NFS driver module."""
   16 
   17 import errno
   18 import os
   19 from unittest import mock
   20 
   21 import castellan
   22 import ddt
   23 from oslo_utils import imageutils
   24 from oslo_utils import units
   25 
   26 from cinder import context
   27 from cinder import exception
   28 from cinder.image import image_utils
   29 from cinder.tests.unit import fake_constants as fake
   30 from cinder.tests.unit import fake_snapshot
   31 from cinder.tests.unit import fake_volume
   32 from cinder.tests.unit.keymgr import fake as fake_keymgr
   33 from cinder.tests.unit import test
   34 from cinder.volume import configuration as conf
   35 from cinder.volume.drivers import nfs
   36 from cinder.volume.drivers import remotefs
   37 from cinder.volume import volume_utils
   38 
   39 
   40 class KeyObject(object):
   41     def get_encoded(arg):
   42         return "asdf".encode('utf-8')
   43 
   44 
   45 class RemoteFsDriverTestCase(test.TestCase):
   46     TEST_FILE_NAME = 'test.txt'
   47     TEST_EXPORT = 'nas-host1:/export'
   48     TEST_MNT_POINT = '/mnt/nas'
   49 
   50     def setUp(self):
   51         super(RemoteFsDriverTestCase, self).setUp()
   52         self._driver = remotefs.RemoteFSDriver()
   53         self.configuration = mock.Mock(conf.Configuration)
   54         self.configuration.append_config_values(mock.ANY)
   55         self.configuration.nas_secure_file_permissions = 'false'
   56         self.configuration.nas_secure_file_operations = 'false'
   57         self.configuration.nfs_snapshot_support = True
   58         self.configuration.max_over_subscription_ratio = 1.0
   59         self.configuration.reserved_percentage = 5
   60         self._driver = remotefs.RemoteFSDriver(
   61             configuration=self.configuration)
   62         mock_exc = mock.patch.object(self._driver, '_execute')
   63         self._execute = mock_exc.start()
   64         self.addCleanup(mock_exc.stop)
   65 
   66     def test_create_sparsed_file(self):
   67         self._driver._create_sparsed_file('/path', 1)
   68         self._execute.assert_called_once_with('truncate', '-s', '1G',
   69                                               '/path', run_as_root=True)
   70 
   71     def test_create_regular_file(self):
   72         self._driver._create_regular_file('/path', 1)
   73         self._execute.assert_called_once_with('dd', 'if=/dev/zero',
   74                                               'of=/path', 'bs=1M',
   75                                               'count=1024', run_as_root=True)
   76 
   77     def test_create_qcow2_file(self):
   78         file_size = 1
   79         self._driver._create_qcow2_file('/path', file_size)
   80         self._execute.assert_called_once_with('qemu-img', 'create', '-f',
   81                                               'qcow2', '-o',
   82                                               'preallocation=metadata',
   83                                               '/path', '%s' %
   84                                               str(file_size * units.Gi),
   85                                               run_as_root=True)
   86 
   87     def test_set_rw_permissions_for_all(self):
   88         self._driver._set_rw_permissions_for_all('/path')
   89         self._execute.assert_called_once_with('chmod', 'ugo+rw', '/path',
   90                                               run_as_root=True)
   91 
   92     @mock.patch.object(remotefs, 'LOG')
   93     def test_set_rw_permissions_with_secure_file_permissions(self, LOG):
   94         self._driver._mounted_shares = [self.TEST_EXPORT]
   95         self.configuration.nas_secure_file_permissions = 'true'
   96         self._driver._set_rw_permissions(self.TEST_FILE_NAME)
   97 
   98         self.assertFalse(LOG.warning.called)
   99 
  100     @mock.patch.object(remotefs, 'LOG')
  101     def test_set_rw_permissions_without_secure_file_permissions(self, LOG):
  102         self.configuration.nas_secure_file_permissions = 'false'
  103         self._driver._set_rw_permissions(self.TEST_FILE_NAME)
  104 
  105         self.assertTrue(LOG.warning.called)
  106         warn_msg = "%(path)s is being set with open permissions: %(perm)s"
  107         LOG.warning.assert_called_once_with(
  108             warn_msg, {'path': self.TEST_FILE_NAME, 'perm': 'ugo+rw'})
  109 
  110     @mock.patch('os.path.join')
  111     @mock.patch('os.path.isfile', return_value=False)
  112     def test_determine_nas_security_options_when_auto_and_new_install(
  113             self,
  114             mock_isfile,
  115             mock_join):
  116         """Test the setting of the NAS Security Option
  117 
  118          In this test case, we will create the marker file. No pre-exxisting
  119          Cinder volumes found during bootup.
  120          """
  121         self._driver._mounted_shares = [self.TEST_EXPORT]
  122         file_path = '%s/.cinderSecureEnvIndicator' % self.TEST_MNT_POINT
  123         is_new_install = True
  124 
  125         self._driver._ensure_shares_mounted = mock.Mock()
  126         nas_mount = self._driver._get_mount_point_for_share = mock.Mock(
  127             return_value=self.TEST_MNT_POINT)
  128         mock_join.return_value = file_path
  129 
  130         secure_file_permissions = 'auto'
  131         nas_option = self._driver._determine_nas_security_option_setting(
  132             secure_file_permissions,
  133             nas_mount, is_new_install)
  134 
  135         self.assertEqual('true', nas_option)
  136 
  137         secure_file_operations = 'auto'
  138         nas_option = self._driver._determine_nas_security_option_setting(
  139             secure_file_operations,
  140             nas_mount, is_new_install)
  141 
  142         self.assertEqual('true', nas_option)
  143 
  144     @mock.patch('os.path.join')
  145     @mock.patch('os.path.isfile')
  146     def test_determine_nas_security_options_when_auto_and_new_install_exists(
  147             self,
  148             isfile,
  149             join):
  150         """Test the setting of the NAS Security Option
  151 
  152         In this test case, the marker file already exists. Cinder volumes
  153         found during bootup.
  154         """
  155         drv = self._driver
  156         drv._mounted_shares = [self.TEST_EXPORT]
  157         file_path = '%s/.cinderSecureEnvIndicator' % self.TEST_MNT_POINT
  158         is_new_install = False
  159 
  160         drv._ensure_shares_mounted = mock.Mock()
  161         nas_mount = drv._get_mount_point_for_share = mock.Mock(
  162             return_value=self.TEST_MNT_POINT)
  163         join.return_value = file_path
  164         isfile.return_value = True
  165 
  166         secure_file_permissions = 'auto'
  167         nas_option = drv._determine_nas_security_option_setting(
  168             secure_file_permissions,
  169             nas_mount, is_new_install)
  170 
  171         self.assertEqual('true', nas_option)
  172 
  173         secure_file_operations = 'auto'
  174         nas_option = drv._determine_nas_security_option_setting(
  175             secure_file_operations,
  176             nas_mount, is_new_install)
  177 
  178         self.assertEqual('true', nas_option)
  179 
  180     @mock.patch('os.path.join')
  181     @mock.patch('os.path.isfile')
  182     def test_determine_nas_security_options_when_auto_and_old_install(self,
  183                                                                       isfile,
  184                                                                       join):
  185         """Test the setting of the NAS Security Option
  186 
  187         In this test case, the marker file does not exist. There are also
  188         pre-existing Cinder volumes.
  189         """
  190         drv = self._driver
  191         drv._mounted_shares = [self.TEST_EXPORT]
  192         file_path = '%s/.cinderSecureEnvIndicator' % self.TEST_MNT_POINT
  193         is_new_install = False
  194 
  195         drv._ensure_shares_mounted = mock.Mock()
  196         nas_mount = drv._get_mount_point_for_share = mock.Mock(
  197             return_value=self.TEST_MNT_POINT)
  198         join.return_value = file_path
  199         isfile.return_value = False
  200 
  201         secure_file_permissions = 'auto'
  202         nas_option = drv._determine_nas_security_option_setting(
  203             secure_file_permissions,
  204             nas_mount, is_new_install)
  205 
  206         self.assertEqual('false', nas_option)
  207 
  208         secure_file_operations = 'auto'
  209         nas_option = drv._determine_nas_security_option_setting(
  210             secure_file_operations,
  211             nas_mount, is_new_install)
  212 
  213         self.assertEqual('false', nas_option)
  214 
  215     def test_determine_nas_security_options_when_admin_set_true(self):
  216         """Test the setting of the NAS Security Option
  217 
  218         In this test case, the Admin set the flag to 'true'.
  219         """
  220         drv = self._driver
  221         drv._mounted_shares = [self.TEST_EXPORT]
  222         is_new_install = False
  223 
  224         drv._ensure_shares_mounted = mock.Mock()
  225         nas_mount = drv._get_mount_point_for_share = mock.Mock(
  226             return_value=self.TEST_MNT_POINT)
  227 
  228         secure_file_permissions = 'true'
  229         nas_option = drv._determine_nas_security_option_setting(
  230             secure_file_permissions,
  231             nas_mount, is_new_install)
  232 
  233         self.assertEqual('true', nas_option)
  234 
  235         secure_file_operations = 'true'
  236         nas_option = drv._determine_nas_security_option_setting(
  237             secure_file_operations,
  238             nas_mount, is_new_install)
  239 
  240         self.assertEqual('true', nas_option)
  241 
  242     def test_determine_nas_security_options_when_admin_set_false(self):
  243         """Test the setting of the NAS Security Option
  244 
  245         In this test case, the Admin set the flag to 'false'.
  246         """
  247         drv = self._driver
  248         drv._mounted_shares = [self.TEST_EXPORT]
  249         is_new_install = False
  250 
  251         drv._ensure_shares_mounted = mock.Mock()
  252         nas_mount = drv._get_mount_point_for_share = mock.Mock(
  253             return_value=self.TEST_MNT_POINT)
  254 
  255         secure_file_permissions = 'false'
  256         nas_option = drv._determine_nas_security_option_setting(
  257             secure_file_permissions,
  258             nas_mount, is_new_install)
  259 
  260         self.assertEqual('false', nas_option)
  261 
  262         secure_file_operations = 'false'
  263         nas_option = drv._determine_nas_security_option_setting(
  264             secure_file_operations,
  265             nas_mount, is_new_install)
  266 
  267         self.assertEqual('false', nas_option)
  268 
  269     @mock.patch.object(remotefs, 'LOG')
  270     def test_set_nas_security_options(self, LOG):
  271         """Test setting of NAS Security options.
  272 
  273         The RemoteFS driver will force set options to false. The derived
  274         objects will provide an inherited interface to properly set options.
  275         """
  276         drv = self._driver
  277         is_new_install = False
  278 
  279         drv.set_nas_security_options(is_new_install)
  280 
  281         self.assertEqual('false', drv.configuration.nas_secure_file_operations)
  282         self.assertEqual('false',
  283                          drv.configuration.nas_secure_file_permissions)
  284         self.assertTrue(LOG.warning.called)
  285 
  286     def test_secure_file_operations_enabled_true(self):
  287         """Test nas_secure_file_operations = 'true'
  288 
  289         Networked file system based drivers may support secure file
  290         operations. This test verifies the settings when secure.
  291         """
  292         drv = self._driver
  293         self.configuration.nas_secure_file_operations = 'true'
  294         ret_flag = drv.secure_file_operations_enabled()
  295         self.assertTrue(ret_flag)
  296 
  297     def test_secure_file_operations_enabled_false(self):
  298         """Test nas_secure_file_operations = 'false'
  299 
  300         Networked file system based drivers may support secure file
  301         operations. This test verifies the settings when not secure.
  302         """
  303         drv = self._driver
  304         self.configuration.nas_secure_file_operations = 'false'
  305         ret_flag = drv.secure_file_operations_enabled()
  306         self.assertFalse(ret_flag)
  307 
  308 
  309 # NFS configuration scenarios
  310 NFS_CONFIG1 = {'max_over_subscription_ratio': 1.0,
  311                'reserved_percentage': 0,
  312                'nfs_sparsed_volumes': True,
  313                'nfs_qcow2_volumes': False,
  314                'nas_secure_file_permissions': 'false',
  315                'nas_secure_file_operations': 'false'}
  316 
  317 NFS_CONFIG2 = {'max_over_subscription_ratio': 10.0,
  318                'reserved_percentage': 5,
  319                'nfs_sparsed_volumes': False,
  320                'nfs_qcow2_volumes': True,
  321                'nas_secure_file_permissions': 'true',
  322                'nas_secure_file_operations': 'true'}
  323 
  324 NFS_CONFIG3 = {'max_over_subscription_ratio': 15.0,
  325                'reserved_percentage': 10,
  326                'nfs_sparsed_volumes': False,
  327                'nfs_qcow2_volumes': False,
  328                'nas_secure_file_permissions': 'auto',
  329                'nas_secure_file_operations': 'auto'}
  330 
  331 NFS_CONFIG4 = {'max_over_subscription_ratio': 20.0,
  332                'reserved_percentage': 60,
  333                'nfs_sparsed_volumes': True,
  334                'nfs_qcow2_volumes': True,
  335                'nas_secure_file_permissions': 'false',
  336                'nas_secure_file_operations': 'true'}
  337 
  338 QEMU_IMG_INFO_OUT1 = """image: %(volid)s
  339         file format: raw
  340         virtual size: %(size_gb)sG (%(size_b)s bytes)
  341         disk size: 173K
  342         """
  343 
  344 QEMU_IMG_INFO_OUT2 = """image: %(volid)s
  345 file format: qcow2
  346 virtual size: %(size_gb)sG (%(size_b)s bytes)
  347 disk size: 196K
  348 cluster_size: 65536
  349 Format specific information:
  350     compat: 1.1
  351     lazy refcounts: false
  352     refcount bits: 16
  353     corrupt: false
  354     """
  355 
  356 QEMU_IMG_INFO_OUT3 = """image: volume-%(volid)s.%(snapid)s
  357 file format: qcow2
  358 virtual size: %(size_gb)sG (%(size_b)s bytes)
  359 disk size: 196K
  360 cluster_size: 65536
  361 backing file: volume-%(volid)s
  362 backing file format: qcow2
  363 Format specific information:
  364     compat: 1.1
  365     lazy refcounts: false
  366     refcount bits: 16
  367     corrupt: false
  368     """
  369 
  370 QEMU_IMG_INFO_OUT4 = """image: volume-%(volid)s.%(snapid)s
  371 file format: raw
  372 virtual size: %(size_gb)sG (%(size_b)s bytes)
  373 disk size: 196K
  374 cluster_size: 65536
  375 backing file: volume-%(volid)s
  376 backing file format: raw
  377 Format specific information:
  378     compat: 1.1
  379     lazy refcounts: false
  380     refcount bits: 16
  381     corrupt: false
  382     """
  383 
  384 QEMU_IMG_INFO_OUT5 = """image: volume-%(volid)s.%(snapid)s
  385 file format: qcow2
  386 virtual size: %(size_gb)sG (%(size_b)s bytes)
  387 disk size: 196K
  388 encrypted: yes
  389 cluster_size: 65536
  390 backing file: volume-%(volid)s
  391 backing file format: raw
  392 Format specific information:
  393     compat: 1.1
  394     lazy refcounts: false
  395     refcount bits: 16
  396     encrypt:
  397         ivgen alg: plain64
  398         hash alg: sha256
  399         cipher alg: aes-256
  400         uuid: 386f8626-33f0-4683-a517-78ddfe385e33
  401         format: luks
  402         cipher mode: xts
  403         slots:
  404             [0]:
  405                 active: true
  406                 iters: 1892498
  407                 key offset: 4096
  408                 stripes: 4000
  409             [1]:
  410                 active: false
  411                 key offset: 262144
  412             [2]:
  413                 active: false
  414                 key offset: 520192
  415             [3]:
  416                 active: false
  417                 key offset: 778240
  418             [4]:
  419                 active: false
  420                 key offset: 1036288
  421             [5]:
  422                 active: false
  423                 key offset: 1294336
  424             [6]:
  425                 active: false
  426                 key offset: 1552384
  427             [7]:
  428                 active: false
  429                 key offset: 1810432
  430         payload offset: 2068480
  431         master key iters: 459347
  432     corrupt: false
  433 """
  434 
  435 
  436 @ddt.ddt
  437 class NfsDriverTestCase(test.TestCase):
  438     """Test case for NFS driver."""
  439 
  440     TEST_NFS_HOST = 'nfs-host1'
  441     TEST_NFS_SHARE_PATH = '/export'
  442     TEST_NFS_EXPORT1 = '%s:%s' % (TEST_NFS_HOST, TEST_NFS_SHARE_PATH)
  443     TEST_NFS_EXPORT2 = 'nfs-host2:/export'
  444     TEST_NFS_EXPORT2_OPTIONS = '-o intr'
  445     TEST_SIZE_IN_GB = 1
  446     TEST_MNT_POINT = '/mnt/nfs'
  447     TEST_MNT_POINT_BASE_EXTRA_SLASH = '/opt/stack/data/cinder//mnt'
  448     TEST_MNT_POINT_BASE = '/mnt/test'
  449     TEST_LOCAL_PATH = '/mnt/nfs/volume-123'
  450     TEST_FILE_NAME = 'test.txt'
  451     TEST_SHARES_CONFIG_FILE = '/etc/cinder/test-shares.conf'
  452     TEST_NFS_EXPORT_SPACES = 'nfs-host3:/export this'
  453     TEST_MNT_POINT_SPACES = '/ 0 0 0 /foo'
  454 
  455     def setUp(self):
  456         super(NfsDriverTestCase, self).setUp()
  457         self.configuration = mock.Mock(conf.Configuration)
  458         self.configuration.append_config_values(mock.ANY)
  459         self.configuration.max_over_subscription_ratio = 1.0
  460         self.configuration.reserved_percentage = 5
  461         self.configuration.nfs_shares_config = None
  462         self.configuration.nfs_sparsed_volumes = True
  463         self.configuration.nfs_reserved_percentage = 5.0
  464         self.configuration.nfs_mount_point_base = self.TEST_MNT_POINT_BASE
  465         self.configuration.nfs_mount_options = None
  466         self.configuration.nfs_mount_attempts = 3
  467         self.configuration.nfs_qcow2_volumes = False
  468         self.configuration.nas_secure_file_permissions = 'false'
  469         self.configuration.nas_secure_file_operations = 'false'
  470         self.configuration.nas_host = None
  471         self.configuration.nas_share_path = None
  472         self.configuration.nas_mount_options = None
  473         self.configuration.volume_dd_blocksize = '1M'
  474 
  475         self.mock_object(volume_utils, 'get_max_over_subscription_ratio',
  476                          return_value=1)
  477 
  478         self.context = context.get_admin_context()
  479 
  480     def _set_driver(self, extra_confs=None):
  481 
  482         # Overide the default configs
  483         if extra_confs:
  484             for config_name, config_value in extra_confs.items():
  485                 setattr(self.configuration, config_name, config_value)
  486 
  487         self._driver = nfs.NfsDriver(configuration=self.configuration)
  488         self._driver.shares = {}
  489         self.mock_object(self._driver, '_execute')
  490 
  491     @ddt.data(NFS_CONFIG1, NFS_CONFIG2, NFS_CONFIG3, NFS_CONFIG4)
  492     def test_local_path(self, nfs_config):
  493         """local_path common use case."""
  494         self.configuration.nfs_mount_point_base = self.TEST_MNT_POINT_BASE
  495         self._set_driver(extra_confs=nfs_config)
  496         drv = self._driver
  497 
  498         volume = fake_volume.fake_volume_obj(
  499             self.context,
  500             provider_location=self.TEST_NFS_EXPORT1)
  501 
  502         self.assertEqual(
  503             '/mnt/test/2f4f60214cf43c595666dd815f0360a4/%s' % volume.name,
  504             drv.local_path(volume))
  505 
  506     @ddt.data(NFS_CONFIG1, NFS_CONFIG2, NFS_CONFIG3, NFS_CONFIG4)
  507     def test_copy_image_to_volume(self, nfs_config):
  508         """resize_image common case usage."""
  509 
  510         mock_resize = self.mock_object(image_utils, 'resize_image')
  511         mock_fetch = self.mock_object(image_utils, 'fetch_to_raw')
  512 
  513         self._set_driver()
  514         drv = self._driver
  515         volume = fake_volume.fake_volume_obj(self.context,
  516                                              size=self.TEST_SIZE_IN_GB)
  517         test_img_source = 'volume-%s' % volume.id
  518 
  519         self.mock_object(drv, 'local_path', return_value=test_img_source)
  520 
  521         data = mock.Mock()
  522         data.virtual_size = 1 * units.Gi
  523         self.mock_object(image_utils, 'qemu_img_info', return_value=data)
  524         drv.copy_image_to_volume(None, volume, None, None)
  525 
  526         mock_fetch.assert_called_once_with(
  527             None, None, None, test_img_source, mock.ANY, run_as_root=True,
  528             size=self.TEST_SIZE_IN_GB)
  529         mock_resize.assert_called_once_with(test_img_source,
  530                                             self.TEST_SIZE_IN_GB,
  531                                             run_as_root=True)
  532 
  533     def test_get_mount_point_for_share(self):
  534         """_get_mount_point_for_share should calculate correct value."""
  535         self._set_driver()
  536         drv = self._driver
  537 
  538         self.configuration.nfs_mount_point_base = self.TEST_MNT_POINT_BASE
  539 
  540         self.assertEqual('/mnt/test/2f4f60214cf43c595666dd815f0360a4',
  541                          drv._get_mount_point_for_share(self.TEST_NFS_EXPORT1))
  542 
  543     def test_get_mount_point_for_share_given_extra_slash_in_state_path(self):
  544         """_get_mount_point_for_share should calculate correct value."""
  545         # This test gets called with the extra slash
  546         self.configuration.nfs_mount_point_base = (
  547             self.TEST_MNT_POINT_BASE_EXTRA_SLASH)
  548 
  549         # The driver gets called with the correct configuration and removes
  550         # the extra slash
  551         drv = nfs.NfsDriver(configuration=self.configuration)
  552 
  553         self.assertEqual('/opt/stack/data/cinder/mnt', drv.base)
  554 
  555         self.assertEqual(
  556             '/opt/stack/data/cinder/mnt/2f4f60214cf43c595666dd815f0360a4',
  557             drv._get_mount_point_for_share(self.TEST_NFS_EXPORT1))
  558 
  559     def test_get_capacity_info(self):
  560         """_get_capacity_info should calculate correct value."""
  561         self._set_driver()
  562         drv = self._driver
  563         stat_total_size = 2620544
  564         stat_avail = 2129984
  565         stat_output = '1 %d %d' % (stat_total_size, stat_avail)
  566 
  567         du_used = 490560
  568         du_output = '%d /mnt' % du_used
  569 
  570         with mock.patch.object(
  571                 drv, '_get_mount_point_for_share') as mock_get_mount:
  572             mock_get_mount.return_value = self.TEST_MNT_POINT
  573             drv._execute.side_effect = [(stat_output, None),
  574                                         (du_output, None)]
  575 
  576             self.assertEqual((stat_total_size, stat_avail, du_used),
  577                              drv._get_capacity_info(self.TEST_NFS_EXPORT1))
  578 
  579             mock_get_mount.assert_called_once_with(self.TEST_NFS_EXPORT1)
  580 
  581             calls = [mock.call('stat', '-f', '-c', '%S %b %a',
  582                                self.TEST_MNT_POINT, run_as_root=True),
  583                      mock.call('du', '-sb', '--apparent-size',
  584                                '--exclude', '*snapshot*',
  585                                self.TEST_MNT_POINT, run_as_root=True)]
  586 
  587             drv._execute.assert_has_calls(calls)
  588 
  589     def test_get_capacity_info_for_share_and_mount_point_with_spaces(self):
  590         """_get_capacity_info should calculate correct value."""
  591         self._set_driver()
  592         drv = self._driver
  593         stat_total_size = 2620544
  594         stat_avail = 2129984
  595         stat_output = '1 %d %d' % (stat_total_size, stat_avail)
  596 
  597         du_used = 490560
  598         du_output = '%d /mnt' % du_used
  599 
  600         with mock.patch.object(
  601                 drv, '_get_mount_point_for_share') as mock_get_mount:
  602             mock_get_mount.return_value = self.TEST_MNT_POINT_SPACES
  603             drv._execute.side_effect = [(stat_output, None),
  604                                         (du_output, None)]
  605 
  606             self.assertEqual((stat_total_size, stat_avail, du_used),
  607                              drv._get_capacity_info(
  608                                  self.TEST_NFS_EXPORT_SPACES))
  609 
  610             mock_get_mount.assert_called_once_with(
  611                 self.TEST_NFS_EXPORT_SPACES)
  612 
  613             calls = [mock.call('stat', '-f', '-c', '%S %b %a',
  614                                self.TEST_MNT_POINT_SPACES, run_as_root=True),
  615                      mock.call('du', '-sb', '--apparent-size',
  616                                '--exclude', '*snapshot*',
  617                                self.TEST_MNT_POINT_SPACES, run_as_root=True)]
  618 
  619             drv._execute.assert_has_calls(calls)
  620 
  621     def test_load_shares_config(self):
  622         self._set_driver()
  623         drv = self._driver
  624 
  625         drv.configuration.nfs_shares_config = self.TEST_SHARES_CONFIG_FILE
  626 
  627         with mock.patch.object(
  628                 drv, '_read_config_file') as mock_read_config:
  629             config_data = []
  630             config_data.append(self.TEST_NFS_EXPORT1)
  631             config_data.append('#' + self.TEST_NFS_EXPORT2)
  632             config_data.append('')
  633             config_data.append(self.TEST_NFS_EXPORT2 + ' ' +
  634                                self.TEST_NFS_EXPORT2_OPTIONS)
  635             config_data.append('broken:share_format')
  636             mock_read_config.return_value = config_data
  637 
  638             drv._load_shares_config(drv.configuration.nfs_shares_config)
  639 
  640             mock_read_config.assert_called_once_with(
  641                 self.TEST_SHARES_CONFIG_FILE)
  642             self.assertIn(self.TEST_NFS_EXPORT1, drv.shares)
  643             self.assertIn(self.TEST_NFS_EXPORT2, drv.shares)
  644             self.assertEqual(2, len(drv.shares))
  645 
  646             self.assertEqual(self.TEST_NFS_EXPORT2_OPTIONS,
  647                              drv.shares[self.TEST_NFS_EXPORT2])
  648 
  649     def test_load_shares_config_nas_opts(self):
  650         self._set_driver()
  651         drv = self._driver
  652         drv.configuration.nas_host = self.TEST_NFS_HOST
  653         drv.configuration.nas_share_path = self.TEST_NFS_SHARE_PATH
  654         drv.configuration.nfs_shares_config = self.TEST_SHARES_CONFIG_FILE
  655 
  656         drv._load_shares_config(drv.configuration.nfs_shares_config)
  657 
  658         self.assertIn(self.TEST_NFS_EXPORT1, drv.shares)
  659         self.assertEqual(1, len(drv.shares))
  660 
  661     def test_ensure_shares_mounted_should_save_mounting_successfully(self):
  662         """_ensure_shares_mounted should save share if mounted with success."""
  663         self._set_driver()
  664         drv = self._driver
  665         config_data = []
  666         config_data.append(self.TEST_NFS_EXPORT1)
  667         drv.configuration.nfs_shares_config = self.TEST_SHARES_CONFIG_FILE
  668 
  669         with mock.patch.object(
  670                 drv, '_read_config_file') as mock_read_config:
  671             with mock.patch.object(
  672                     drv, '_ensure_share_mounted') as mock_ensure:
  673                 mock_read_config.return_value = config_data
  674                 drv._ensure_share_mounted(self.TEST_NFS_EXPORT1)
  675                 mock_ensure.assert_called_once_with(self.TEST_NFS_EXPORT1)
  676 
  677     @mock.patch.object(remotefs, 'LOG')
  678     def test_ensure_shares_mounted_should_not_save_mounting_with_error(self,
  679                                                                        LOG):
  680         """_ensure_shares_mounted should not save share if failed to mount."""
  681         self._set_driver()
  682         drv = self._driver
  683         config_data = []
  684         config_data.append(self.TEST_NFS_EXPORT1)
  685         drv.configuration.nfs_shares_config = self.TEST_SHARES_CONFIG_FILE
  686         with mock.patch.object(
  687                 drv, '_read_config_file') as mock_read_config:
  688             with mock.patch.object(
  689                     drv, '_ensure_share_mounted') as mock_ensure:
  690                 mock_read_config.return_value = config_data
  691                 drv._ensure_share_mounted()
  692                 self.assertEqual(0, len(drv._mounted_shares))
  693                 mock_ensure.assert_called_once_with()
  694 
  695     def test_find_share_should_throw_error_if_there_is_no_mounted_share(self):
  696         """_find_share should throw error if there is no mounted shares."""
  697         self._set_driver()
  698         drv = self._driver
  699 
  700         drv._mounted_shares = []
  701 
  702         self.assertRaises(exception.NfsNoSharesMounted, drv._find_share,
  703                           self._simple_volume())
  704 
  705     def test_find_share(self):
  706         """_find_share simple use case."""
  707         self._set_driver()
  708         drv = self._driver
  709         drv._mounted_shares = [self.TEST_NFS_EXPORT1, self.TEST_NFS_EXPORT2]
  710 
  711         volume = fake_volume.fake_volume_obj(self.context,
  712                                              size=self.TEST_SIZE_IN_GB)
  713 
  714         with mock.patch.object(
  715                 drv, '_get_capacity_info') as mock_get_capacity_info:
  716             mock_get_capacity_info.side_effect = [
  717                 (5 * units.Gi, 2 * units.Gi, 2 * units.Gi),
  718                 (10 * units.Gi, 3 * units.Gi, 1 * units.Gi)]
  719             self.assertEqual(self.TEST_NFS_EXPORT2,
  720                              drv._find_share(volume))
  721             calls = [mock.call(self.TEST_NFS_EXPORT1),
  722                      mock.call(self.TEST_NFS_EXPORT2)]
  723             mock_get_capacity_info.assert_has_calls(calls)
  724             self.assertEqual(2, mock_get_capacity_info.call_count)
  725 
  726     def test_find_share_should_throw_error_if_there_is_not_enough_space(self):
  727         """_find_share should throw error if there is no share to host vol."""
  728         self._set_driver()
  729         drv = self._driver
  730         drv._mounted_shares = [self.TEST_NFS_EXPORT1, self.TEST_NFS_EXPORT2]
  731 
  732         with mock.patch.object(
  733                 drv, '_get_capacity_info') as mock_get_capacity_info:
  734             mock_get_capacity_info.side_effect = [
  735                 (5 * units.Gi, 0, 5 * units.Gi),
  736                 (10 * units.Gi, 0, 10 * units.Gi)]
  737 
  738             self.assertRaises(exception.NfsNoSuitableShareFound,
  739                               drv._find_share, self._simple_volume())
  740             calls = [mock.call(self.TEST_NFS_EXPORT1),
  741                      mock.call(self.TEST_NFS_EXPORT2)]
  742             mock_get_capacity_info.assert_has_calls(calls)
  743             self.assertEqual(2, mock_get_capacity_info.call_count)
  744 
  745     def _simple_volume(self, size=10):
  746         loc = self.TEST_NFS_EXPORT1
  747         return fake_volume.fake_volume_obj(self.context,
  748                                            display_name='volume_name',
  749                                            provider_location=loc,
  750                                            size=size)
  751 
  752     def _simple_encrypted_volume(self, size=10):
  753         loc = self.TEST_NFS_EXPORT1
  754         info_dic = {'name': u'volume-0000000a',
  755                     'id': '55555555-222f-4b32-b585-9991b3bf0a99',
  756                     'size': size,
  757                     'encryption_key_id': fake.ENCRYPTION_KEY_ID}
  758 
  759         return fake_volume.fake_volume_obj(self.context,
  760                                            provider_location=loc,
  761                                            **info_dic)
  762 
  763     def test_get_provisioned_capacity(self):
  764         self._set_driver()
  765         drv = self._driver
  766 
  767         mock_execute = self.mock_object(drv, '_execute')
  768         mock_execute.return_value = ("148418423\t/dir", "")
  769 
  770         with mock.patch.object(drv, 'shares') as shares:
  771             shares.keys.return_value = {'192.0.2.1:/srv/nfs1'}
  772             shares.return_value = {'192.0.2.1:/srv/nfs1', ''}
  773             ret = drv._get_provisioned_capacity()
  774 
  775             self.assertEqual(ret, 0.14)
  776 
  777     def test_create_sparsed_volume(self):
  778         self._set_driver()
  779         drv = self._driver
  780         volume = self._simple_volume()
  781 
  782         self.override_config('nfs_sparsed_volumes', True)
  783 
  784         with mock.patch.object(
  785                 drv, '_create_sparsed_file') as mock_create_sparsed_file:
  786             with mock.patch.object(
  787                     drv, '_set_rw_permissions') as mock_set_rw_permissions:
  788                 drv._do_create_volume(volume)
  789 
  790                 mock_create_sparsed_file.assert_called_once_with(mock.ANY,
  791                                                                  mock.ANY)
  792                 mock_set_rw_permissions.assert_called_once_with(mock.ANY)
  793 
  794     def test_create_nonsparsed_volume(self):
  795         self._set_driver()
  796         drv = self._driver
  797         self.configuration.nfs_sparsed_volumes = False
  798         volume = self._simple_volume()
  799 
  800         self.override_config('nfs_sparsed_volumes', False)
  801 
  802         with mock.patch.object(
  803                 drv, '_create_regular_file') as mock_create_regular_file:
  804             with mock.patch.object(
  805                     drv, '_set_rw_permissions') as mock_set_rw_permissions:
  806                 drv._do_create_volume(volume)
  807 
  808                 mock_create_regular_file.assert_called_once_with(mock.ANY,
  809                                                                  mock.ANY)
  810                 mock_set_rw_permissions.assert_called_once_with(mock.ANY)
  811 
  812     @mock.patch.object(nfs, 'LOG')
  813     def test_create_volume_should_ensure_nfs_mounted(self, mock_log):
  814         """create_volume ensures shares provided in config are mounted."""
  815         self._set_driver()
  816         drv = self._driver
  817         drv._find_share = mock.Mock()
  818         drv._find_share.return_value = self.TEST_NFS_EXPORT1
  819         drv._do_create_volume = mock.Mock()
  820 
  821         with mock.patch.object(
  822                 drv, '_ensure_share_mounted') as mock_ensure_share:
  823             drv._ensure_share_mounted()
  824             volume = fake_volume.fake_volume_obj(self.context,
  825                                                  size=self.TEST_SIZE_IN_GB)
  826             drv.create_volume(volume)
  827 
  828             mock_ensure_share.assert_called_once_with()
  829 
  830     @mock.patch.object(nfs, 'LOG')
  831     def test_create_volume_should_return_provider_location(self, mock_log):
  832         """create_volume should return provider_location with found share."""
  833         self._set_driver()
  834         drv = self._driver
  835         drv._ensure_shares_mounted = mock.Mock()
  836         drv._do_create_volume = mock.Mock()
  837 
  838         with mock.patch.object(drv, '_find_share') as mock_find_share:
  839             mock_find_share.return_value = self.TEST_NFS_EXPORT1
  840             volume = fake_volume.fake_volume_obj(self.context,
  841                                                  size=self.TEST_SIZE_IN_GB)
  842             result = drv.create_volume(volume)
  843             self.assertEqual(self.TEST_NFS_EXPORT1,
  844                              result['provider_location'])
  845             mock_find_share.assert_called_once_with(volume)
  846 
  847     def test_delete_volume(self):
  848         """delete_volume simple test case."""
  849         self._set_driver()
  850         drv = self._driver
  851         drv._ensure_share_mounted = mock.Mock()
  852 
  853         volume = fake_volume.fake_volume_obj(
  854             self.context,
  855             display_name='volume-123',
  856             provider_location=self.TEST_NFS_EXPORT1)
  857 
  858         with mock.patch.object(drv, 'local_path') as mock_local_path:
  859             mock_local_path.return_value = self.TEST_LOCAL_PATH
  860             drv.delete_volume(volume)
  861             mock_local_path.assert_called_with(volume)
  862             drv._execute.assert_called_once()
  863 
  864     def test_delete_should_ensure_share_mounted(self):
  865         """delete_volume should ensure that corresponding share is mounted."""
  866         self._set_driver()
  867         drv = self._driver
  868         volume = fake_volume.fake_volume_obj(
  869             self.context,
  870             display_name='volume-123',
  871             provider_location=self.TEST_NFS_EXPORT1)
  872 
  873         with mock.patch.object(drv, '_ensure_share_mounted'):
  874             drv.delete_volume(volume)
  875 
  876     def test_delete_should_not_delete_if_provider_location_not_provided(self):
  877         """delete_volume shouldn't delete if provider_location missed."""
  878         self._set_driver()
  879         drv = self._driver
  880         volume = fake_volume.fake_volume_obj(self.context,
  881                                              name='volume-123',
  882                                              provider_location=None)
  883 
  884         with mock.patch.object(drv, '_ensure_share_mounted'):
  885             drv.delete_volume(volume)
  886             self.assertFalse(drv._execute.called)
  887 
  888     def test_get_volume_stats(self):
  889         """get_volume_stats must fill the correct values."""
  890         self._set_driver()
  891         drv = self._driver
  892         drv._mounted_shares = [self.TEST_NFS_EXPORT1, self.TEST_NFS_EXPORT2]
  893 
  894         with mock.patch.object(
  895                 drv, '_ensure_shares_mounted') as mock_ensure_share:
  896             with mock.patch.object(
  897                     drv, '_get_capacity_info') as mock_get_capacity_info:
  898                 mock_get_capacity_info.side_effect = [
  899                     (10 * units.Gi, 2 * units.Gi, 2 * units.Gi),
  900                     (20 * units.Gi, 3 * units.Gi, 3 * units.Gi)]
  901 
  902                 drv._ensure_shares_mounted()
  903                 drv.get_volume_stats()
  904 
  905                 calls = [mock.call(self.TEST_NFS_EXPORT1),
  906                          mock.call(self.TEST_NFS_EXPORT2)]
  907                 mock_get_capacity_info.assert_has_calls(calls)
  908 
  909                 self.assertTrue(mock_ensure_share.called)
  910                 self.assertEqual(30.0, drv._stats['total_capacity_gb'])
  911                 self.assertEqual(5.0, drv._stats['free_capacity_gb'])
  912                 self.assertEqual(5, drv._stats['reserved_percentage'])
  913                 self.assertTrue(drv._stats['sparse_copy_volume'])
  914 
  915     def test_get_volume_stats_with_non_zero_reserved_percentage(self):
  916         """get_volume_stats must fill the correct values."""
  917         self.configuration.reserved_percentage = 10.0
  918         drv = nfs.NfsDriver(configuration=self.configuration)
  919 
  920         drv._mounted_shares = [self.TEST_NFS_EXPORT1, self.TEST_NFS_EXPORT2]
  921 
  922         with mock.patch.object(
  923                 drv, '_ensure_shares_mounted') as mock_ensure_share:
  924             with mock.patch.object(
  925                     drv, '_get_capacity_info') as mock_get_capacity_info:
  926                 mock_get_capacity_info.side_effect = [
  927                     (10 * units.Gi, 2 * units.Gi, 2 * units.Gi),
  928                     (20 * units.Gi, 3 * units.Gi, 3 * units.Gi)]
  929 
  930                 drv._ensure_shares_mounted()
  931                 drv.get_volume_stats()
  932 
  933                 calls = [mock.call(self.TEST_NFS_EXPORT1),
  934                          mock.call(self.TEST_NFS_EXPORT2)]
  935                 mock_get_capacity_info.assert_has_calls(calls)
  936 
  937                 self.assertTrue(mock_ensure_share.called)
  938                 self.assertEqual(30.0, drv._stats['total_capacity_gb'])
  939                 self.assertEqual(5.0, drv._stats['free_capacity_gb'])
  940                 self.assertEqual(10.0, drv._stats['reserved_percentage'])
  941 
  942     @ddt.data(True, False)
  943     def test_update_volume_stats(self, thin):
  944         self._set_driver()
  945         self._driver.configuration.max_over_subscription_ratio = 20.0
  946         self._driver.configuration.reserved_percentage = 5.0
  947         self._driver.configuration.nfs_sparsed_volumes = thin
  948 
  949         remotefs_volume_stats = {
  950             'volume_backend_name': 'fake_backend_name',
  951             'vendor_name': 'fake_vendor',
  952             'driver_version': 'fake_version',
  953             'storage_protocol': 'NFS',
  954             'total_capacity_gb': 100.0,
  955             'free_capacity_gb': 20.0,
  956             'reserved_percentage': 5.0,
  957             'QoS_support': False,
  958         }
  959         self.mock_object(remotefs.RemoteFSDriver, '_update_volume_stats')
  960         self._driver._stats = remotefs_volume_stats
  961 
  962         mock_get_provisioned_capacity = self.mock_object(
  963             self._driver, '_get_provisioned_capacity', return_value=25.0)
  964 
  965         self._driver._update_volume_stats()
  966 
  967         nfs_added_volume_stats = {
  968             'provisioned_capacity_gb': 25.0 if thin else 80.0,
  969             'max_over_subscription_ratio': 20.0,
  970             'reserved_percentage': 5.0,
  971             'thin_provisioning_support': thin,
  972             'thick_provisioning_support': not thin,
  973         }
  974         expected = remotefs_volume_stats
  975         expected.update(nfs_added_volume_stats)
  976 
  977         self.assertEqual(expected, self._driver._stats)
  978         self.assertEqual(thin, mock_get_provisioned_capacity.called)
  979 
  980     def _check_is_share_eligible(self, total_size, total_available,
  981                                  total_allocated, requested_volume_size):
  982         self._set_driver()
  983         with mock.patch.object(self._driver, '_get_capacity_info')\
  984                 as mock_get_capacity_info:
  985             mock_get_capacity_info.return_value = (total_size,
  986                                                    total_available,
  987                                                    total_allocated)
  988             return self._driver._is_share_eligible('fake_share',
  989                                                    requested_volume_size)
  990 
  991     def test_is_share_eligible(self):
  992         self._set_driver()
  993         total_size = 100.0 * units.Gi
  994         total_available = 90.0 * units.Gi
  995         total_allocated = 10.0 * units.Gi
  996         requested_volume_size = 1  # GiB
  997 
  998         self.assertTrue(self._check_is_share_eligible(total_size,
  999                                                       total_available,
 1000                                                       total_allocated,
 1001                                                       requested_volume_size))
 1002 
 1003     def test_share_eligibility_with_reserved_percentage(self):
 1004         self._set_driver()
 1005         total_size = 100.0 * units.Gi
 1006         total_available = 4.0 * units.Gi
 1007         total_allocated = 96.0 * units.Gi
 1008         requested_volume_size = 1  # GiB
 1009 
 1010         # Check used > used_ratio statement entered
 1011         self.assertFalse(self._check_is_share_eligible(total_size,
 1012                                                        total_available,
 1013                                                        total_allocated,
 1014                                                        requested_volume_size))
 1015 
 1016     def test_is_share_eligible_above_oversub_ratio(self):
 1017         self._set_driver()
 1018         total_size = 100.0 * units.Gi
 1019         total_available = 10.0 * units.Gi
 1020         total_allocated = 90.0 * units.Gi
 1021         requested_volume_size = 10  # GiB
 1022 
 1023         # Check apparent_available <= requested_volume_size statement entered
 1024         self.assertFalse(self._check_is_share_eligible(total_size,
 1025                                                        total_available,
 1026                                                        total_allocated,
 1027                                                        requested_volume_size))
 1028 
 1029     def test_is_share_eligible_reserved_space_above_oversub_ratio(self):
 1030         self._set_driver()
 1031         total_size = 100.0 * units.Gi
 1032         total_available = 10.0 * units.Gi
 1033         total_allocated = 100.0 * units.Gi
 1034         requested_volume_size = 1  # GiB
 1035 
 1036         # Check total_allocated / total_size >= oversub_ratio
 1037         # statement entered
 1038         self.assertFalse(self._check_is_share_eligible(total_size,
 1039                                                        total_available,
 1040                                                        total_allocated,
 1041                                                        requested_volume_size))
 1042 
 1043     def test_extend_volume(self):
 1044         """Extend a volume by 1."""
 1045         self._set_driver()
 1046         drv = self._driver
 1047         volume = fake_volume.fake_volume_obj(
 1048             self.context,
 1049             id='80ee16b6-75d2-4d54-9539-ffc1b4b0fb10',
 1050             size=1,
 1051             provider_location='nfs_share')
 1052         path = 'path'
 1053         newSize = volume['size'] + 1
 1054 
 1055         with mock.patch.object(image_utils, 'resize_image') as resize:
 1056             with mock.patch.object(drv, 'local_path', return_value=path):
 1057                 with mock.patch.object(drv, '_is_share_eligible',
 1058                                        return_value=True):
 1059                     with mock.patch.object(drv, '_is_file_size_equal',
 1060                                            return_value=True):
 1061                         drv.extend_volume(volume, newSize)
 1062 
 1063                         resize.assert_called_once_with(path, newSize,
 1064                                                        run_as_root=True)
 1065 
 1066     def test_extend_volume_attached_fail(self):
 1067         """Extend a volume by 1."""
 1068         self._set_driver()
 1069         drv = self._driver
 1070         volume = fake_volume.fake_volume_obj(
 1071             self.context,
 1072             id='80ee16b6-75d2-4d54-9539-ffc1b4b0fb10',
 1073             size=1,
 1074             provider_location='nfs_share')
 1075         path = 'path'
 1076         newSize = volume['size'] + 1
 1077 
 1078         with mock.patch.object(drv, 'local_path', return_value=path):
 1079             with mock.patch.object(drv, '_is_share_eligible',
 1080                                    return_value=True):
 1081                 with mock.patch.object(drv, '_is_file_size_equal',
 1082                                        return_value=True):
 1083                     with mock.patch.object(drv, '_is_volume_attached',
 1084                                            return_value=True):
 1085                         self.assertRaises(exception.ExtendVolumeError,
 1086                                           drv.extend_volume, volume, newSize)
 1087 
 1088     def test_extend_volume_failure(self):
 1089         """Error during extend operation."""
 1090         self._set_driver()
 1091         drv = self._driver
 1092         volume = fake_volume.fake_volume_obj(
 1093             self.context,
 1094             id='80ee16b6-75d2-4d54-9539-ffc1b4b0fb10',
 1095             size=1,
 1096             provider_location='nfs_share')
 1097 
 1098         with mock.patch.object(image_utils, 'resize_image'):
 1099             with mock.patch.object(drv, 'local_path', return_value='path'):
 1100                 with mock.patch.object(drv, '_is_share_eligible',
 1101                                        return_value=True):
 1102                     with mock.patch.object(drv, '_is_file_size_equal',
 1103                                            return_value=False):
 1104                         self.assertRaises(exception.ExtendVolumeError,
 1105                                           drv.extend_volume, volume, 2)
 1106 
 1107     def test_extend_volume_insufficient_space(self):
 1108         """Insufficient space on nfs_share during extend operation."""
 1109         self._set_driver()
 1110         drv = self._driver
 1111         volume = fake_volume.fake_volume_obj(
 1112             self.context,
 1113             id='80ee16b6-75d2-4d54-9539-ffc1b4b0fb10',
 1114             size=1,
 1115             provider_location='nfs_share')
 1116 
 1117         with mock.patch.object(image_utils, 'resize_image'):
 1118             with mock.patch.object(drv, 'local_path', return_value='path'):
 1119                 with mock.patch.object(drv, '_is_share_eligible',
 1120                                        return_value=False):
 1121                     with mock.patch.object(drv, '_is_file_size_equal',
 1122                                            return_value=False):
 1123                         self.assertRaises(exception.ExtendVolumeError,
 1124                                           drv.extend_volume, volume, 2)
 1125 
 1126     def test_is_file_size_equal(self):
 1127         """File sizes are equal."""
 1128         self._set_driver()
 1129         drv = self._driver
 1130         path = 'fake/path'
 1131         size = 2
 1132         data = mock.MagicMock()
 1133         data.virtual_size = size * units.Gi
 1134 
 1135         with mock.patch.object(image_utils, 'qemu_img_info',
 1136                                return_value=data):
 1137             self.assertTrue(drv._is_file_size_equal(path, size))
 1138 
 1139     def test_is_file_size_equal_false(self):
 1140         """File sizes are not equal."""
 1141         self._set_driver()
 1142         drv = self._driver
 1143         path = 'fake/path'
 1144         size = 2
 1145         data = mock.MagicMock()
 1146         data.virtual_size = (size + 1) * units.Gi
 1147 
 1148         with mock.patch.object(image_utils, 'qemu_img_info',
 1149                                return_value=data):
 1150             self.assertFalse(drv._is_file_size_equal(path, size))
 1151 
 1152     @mock.patch.object(nfs, 'LOG')
 1153     def test_set_nas_security_options_when_true(self, LOG):
 1154         """Test higher level setting of NAS Security options.
 1155 
 1156         The NFS driver overrides the base method with a driver specific
 1157         version.
 1158         """
 1159         self._set_driver()
 1160         drv = self._driver
 1161         drv._mounted_shares = [self.TEST_NFS_EXPORT1]
 1162         is_new_install = True
 1163 
 1164         drv._ensure_shares_mounted = mock.Mock()
 1165         drv._get_mount_point_for_share = mock.Mock(
 1166             return_value=self.TEST_MNT_POINT)
 1167         drv._determine_nas_security_option_setting = mock.Mock(
 1168             return_value='true')
 1169 
 1170         drv.set_nas_security_options(is_new_install)
 1171 
 1172         self.assertEqual('true', drv.configuration.nas_secure_file_operations)
 1173         self.assertEqual('true', drv.configuration.nas_secure_file_permissions)
 1174         self.assertFalse(LOG.warning.called)
 1175 
 1176     @mock.patch.object(nfs, 'LOG')
 1177     def test_set_nas_security_options_when_false(self, LOG):
 1178         """Test higher level setting of NAS Security options.
 1179 
 1180         The NFS driver overrides the base method with a driver specific
 1181         version.
 1182         """
 1183         self._set_driver()
 1184         drv = self._driver
 1185         drv._mounted_shares = [self.TEST_NFS_EXPORT1]
 1186         is_new_install = False
 1187 
 1188         drv._ensure_shares_mounted = mock.Mock()
 1189         drv._get_mount_point_for_share = mock.Mock(
 1190             return_value=self.TEST_MNT_POINT)
 1191         drv._determine_nas_security_option_setting = mock.Mock(
 1192             return_value='false')
 1193 
 1194         drv.set_nas_security_options(is_new_install)
 1195 
 1196         self.assertEqual('false', drv.configuration.nas_secure_file_operations)
 1197         self.assertEqual('false',
 1198                          drv.configuration.nas_secure_file_permissions)
 1199         self.assertTrue(LOG.warning.called)
 1200 
 1201     def test_set_nas_security_options_exception_if_no_mounted_shares(self):
 1202         """Ensure proper exception is raised if there are no mounted shares."""
 1203 
 1204         self._set_driver()
 1205         drv = self._driver
 1206         drv._ensure_shares_mounted = mock.Mock()
 1207         drv._mounted_shares = []
 1208         is_new_cinder_install = 'does not matter'
 1209 
 1210         self.assertRaises(exception.NfsNoSharesMounted,
 1211                           drv.set_nas_security_options,
 1212                           is_new_cinder_install)
 1213 
 1214     def test_ensure_share_mounted(self):
 1215         """Case where the mount works the first time."""
 1216 
 1217         self._set_driver()
 1218         self.mock_object(self._driver._remotefsclient, 'mount', autospec=True)
 1219         drv = self._driver
 1220         drv.configuration.nfs_mount_attempts = 3
 1221         drv.shares = {self.TEST_NFS_EXPORT1: ''}
 1222 
 1223         drv._ensure_share_mounted(self.TEST_NFS_EXPORT1)
 1224 
 1225         drv._remotefsclient.mount.assert_called_once_with(
 1226             self.TEST_NFS_EXPORT1, [])
 1227 
 1228     @mock.patch('time.sleep')
 1229     def test_ensure_share_mounted_exception(self, _mock_sleep):
 1230         """Make the configured number of attempts when mounts fail."""
 1231 
 1232         num_attempts = 3
 1233 
 1234         self._set_driver()
 1235         self.mock_object(self._driver._remotefsclient, 'mount',
 1236                          side_effect=Exception)
 1237         drv = self._driver
 1238         drv.configuration.nfs_mount_attempts = num_attempts
 1239         drv.shares = {self.TEST_NFS_EXPORT1: ''}
 1240 
 1241         self.assertRaises(exception.NfsException, drv._ensure_share_mounted,
 1242                           self.TEST_NFS_EXPORT1)
 1243 
 1244         self.assertEqual(num_attempts, drv._remotefsclient.mount.call_count)
 1245 
 1246     def test_ensure_share_mounted_at_least_one_attempt(self):
 1247         """Make at least one mount attempt even if configured for less."""
 1248 
 1249         min_num_attempts = 1
 1250         num_attempts = 0
 1251         self._set_driver()
 1252         self.mock_object(self._driver._remotefsclient, 'mount',
 1253                          side_effect=Exception)
 1254         drv = self._driver
 1255         drv.configuration.nfs_mount_attempts = num_attempts
 1256         drv.shares = {self.TEST_NFS_EXPORT1: ''}
 1257 
 1258         self.assertRaises(exception.NfsException, drv._ensure_share_mounted,
 1259                           self.TEST_NFS_EXPORT1)
 1260 
 1261         self.assertEqual(min_num_attempts,
 1262                          drv._remotefsclient.mount.call_count)
 1263 
 1264     @mock.patch('tempfile.NamedTemporaryFile')
 1265     @ddt.data([NFS_CONFIG1, QEMU_IMG_INFO_OUT3, False],
 1266               [NFS_CONFIG2, QEMU_IMG_INFO_OUT4, False],
 1267               [NFS_CONFIG3, QEMU_IMG_INFO_OUT3, False],
 1268               [NFS_CONFIG4, QEMU_IMG_INFO_OUT4, False],
 1269               [NFS_CONFIG4, QEMU_IMG_INFO_OUT5, True])
 1270     @ddt.unpack
 1271     def test_copy_volume_from_snapshot(self, nfs_conf, qemu_img_info,
 1272                                        encryption, mock_temp_file):
 1273 
 1274         class DictObj(object):
 1275             # convert a dict to object w/ attributes
 1276             def __init__(self, d):
 1277                 self.__dict__ = d
 1278 
 1279         self._set_driver(extra_confs=nfs_conf)
 1280         drv = self._driver
 1281 
 1282         src_encryption_key_id = None
 1283         dest_encryption_key_id = None
 1284 
 1285         if encryption:
 1286             mock_temp_file.return_value.__enter__.side_effect = [
 1287                 DictObj({'name': '/tmp/imgfile'}),
 1288                 DictObj({'name': '/tmp/passfile'})]
 1289 
 1290             dest_volume = self._simple_encrypted_volume()
 1291             src_volume = self._simple_encrypted_volume()
 1292 
 1293             key_mgr = fake_keymgr.fake_api()
 1294             self.mock_object(castellan.key_manager, 'API',
 1295                              return_value=key_mgr)
 1296             key_id = key_mgr.store(self.context, KeyObject())
 1297 
 1298             src_volume.encryption_key_id = key_id
 1299             dest_volume.encryption_key_id = key_id
 1300 
 1301             src_encryption_key_id = src_volume.encryption_key_id
 1302             dest_encryption_key_id = dest_volume.encryption_key_id
 1303         else:
 1304             dest_volume = self._simple_volume()
 1305             src_volume = self._simple_volume()
 1306 
 1307         fake_snap = fake_snapshot.fake_snapshot_obj(self.context)
 1308         fake_snap.volume = src_volume
 1309 
 1310         img_out = qemu_img_info % {'volid': src_volume.id,
 1311                                    'snapid': fake_snap.id,
 1312                                    'size_gb': src_volume.size,
 1313                                    'size_b': src_volume.size * units.Gi}
 1314 
 1315         img_info = imageutils.QemuImgInfo(img_out)
 1316         mock_img_info = self.mock_object(image_utils, 'qemu_img_info')
 1317         mock_img_info.return_value = img_info
 1318         mock_convert_image = self.mock_object(image_utils, 'convert_image')
 1319 
 1320         vol_dir = os.path.join(self.TEST_MNT_POINT_BASE,
 1321                                drv._get_hash_str(src_volume.provider_location))
 1322         src_vol_path = os.path.join(vol_dir, img_info.backing_file)
 1323         dest_vol_path = os.path.join(vol_dir, dest_volume.name)
 1324         info_path = os.path.join(vol_dir, src_volume.name) + '.info'
 1325 
 1326         snap_file = dest_volume.name + '.' + fake_snap.id
 1327         snap_path = os.path.join(vol_dir, snap_file)
 1328         size = dest_volume.size
 1329 
 1330         mock_read_info_file = self.mock_object(drv, '_read_info_file')
 1331         mock_read_info_file.return_value = {'active': snap_file,
 1332                                             fake_snap.id: snap_file}
 1333 
 1334         mock_permission = self.mock_object(drv, '_set_rw_permissions_for_all')
 1335 
 1336         drv._copy_volume_from_snapshot(fake_snap, dest_volume, size,
 1337                                        src_encryption_key_id,
 1338                                        dest_encryption_key_id)
 1339 
 1340         mock_read_info_file.assert_called_once_with(info_path)
 1341         mock_img_info.assert_called_once_with(snap_path,
 1342                                               force_share=True,
 1343                                               run_as_root=True)
 1344         used_qcow = nfs_conf['nfs_qcow2_volumes']
 1345         if encryption:
 1346             mock_convert_image.assert_called_once_with(
 1347                 src_vol_path, dest_vol_path, 'luks',
 1348                 passphrase_file='/tmp/passfile',
 1349                 run_as_root=True,
 1350                 src_passphrase_file='/tmp/imgfile')
 1351         else:
 1352             mock_convert_image.assert_called_once_with(
 1353                 src_vol_path, dest_vol_path, 'qcow2' if used_qcow else 'raw',
 1354                 run_as_root=True)
 1355         mock_permission.assert_called_once_with(dest_vol_path)
 1356 
 1357     @ddt.data([NFS_CONFIG1, QEMU_IMG_INFO_OUT3, 'available'],
 1358               [NFS_CONFIG2, QEMU_IMG_INFO_OUT4, 'backing-up'],
 1359               [NFS_CONFIG3, QEMU_IMG_INFO_OUT3, 'available'],
 1360               [NFS_CONFIG4, QEMU_IMG_INFO_OUT4, 'backing-up'])
 1361     @ddt.unpack
 1362     def test_create_volume_from_snapshot(self, nfs_conf, qemu_img_info,
 1363                                          snap_status):
 1364         self._set_driver(extra_confs=nfs_conf)
 1365         drv = self._driver
 1366 
 1367         # Volume source of the snapshot we are trying to clone from. We need it
 1368         # to have a different id than the default provided.
 1369         src_volume = self._simple_volume(size=10)
 1370         src_volume.id = fake.VOLUME_ID
 1371         src_volume_dir = os.path.join(self.TEST_MNT_POINT_BASE,
 1372                                       drv._get_hash_str(
 1373                                           src_volume.provider_location))
 1374         src_volume_path = os.path.join(src_volume_dir, src_volume.name)
 1375         fake_snap = fake_snapshot.fake_snapshot_obj(self.context)
 1376 
 1377         # Fake snapshot based in the previous created volume
 1378         snap_file = src_volume.name + '.' + fake_snap.id
 1379         fake_snap.volume = src_volume
 1380         fake_snap.status = snap_status
 1381         fake_snap.size = 10
 1382 
 1383         # New fake volume where the snap will be copied
 1384         new_volume = self._simple_volume(size=10)
 1385         new_volume_dir = os.path.join(self.TEST_MNT_POINT_BASE,
 1386                                       drv._get_hash_str(
 1387                                           src_volume.provider_location))
 1388         new_volume_path = os.path.join(new_volume_dir, new_volume.name)
 1389 
 1390         # Mocks
 1391         img_out = qemu_img_info % {'volid': src_volume.id,
 1392                                    'snapid': fake_snap.id,
 1393                                    'size_gb': src_volume.size,
 1394                                    'size_b': src_volume.size * units.Gi}
 1395         img_info = imageutils.QemuImgInfo(img_out)
 1396         mock_img_info = self.mock_object(image_utils, 'qemu_img_info')
 1397         mock_img_info.return_value = img_info
 1398 
 1399         mock_ensure = self.mock_object(drv, '_ensure_shares_mounted')
 1400         mock_find_share = self.mock_object(drv, '_find_share',
 1401                                            return_value=self.TEST_NFS_EXPORT1)
 1402         mock_read_info_file = self.mock_object(drv, '_read_info_file')
 1403         mock_read_info_file.return_value = {'active': snap_file,
 1404                                             fake_snap.id: snap_file}
 1405         mock_convert_image = self.mock_object(image_utils, 'convert_image')
 1406         self.mock_object(drv, '_create_qcow2_file')
 1407         self.mock_object(drv, '_create_regular_file')
 1408         self.mock_object(drv, '_create_regular_file')
 1409         self.mock_object(drv, '_set_rw_permissions')
 1410         self.mock_object(drv, '_read_file')
 1411 
 1412         ret = drv.create_volume_from_snapshot(new_volume, fake_snap)
 1413 
 1414         # Test asserts
 1415         self.assertEqual(self.TEST_NFS_EXPORT1, ret['provider_location'])
 1416         used_qcow = nfs_conf['nfs_qcow2_volumes']
 1417         mock_convert_image.assert_called_once_with(
 1418             src_volume_path, new_volume_path, 'qcow2' if used_qcow else 'raw',
 1419             run_as_root=True)
 1420         mock_ensure.assert_called_once()
 1421         mock_find_share.assert_called_once_with(new_volume)
 1422 
 1423     @ddt.data('error', 'creating', 'deleting', 'deleted', 'updating',
 1424               'error_deleting', 'unmanaging', 'restoring')
 1425     def test_create_volume_from_snapshot_invalid_status(self, snap_status):
 1426         """Expect an error when the snapshot's status is not 'available'."""
 1427         self._set_driver()
 1428         drv = self._driver
 1429 
 1430         src_volume = self._simple_volume()
 1431 
 1432         fake_snap = fake_snapshot.fake_snapshot_obj(self.context)
 1433         fake_snap.volume = src_volume
 1434         fake_snap.status = snap_status
 1435 
 1436         new_volume = self._simple_volume()
 1437         new_volume['size'] = fake_snap['volume_size']
 1438 
 1439         self.assertRaises(exception.InvalidSnapshot,
 1440                           drv.create_volume_from_snapshot,
 1441                           new_volume,
 1442                           fake_snap)
 1443 
 1444     @ddt.data([NFS_CONFIG1, QEMU_IMG_INFO_OUT1],
 1445               [NFS_CONFIG2, QEMU_IMG_INFO_OUT2],
 1446               [NFS_CONFIG3, QEMU_IMG_INFO_OUT1],
 1447               [NFS_CONFIG4, QEMU_IMG_INFO_OUT2])
 1448     @ddt.unpack
 1449     def test_initialize_connection(self, nfs_confs, qemu_img_info):
 1450         self._set_driver(extra_confs=nfs_confs)
 1451         drv = self._driver
 1452 
 1453         volume = self._simple_volume()
 1454         vol_dir = os.path.join(self.TEST_MNT_POINT_BASE,
 1455                                drv._get_hash_str(volume.provider_location))
 1456         vol_path = os.path.join(vol_dir, volume.name)
 1457 
 1458         mock_img_utils = self.mock_object(image_utils, 'qemu_img_info')
 1459         img_out = qemu_img_info % {'volid': volume.id, 'size_gb': volume.size,
 1460                                    'size_b': volume.size * units.Gi}
 1461         mock_img_utils.return_value = imageutils.QemuImgInfo(img_out)
 1462         self.mock_object(drv, '_read_info_file',
 1463                          return_value={'active': "volume-%s" % volume.id})
 1464 
 1465         conn_info = drv.initialize_connection(volume, None)
 1466 
 1467         mock_img_utils.assert_called_once_with(vol_path,
 1468                                                force_share=True,
 1469                                                run_as_root=True)
 1470         self.assertEqual('nfs', conn_info['driver_volume_type'])
 1471         self.assertEqual(volume.name, conn_info['data']['name'])
 1472         self.assertEqual(self.TEST_MNT_POINT_BASE,
 1473                          conn_info['mount_point_base'])
 1474 
 1475     @mock.patch.object(image_utils, 'qemu_img_info')
 1476     def test_initialize_connection_raise_exception(self, mock_img_info):
 1477         self._set_driver()
 1478         drv = self._driver
 1479         volume = self._simple_volume()
 1480 
 1481         qemu_img_output = """image: %s
 1482         file format: iso
 1483         virtual size: 1.0G (1073741824 bytes)
 1484         disk size: 173K
 1485         """ % volume['name']
 1486         mock_img_info.return_value = imageutils.QemuImgInfo(qemu_img_output)
 1487 
 1488         self.assertRaises(exception.InvalidVolume,
 1489                           drv.initialize_connection,
 1490                           volume,
 1491                           None)
 1492 
 1493     def test_create_snapshot(self):
 1494         self._set_driver()
 1495         drv = self._driver
 1496         volume = self._simple_volume()
 1497         self.configuration.nfs_snapshot_support = True
 1498         fake_snap = fake_snapshot.fake_snapshot_obj(self.context)
 1499         fake_snap.volume = volume
 1500         vol_dir = os.path.join(self.TEST_MNT_POINT_BASE,
 1501                                drv._get_hash_str(self.TEST_NFS_EXPORT1))
 1502         snap_file = volume['name'] + '.' + fake_snap.id
 1503         snap_path = os.path.join(vol_dir, snap_file)
 1504         info_path = os.path.join(vol_dir, volume['name']) + '.info'
 1505 
 1506         with mock.patch.object(drv, '_local_path_volume_info',
 1507                                return_value=info_path), \
 1508                 mock.patch.object(drv, '_read_info_file', return_value={}), \
 1509                 mock.patch.object(drv, '_do_create_snapshot') \
 1510                 as mock_do_create_snapshot, \
 1511                 mock.patch.object(drv, '_write_info_file') \
 1512                 as mock_write_info_file, \
 1513                 mock.patch.object(drv, 'get_active_image_from_info',
 1514                                   return_value=volume['name']), \
 1515                 mock.patch.object(drv, '_get_new_snap_path',
 1516                                   return_value=snap_path):
 1517             self._driver.create_snapshot(fake_snap)
 1518 
 1519         mock_do_create_snapshot.assert_called_with(fake_snap, volume['name'],
 1520                                                    snap_path)
 1521         mock_write_info_file.assert_called_with(
 1522             info_path, {'active': snap_file, fake_snap.id: snap_file})
 1523 
 1524     @ddt.data({'volume_status': 'available',
 1525                'original_provider': 'original_provider',
 1526                'rename_side_effect': None},
 1527               {'volume_status': 'available',
 1528                'original_provider': 'current_provider',
 1529                'rename_side_effect': None},
 1530               {'volume_status': 'in-use',
 1531                'original_provider': 'original_provider',
 1532                'rename_side_effect': None},
 1533               {'volume_status': 'available',
 1534                'original_provider': 'original_provider',
 1535                'rename_side_effect': OSError})
 1536     @ddt.unpack
 1537     @mock.patch('os.rename')
 1538     def test_update_migrated_volume(self,
 1539                                     mock_rename,
 1540                                     rename_side_effect,
 1541                                     original_provider,
 1542                                     volume_status):
 1543         drv = nfs.NfsDriver(configuration=self.configuration)
 1544         base_dir = '/dir_base/'
 1545         current_path = base_dir + fake.VOLUME2_NAME
 1546         current_provider = 'current_provider'
 1547         mock_rename.side_effect = rename_side_effect
 1548 
 1549         volume = fake_volume.fake_volume_obj(
 1550             self.context,
 1551             id=fake.VOLUME_ID,
 1552             provider_location=original_provider,
 1553             _name_id=None)
 1554 
 1555         new_volume = fake_volume.fake_volume_obj(
 1556             self.context,
 1557             id=fake.VOLUME2_ID,
 1558             provider_location=current_provider,
 1559             _name_id=None)
 1560 
 1561         with mock.patch.object(drv, 'local_path') as local_path:
 1562             local_path.return_value = current_path
 1563             update = drv.update_migrated_volume(self.context,
 1564                                                 volume,
 1565                                                 new_volume,
 1566                                                 volume_status)
 1567 
 1568             if (volume_status == 'available' and
 1569                     original_provider != current_provider):
 1570                 original_path = base_dir + fake.VOLUME_NAME
 1571                 mock_rename.assert_called_once_with(current_path,
 1572                                                     original_path)
 1573             else:
 1574                 mock_rename.assert_not_called()
 1575 
 1576             if mock_rename.call_count > 0 and rename_side_effect is None:
 1577                 self.assertEqual({'_name_id': None,
 1578                                   'provider_location': current_provider},
 1579                                  update)
 1580             else:
 1581                 self.assertEqual({'_name_id': fake.VOLUME2_ID,
 1582                                   'provider_location': current_provider},
 1583                                  update)
 1584 
 1585 
 1586 class NfsDriverDoSetupTestCase(test.TestCase):
 1587 
 1588     def setUp(self):
 1589         super(NfsDriverDoSetupTestCase, self).setUp()
 1590         self.context = mock.Mock()
 1591         self.create_configuration()
 1592         self.override_config('compute_api_class', 'unittest.mock.Mock')
 1593 
 1594     def create_configuration(self):
 1595         config = conf.Configuration(None)
 1596         config.append_config_values(nfs.nfs_opts)
 1597         self.configuration = config
 1598 
 1599     def test_setup_should_throw_error_if_shares_config_not_configured(self):
 1600         """do_setup should throw error if shares config is not configured."""
 1601 
 1602         self.override_config('nfs_shares_config', None)
 1603         drv = nfs.NfsDriver(configuration=self.configuration)
 1604 
 1605         mock_os_path_exists = self.mock_object(os.path, 'exists')
 1606 
 1607         with self.assertRaisesRegex(exception.NfsException,
 1608                                     ".*no NFS config file configured.*"):
 1609             drv.do_setup(self.context)
 1610 
 1611         self.assertEqual(0, mock_os_path_exists.call_count)
 1612 
 1613     def test_setup_should_throw_error_if_shares_file_does_not_exist(self):
 1614         """do_setup should throw error if shares file does not exist."""
 1615 
 1616         drv = nfs.NfsDriver(configuration=self.configuration)
 1617 
 1618         mock_os_path_exists = self.mock_object(os.path, 'exists')
 1619         mock_os_path_exists.return_value = False
 1620 
 1621         with self.assertRaisesRegex(exception.NfsException,
 1622                                     "NFS config file.*doesn't exist"):
 1623             drv.do_setup(self.context)
 1624 
 1625         mock_os_path_exists.assert_has_calls(
 1626             [mock.call(self.configuration.nfs_shares_config)])
 1627 
 1628     def test_setup_should_not_throw_error_if_host_and_share_set(self):
 1629         """do_setup shouldn't throw shares file error if host and share set."""
 1630 
 1631         drv = nfs.NfsDriver(configuration=self.configuration)
 1632 
 1633         self.override_config('nas_host', 'nfs-host1')
 1634         self.override_config('nas_share_path', '/export')
 1635         mock_os_path_exists = self.mock_object(os.path, 'exists')
 1636         mock_os_path_exists.return_value = False
 1637         mock_set_nas_sec_options = self.mock_object(nfs.NfsDriver,
 1638                                                     'set_nas_security_options')
 1639         mock_set_nas_sec_options.return_value = True
 1640         mock_execute = self.mock_object(drv, '_execute')
 1641         mock_execute.return_value = True
 1642 
 1643         drv.do_setup(self.context)
 1644 
 1645         mock_os_path_exists.assert_not_called()
 1646 
 1647     def test_setup_throw_error_if_shares_file_does_not_exist_no_host(self):
 1648         """do_setup should throw error if no shares file and no host set."""
 1649 
 1650         drv = nfs.NfsDriver(configuration=self.configuration)
 1651 
 1652         self.override_config('nas_share_path', '/export')
 1653         mock_os_path_exists = self.mock_object(os.path, 'exists')
 1654         mock_os_path_exists.return_value = False
 1655 
 1656         with self.assertRaisesRegex(exception.NfsException,
 1657                                     "NFS config file.*doesn't exist"):
 1658             drv.do_setup(self.context)
 1659 
 1660         mock_os_path_exists.assert_has_calls(
 1661             [mock.call(self.configuration.nfs_shares_config)])
 1662 
 1663     def test_setup_throw_error_if_shares_file_does_not_exist_no_share(self):
 1664         """do_setup should throw error if no shares file and no share set."""
 1665 
 1666         drv = nfs.NfsDriver(configuration=self.configuration)
 1667 
 1668         self.override_config('nas_host', 'nfs-host1')
 1669         mock_os_path_exists = self.mock_object(os.path, 'exists')
 1670         mock_os_path_exists.return_value = False
 1671 
 1672         with self.assertRaisesRegex(exception.NfsException,
 1673                                     "NFS config file.*doesn't exist"):
 1674             drv.do_setup(self.context)
 1675 
 1676         mock_os_path_exists.assert_has_calls(
 1677             [mock.call(self.configuration.nfs_shares_config)])
 1678 
 1679     def test_setup_throw_error_if_shares_file_doesnt_exist_no_share_host(self):
 1680         """do_setup should throw error if no shares file and no host/share."""
 1681 
 1682         drv = nfs.NfsDriver(configuration=self.configuration)
 1683 
 1684         mock_os_path_exists = self.mock_object(os.path, 'exists')
 1685         mock_os_path_exists.return_value = False
 1686 
 1687         with self.assertRaisesRegex(exception.NfsException,
 1688                                     "NFS config file.*doesn't exist"):
 1689             drv.do_setup(self.context)
 1690 
 1691         mock_os_path_exists.assert_has_calls(
 1692             [mock.call(self.configuration.nfs_shares_config)])
 1693 
 1694     def test_setup_should_throw_exception_if_nfs_client_is_not_installed(self):
 1695         """do_setup should throw error if nfs client is not installed."""
 1696 
 1697         drv = nfs.NfsDriver(configuration=self.configuration)
 1698 
 1699         mock_os_path_exists = self.mock_object(os.path, 'exists')
 1700         mock_os_path_exists.return_value = True
 1701         mock_execute = self.mock_object(drv, '_execute')
 1702         mock_execute.side_effect = OSError(
 1703             errno.ENOENT, 'No such file or directory.')
 1704 
 1705         with self.assertRaisesRegex(exception.NfsException,
 1706                                     'mount.nfs is not installed'):
 1707             drv.do_setup(self.context)
 1708 
 1709         mock_os_path_exists.assert_has_calls(
 1710             [mock.call(self.configuration.nfs_shares_config)])
 1711         mock_execute.assert_has_calls(
 1712             [mock.call('mount.nfs',
 1713                        check_exit_code=False,
 1714                        run_as_root=True)])
 1715 
 1716     def test_setup_should_throw_exception_if_mount_nfs_command_fails(self):
 1717         """do_setup should throw error if mount.nfs fails with OSError
 1718 
 1719            This test covers the OSError path when mount.nfs is installed.
 1720         """
 1721 
 1722         drv = nfs.NfsDriver(configuration=self.configuration)
 1723 
 1724         mock_os_path_exists = self.mock_object(os.path, 'exists')
 1725         mock_os_path_exists.return_value = True
 1726         mock_execute = self.mock_object(drv, '_execute')
 1727         mock_execute.side_effect = OSError(
 1728             errno.EPERM, 'Operation... BROKEN')
 1729 
 1730         with self.assertRaisesRegex(OSError, '.*Operation... BROKEN'):
 1731             drv.do_setup(self.context)
 1732 
 1733         mock_os_path_exists.assert_has_calls(
 1734             [mock.call(self.configuration.nfs_shares_config)])
 1735         mock_execute.assert_has_calls(
 1736             [mock.call('mount.nfs',
 1737                        check_exit_code=False,
 1738                        run_as_root=True)])
 1739 
 1740     def test_retype_is_there(self):
 1741         "Ensure that driver.retype() is there."""
 1742 
 1743         drv = nfs.NfsDriver(configuration=self.configuration)
 1744         v1 = fake_volume.fake_volume_obj(self.context)
 1745 
 1746         ret = drv.retype(self.context,
 1747                          v1,
 1748                          mock.sentinel.new_type,
 1749                          mock.sentinel.diff,
 1750                          mock.sentinel.host)
 1751 
 1752         self.assertEqual((False, None), ret)