"Fossies" - the Fresh Open Source Software Archive

Member "manila-8.1.4/manila/tests/share/drivers/test_lvm.py" (19 Nov 2020, 30389 Bytes) of package /linux/misc/openstack/manila-8.1.4.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_lvm.py": 8.1.3_vs_8.1.4.

    1 # Copyright 2012 NetApp
    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 LVM driver module."""
   16 
   17 import os
   18 
   19 import ddt
   20 import mock
   21 from oslo_concurrency import processutils
   22 from oslo_config import cfg
   23 from oslo_utils import timeutils
   24 
   25 from manila.common import constants as const
   26 from manila import context
   27 from manila import exception
   28 from manila.share import configuration
   29 from manila.share.drivers import lvm
   30 from manila import test
   31 from manila.tests.db import fakes as db_fakes
   32 from manila.tests import fake_utils
   33 from manila.tests.share.drivers import test_generic
   34 
   35 
   36 CONF = cfg.CONF
   37 
   38 
   39 def fake_share(**kwargs):
   40     share = {
   41         'id': 'fakeid',
   42         'name': 'fakename',
   43         'size': 1,
   44         'share_proto': 'NFS',
   45         'export_location': '127.0.0.1:/mnt/nfs/volume-00002',
   46     }
   47     share.update(kwargs)
   48     return db_fakes.FakeModel(share)
   49 
   50 
   51 def fake_snapshot(**kwargs):
   52     snapshot = {
   53         'id': 'fakesnapshotid',
   54         'share_name': 'fakename',
   55         'share_id': 'fakeid',
   56         'name': 'fakesnapshotname',
   57         'share_proto': 'NFS',
   58         'export_location': '127.0.0.1:/mnt/nfs/volume-00002',
   59         'share': {
   60             'id': 'fakeid',
   61             'name': 'fakename',
   62             'size': 1,
   63             'share_proto': 'NFS',
   64         },
   65     }
   66     snapshot.update(kwargs)
   67     return db_fakes.FakeModel(snapshot)
   68 
   69 
   70 def fake_access(**kwargs):
   71     access = {
   72         'id': 'fakeaccid',
   73         'access_type': 'ip',
   74         'access_to': '10.0.0.2',
   75         'access_level': 'rw',
   76         'state': 'active',
   77     }
   78     access.update(kwargs)
   79     return db_fakes.FakeModel(access)
   80 
   81 
   82 @ddt.ddt
   83 class LVMShareDriverTestCase(test.TestCase):
   84     """Tests LVMShareDriver."""
   85 
   86     def setUp(self):
   87         super(LVMShareDriverTestCase, self).setUp()
   88         fake_utils.stub_out_utils_execute(self)
   89         self._context = context.get_admin_context()
   90 
   91         CONF.set_default('lvm_share_volume_group', 'fakevg')
   92         CONF.set_default('lvm_share_export_ips', ['10.0.0.1', '10.0.0.2'])
   93         CONF.set_default('driver_handles_share_servers', False)
   94         CONF.set_default('reserved_share_percentage', 50)
   95 
   96         self._helper_cifs = mock.Mock()
   97         self._helper_nfs = mock.Mock()
   98         self.fake_conf = configuration.Configuration(None)
   99         self._db = mock.Mock()
  100         self._os = lvm.os = mock.Mock()
  101         self._os.path.join = os.path.join
  102         self._driver = lvm.LVMShareDriver(self._db,
  103                                           configuration=self.fake_conf)
  104         self._driver._helpers = {
  105             'CIFS': self._helper_cifs,
  106             'NFS': self._helper_nfs,
  107         }
  108 
  109         self.share = fake_share()
  110         self.access = fake_access()
  111         self.snapshot = fake_snapshot()
  112         self.server = {
  113             'public_addresses': self.fake_conf.lvm_share_export_ips,
  114             'instance_id': 'LVM',
  115             'lock_name': 'manila_lvm',
  116         }
  117 
  118         # Used only to test compatibility with share manager
  119         self.share_server = "fake_share_server"
  120 
  121     def tearDown(self):
  122         super(LVMShareDriverTestCase, self).tearDown()
  123         fake_utils.fake_execute_set_repliers([])
  124         fake_utils.fake_execute_clear_log()
  125 
  126     def test_do_setup(self):
  127         CONF.set_default('lvm_share_helpers', ['NFS=fakenfs'])
  128         lvm.importutils = mock.Mock()
  129         lvm.importutils.import_class.return_value = self._helper_nfs
  130         self._driver.do_setup(self._context)
  131         lvm.importutils.import_class.assert_has_calls([
  132             mock.call('fakenfs')
  133         ])
  134 
  135     def test_check_for_setup_error(self):
  136         def exec_runner(*ignore_args, **ignore_kwargs):
  137             return '\n   fake1\n   fakevg\n   fake2\n', ''
  138 
  139         expected_exec = ['vgs --noheadings -o name']
  140         fake_utils.fake_execute_set_repliers([(expected_exec[0], exec_runner)])
  141         self._driver.check_for_setup_error()
  142         self.assertEqual(expected_exec, fake_utils.fake_execute_get_log())
  143 
  144     def test_check_for_setup_error_no_vg(self):
  145         def exec_runner(*ignore_args, **ignore_kwargs):
  146             return '\n   fake0\n   fake1\n   fake2\n', ''
  147 
  148         fake_utils.fake_execute_set_repliers([('vgs --noheadings -o name',
  149                                                exec_runner)])
  150         self.assertRaises(exception.InvalidParameterValue,
  151                           self._driver.check_for_setup_error)
  152 
  153     def test_check_for_setup_error_deprecated_export_ip(self):
  154         def exec_runner(*ignore_args, **ignore_kwargs):
  155             return '\n   fake1\n   fakevg\n   fake2\n', ''
  156 
  157         fake_utils.fake_execute_set_repliers([('vgs --noheadings -o name',
  158                                                exec_runner)])
  159         CONF.set_default('lvm_share_export_ip', CONF.lvm_share_export_ips[0])
  160         CONF.set_default('lvm_share_export_ips', None)
  161         self.assertIsNone(self._driver.check_for_setup_error())
  162 
  163     def test_check_for_setup_error_no_export_ips(self):
  164         def exec_runner(*ignore_args, **ignore_kwargs):
  165             return '\n   fake1\n   fakevg\n   fake2\n', ''
  166 
  167         fake_utils.fake_execute_set_repliers([('vgs --noheadings -o name',
  168                                                exec_runner)])
  169         CONF.set_default('lvm_share_export_ips', None)
  170         self.assertRaises(exception.InvalidParameterValue,
  171                           self._driver.check_for_setup_error)
  172 
  173     def test_check_for_setup_error_both_export_ip_and_ips(self):
  174         def exec_runner(*ignore_args, **ignore_kwargs):
  175             return '\n   fake1\n   fakevg\n   fake2\n', ''
  176 
  177         fake_utils.fake_execute_set_repliers([('vgs --noheadings -o name',
  178                                                exec_runner)])
  179         CONF.set_default('lvm_share_export_ip', CONF.lvm_share_export_ips[0])
  180         self.assertRaises(exception.InvalidParameterValue,
  181                           self._driver.check_for_setup_error)
  182 
  183     def test_local_path_normal(self):
  184         share = fake_share(name='fake_sharename')
  185         CONF.set_default('lvm_share_volume_group', 'fake_vg')
  186         ret = self._driver._get_local_path(share)
  187         self.assertEqual('/dev/mapper/fake_vg-fake_sharename', ret)
  188 
  189     def test_local_path_escapes(self):
  190         share = fake_share(name='fake-sharename')
  191         CONF.set_default('lvm_share_volume_group', 'fake-vg')
  192         ret = self._driver._get_local_path(share)
  193         self.assertEqual('/dev/mapper/fake--vg-fake--sharename', ret)
  194 
  195     def test_create_share(self):
  196         CONF.set_default('lvm_share_mirrors', 0)
  197         self._driver._mount_device = mock.Mock()
  198 
  199         ret = self._driver.create_share(self._context, self.share,
  200                                         self.share_server)
  201 
  202         self._driver._mount_device.assert_called_with(
  203             self.share, '/dev/mapper/fakevg-fakename')
  204         expected_exec = [
  205             'lvcreate -L 1G -n fakename fakevg',
  206             'mkfs.ext4 /dev/mapper/fakevg-fakename',
  207         ]
  208         self.assertEqual(expected_exec, fake_utils.fake_execute_get_log())
  209         self.assertEqual(self._helper_nfs.create_exports.return_value, ret)
  210 
  211     def test_create_share_from_snapshot(self):
  212         CONF.set_default('lvm_share_mirrors', 0)
  213         self._driver._mount_device = mock.Mock()
  214         snapshot_instance = {
  215             'snapshot_id': 'fakesnapshotid',
  216             'name': 'fakename'
  217         }
  218         mount_share = '/dev/mapper/fakevg-fakename'
  219         mount_snapshot = '/dev/mapper/fakevg-fakename'
  220         self._helper_nfs.create_export.return_value = 'fakelocation'
  221         self._driver.create_share_from_snapshot(self._context,
  222                                                 self.share,
  223                                                 snapshot_instance,
  224                                                 self.share_server)
  225 
  226         self._driver._mount_device.assert_called_with(self.share,
  227                                                       mount_snapshot)
  228         expected_exec = [
  229             'lvcreate -L 1G -n fakename fakevg',
  230             'mkfs.ext4 /dev/mapper/fakevg-fakename',
  231             'e2fsck -y -f %s' % mount_share,
  232             'tune2fs -U random %s' % mount_share,
  233             ("dd count=0 if=%s of=%s iflag=direct oflag=direct" %
  234              (mount_snapshot, mount_share)),
  235             ("dd if=%s of=%s count=1024 bs=1M iflag=direct oflag=direct" %
  236              (mount_snapshot, mount_share)),
  237         ]
  238         self.assertEqual(expected_exec, fake_utils.fake_execute_get_log())
  239 
  240     def test_create_share_mirrors(self):
  241         share = fake_share(size='2048')
  242         CONF.set_default('lvm_share_mirrors', 2)
  243         self._driver._mount_device = mock.Mock()
  244 
  245         ret = self._driver.create_share(self._context, share,
  246                                         self.share_server)
  247 
  248         self._driver._mount_device.assert_called_with(
  249             share, '/dev/mapper/fakevg-fakename')
  250         expected_exec = [
  251             'lvcreate -L 2048G -n fakename fakevg -m 2 --nosync -R 2',
  252             'mkfs.ext4 /dev/mapper/fakevg-fakename',
  253         ]
  254         self.assertEqual(expected_exec, fake_utils.fake_execute_get_log())
  255         self.assertEqual(self._helper_nfs.create_exports.return_value, ret)
  256 
  257     def test_deallocate_container(self):
  258         expected_exec = ['lvremove -f fakevg/fakename']
  259         self._driver._deallocate_container(self.share['name'])
  260         self.assertEqual(expected_exec, fake_utils.fake_execute_get_log())
  261 
  262     def test_deallocate_container_error(self):
  263         def _fake_exec(*args, **kwargs):
  264             raise exception.ProcessExecutionError(stderr="error")
  265 
  266         self.mock_object(self._driver, '_try_execute', _fake_exec)
  267         self.assertRaises(exception.ProcessExecutionError,
  268                           self._driver._deallocate_container,
  269                           self.share['name'])
  270 
  271     @ddt.data(
  272         'Logical volume "fake/fake-volume" not found\n',
  273         'Failed to find logical volume "fake/fake-volume"\n')
  274     def test_deallocate_container_not_found_error(self, error_msg):
  275         def _fake_exec(*args, **kwargs):
  276             raise exception.ProcessExecutionError(stderr=error_msg)
  277 
  278         self.mock_object(self._driver, '_try_execute', _fake_exec)
  279         self._driver._deallocate_container(self.share['name'])
  280 
  281     @mock.patch.object(lvm.LVMShareDriver, '_update_share_stats', mock.Mock())
  282     def test_get_share_stats(self):
  283         with mock.patch.object(self._driver, '_stats', mock.Mock) as stats:
  284             self.assertEqual(stats, self._driver.get_share_stats())
  285         self.assertFalse(self._driver._update_share_stats.called)
  286 
  287     @mock.patch.object(lvm.LVMShareDriver, '_update_share_stats', mock.Mock())
  288     def test_get_share_stats_refresh(self):
  289         with mock.patch.object(self._driver, '_stats', mock.Mock) as stats:
  290             self.assertEqual(stats,
  291                              self._driver.get_share_stats(refresh=True))
  292         self._driver._update_share_stats.assert_called_once_with()
  293 
  294     def test__unmount_device_not_mounted(self):
  295         def exec_runner(*ignore_args, **ignore_kwargs):
  296             umount_msg = (
  297                 "umount: /opt/stack/data/manila/mnt/share-fake-share: not "
  298                 "mounted.\n"
  299             )
  300             raise exception.ProcessExecutionError(stderr=umount_msg)
  301         self._os.path.exists.return_value = True
  302         mount_path = self._get_mount_path(self.share)
  303         expected_exec = "umount -f %s" % (mount_path)
  304         fake_utils.fake_execute_set_repliers([(expected_exec, exec_runner)])
  305 
  306         self._driver._unmount_device(self.share, raise_if_missing=False)
  307 
  308         self._os.path.exists.assert_called_with(mount_path)
  309 
  310     def test__unmount_device_is_busy_error(self):
  311         def exec_runner(*ignore_args, **ignore_kwargs):
  312             raise exception.ProcessExecutionError(stderr='device is busy')
  313         self._os.path.exists.return_value = True
  314         mount_path = self._get_mount_path(self.share)
  315         expected_exec = [
  316             "umount -f %s" % (mount_path),
  317         ]
  318         fake_utils.fake_execute_set_repliers([(expected_exec[0], exec_runner)])
  319 
  320         self.assertRaises(exception.ShareBusyException,
  321                           self._driver._unmount_device,
  322                           self.share)
  323         self.assertEqual(expected_exec, fake_utils.fake_execute_get_log())
  324 
  325     def test__unmount_device_error(self):
  326         def exec_runner(*ignore_args, **ignore_kwargs):
  327             raise exception.ProcessExecutionError(stderr='fake error')
  328         mount_path = self._get_mount_path(self.share)
  329         self._os.path.exists.return_value = True
  330         cmd = "umount -f %s" % (mount_path)
  331         fake_utils.fake_execute_set_repliers([(cmd, exec_runner)])
  332         self.assertRaises(processutils.ProcessExecutionError,
  333                           self._driver._unmount_device,
  334                           self.share)
  335         self._os.path.exists.assert_called_with(mount_path)
  336 
  337     def test__unmount_device_rmdir_error(self):
  338         def exec_runner(*ignore_args, **ignore_kwargs):
  339             raise exception.ProcessExecutionError(stderr='fake error')
  340         mount_path = self._get_mount_path(self.share)
  341         self._os.path.exists.return_value = True
  342         cmd = "rmdir %s" % (mount_path)
  343         fake_utils.fake_execute_set_repliers([(cmd, exec_runner)])
  344         self.assertRaises(processutils.ProcessExecutionError,
  345                           self._driver._unmount_device,
  346                           self.share)
  347         self._os.path.exists.assert_called_with(mount_path)
  348 
  349     def test_create_snapshot(self):
  350         self._driver.create_snapshot(self._context, self.snapshot,
  351                                      self.share_server)
  352         mount_path = self._get_mount_path(self.snapshot)
  353         expected_exec = [
  354             ("lvcreate -L 1G --name fakesnapshotname --snapshot "
  355              "%s/fakename" % (CONF.lvm_share_volume_group,)),
  356             "e2fsck -y -f /dev/mapper/fakevg-%s" % self.snapshot['name'],
  357             "tune2fs -U random /dev/mapper/fakevg-%s" % self.snapshot['name'],
  358             "mkdir -p " + mount_path,
  359             "mount /dev/mapper/fakevg-fakesnapshotname " + mount_path,
  360             "chmod 777 " + mount_path,
  361         ]
  362         self.assertEqual(expected_exec, fake_utils.fake_execute_get_log())
  363 
  364     def test_ensure_share(self):
  365         device_name = '/dev/mapper/fakevg-fakename'
  366         with mock.patch.object(self._driver,
  367                                '_mount_device',
  368                                mock.Mock(return_value='fake_location')):
  369             self._driver.ensure_share(self._context, self.share,
  370                                       self.share_server)
  371             self._driver._mount_device.assert_called_with(self.share,
  372                                                           device_name)
  373             self._helper_nfs.create_exports.assert_called_once_with(
  374                 self.server, self.share['name'], recreate=True)
  375 
  376     def test_delete_share(self):
  377         mount_path = self._get_mount_path(self.share)
  378         self._helper_nfs.remove_export(mount_path, self.share['name'])
  379         self._driver._delete_share(self._context, self.share)
  380 
  381     def test_delete_snapshot(self):
  382         mount_path = self._get_mount_path(self.snapshot)
  383         expected_exec = [
  384             'umount -f %s' % mount_path,
  385             'rmdir %s' % mount_path,
  386             'lvremove -f fakevg/fakesnapshotname',
  387         ]
  388         self._driver.delete_snapshot(self._context, self.snapshot,
  389                                      self.share_server)
  390         self.assertEqual(expected_exec, fake_utils.fake_execute_get_log())
  391 
  392     def test_delete_share_invalid_share(self):
  393         self._driver._get_helper = mock.Mock(
  394             side_effect=exception.InvalidShare(reason='fake'))
  395         self._driver.delete_share(self._context, self.share, self.share_server)
  396 
  397     def test_delete_share_process_execution_error(self):
  398         self.mock_object(
  399             self._helper_nfs,
  400             'remove_export',
  401             mock.Mock(side_effect=exception.ProcessExecutionError))
  402 
  403         self._driver._delete_share(self._context, self.share)
  404         self._helper_nfs.remove_exports.assert_called_once_with(
  405             self.server,
  406             self.share['name'])
  407 
  408     @ddt.data(const.ACCESS_LEVEL_RW, const.ACCESS_LEVEL_RO)
  409     def test_update_access(self, access_level):
  410         access_rules = [test_generic.get_fake_access_rule(
  411             '1.1.1.1', access_level), ]
  412         add_rules = [test_generic.get_fake_access_rule(
  413             '2.2.2.2', access_level), ]
  414         delete_rules = [test_generic.get_fake_access_rule(
  415             '3.3.3.3', access_level), ]
  416         self._driver.update_access(self._context, self.share, access_rules,
  417                                    add_rules=add_rules,
  418                                    delete_rules=delete_rules,
  419                                    share_server=self.server)
  420         (self._driver._helpers[self.share['share_proto']].
  421             update_access.assert_called_once_with(
  422                 self.server, self.share['name'],
  423                 access_rules, add_rules=add_rules, delete_rules=delete_rules))
  424 
  425     @ddt.data(('1001::1001/129', None, False), ('1.1.1.256', None, False),
  426               ('1001::1001', None, [6]), ('1.1.1.0', None, [4]),
  427               (None, ['1001::1001', '1.1.1.0'], [6, 4]),
  428               (None, ['1001::1001'], [6]), (None, ['1.1.1.0'], [4]),
  429               (None, ['1001::1001/129', '1.1.1.0'], False))
  430     @ddt.unpack
  431     def test_get_configured_ip_versions(
  432             self, configured_ip, configured_ips, configured_ip_version):
  433         CONF.set_default('lvm_share_export_ip', configured_ip)
  434         CONF.set_default('lvm_share_export_ips', configured_ips)
  435         if configured_ip_version:
  436             self.assertEqual(configured_ip_version,
  437                              self._driver.get_configured_ip_versions())
  438         else:
  439             self.assertRaises(exception.InvalidInput,
  440                               self._driver.get_configured_ip_versions)
  441 
  442     def test_mount_device(self):
  443         mount_path = self._get_mount_path(self.share)
  444         ret = self._driver._mount_device(self.share, 'fakedevice')
  445         expected_exec = [
  446             "mkdir -p %s" % (mount_path,),
  447             "mount fakedevice %s" % (mount_path,),
  448             "chmod 777 %s" % (mount_path,),
  449         ]
  450         self.assertEqual(expected_exec, fake_utils.fake_execute_get_log())
  451         self.assertEqual(mount_path, ret)
  452 
  453     def test_mount_device_already(self):
  454         def exec_runner(*args, **kwargs):
  455             if 'mount' in args and '-l' not in args:
  456                 raise exception.ProcessExecutionError()
  457             else:
  458                 return 'fakedevice', ''
  459 
  460         self.mock_object(self._driver, '_execute', exec_runner)
  461         mount_path = self._get_mount_path(self.share)
  462 
  463         ret = self._driver._mount_device(self.share, 'fakedevice')
  464         self.assertEqual(mount_path, ret)
  465 
  466     def test_mount_device_error(self):
  467         def exec_runner(*args, **kwargs):
  468             if 'mount' in args and '-l' not in args:
  469                 raise exception.ProcessExecutionError()
  470             else:
  471                 return 'fake', ''
  472 
  473         self.mock_object(self._driver, '_execute', exec_runner)
  474         self.assertRaises(exception.ProcessExecutionError,
  475                           self._driver._mount_device, self.share, 'fakedevice')
  476 
  477     def test_get_helper(self):
  478         share_cifs = fake_share(share_proto='CIFS')
  479         share_nfs = fake_share(share_proto='NFS')
  480         share_fake = fake_share(share_proto='FAKE')
  481         self.assertEqual(self._driver._get_helper(share_cifs),
  482                          self._helper_cifs)
  483         self.assertEqual(self._driver._get_helper(share_nfs),
  484                          self._helper_nfs)
  485         self.assertRaises(exception.InvalidShare, self._driver._get_helper,
  486                           share_fake)
  487 
  488     def _get_mount_path(self, share):
  489         return os.path.join(CONF.lvm_share_export_root, share['name'])
  490 
  491     def test__unmount_device(self):
  492         mount_path = self._get_mount_path(self.share)
  493         self._os.path.exists.return_value = True
  494         self.mock_object(self._driver, '_execute')
  495         self._driver._unmount_device(self.share)
  496         self._driver._execute.assert_any_call('umount', '-f', mount_path,
  497                                               run_as_root=True)
  498         self._driver._execute.assert_any_call('rmdir', mount_path,
  499                                               run_as_root=True)
  500         self._os.path.exists.assert_called_with(mount_path)
  501 
  502     def test_extend_share(self):
  503         local_path = self._driver._get_local_path(self.share)
  504         self.mock_object(self._driver, '_extend_container')
  505         self.mock_object(self._driver, '_execute')
  506         self._driver.extend_share(self.share, 3)
  507         self._driver._extend_container.assert_called_once_with(self.share,
  508                                                                local_path, 3)
  509 
  510     def test_ssh_exec_as_root(self):
  511         command = ['fake_command']
  512         self.mock_object(self._driver, '_execute')
  513         self._driver._ssh_exec_as_root('fake_server', command)
  514         self._driver._execute.assert_called_once_with('fake_command',
  515                                                       check_exit_code=True)
  516 
  517     def test_ssh_exec_as_root_with_sudo(self):
  518         command = ['sudo', 'fake_command']
  519         self.mock_object(self._driver, '_execute')
  520         self._driver._ssh_exec_as_root('fake_server', command)
  521         self._driver._execute.assert_called_once_with(
  522             'fake_command', run_as_root=True, check_exit_code=True)
  523 
  524     def test_extend_container(self):
  525         self.mock_object(self._driver, '_try_execute')
  526         self._driver._extend_container(self.share, 'device_name', 3)
  527         self._driver._try_execute.assert_called_once_with(
  528             'lvextend',
  529             '-L',
  530             '3G',
  531             '-r',
  532             'device_name',
  533             run_as_root=True)
  534 
  535     def test_get_share_server_pools(self):
  536         expected_result = [{
  537             'pool_name': 'lvm-single-pool',
  538             'total_capacity_gb': 33,
  539             'free_capacity_gb': 22,
  540             'reserved_percentage': 0,
  541         }, ]
  542         self.mock_object(
  543             self._driver,
  544             '_execute',
  545             mock.Mock(return_value=("VSize 33g VFree 22g", None)))
  546 
  547         self.assertEqual(expected_result,
  548                          self._driver.get_share_server_pools())
  549         self._driver._execute.assert_called_once_with(
  550             'vgs', 'fakevg', '--rows', '--units', 'g', run_as_root=True)
  551 
  552     def test_copy_volume_error(self):
  553         def _fake_exec(*args, **kwargs):
  554             if 'count=0' in args:
  555                 raise exception.ProcessExecutionError()
  556 
  557         self.mock_object(self._driver, '_execute',
  558                          mock.Mock(side_effect=_fake_exec))
  559         self._driver._copy_volume('src', 'dest', 1)
  560         self._driver._execute.assert_any_call('dd', 'count=0', 'if=src',
  561                                               'of=dest', 'iflag=direct',
  562                                               'oflag=direct', run_as_root=True)
  563         self._driver._execute.assert_any_call('dd', 'if=src', 'of=dest',
  564                                               'count=1024', 'bs=1M',
  565                                               run_as_root=True)
  566 
  567     @ddt.data(('1.1.1.1', 4), ('1001::1001', 6))
  568     @ddt.unpack
  569     def test_update_share_stats(self, configured_ip, version):
  570         CONF.set_default('lvm_share_export_ip', configured_ip)
  571         self.mock_object(self._driver, 'get_share_server_pools',
  572                          mock.Mock(return_value='test-pool'))
  573 
  574         self._driver._update_share_stats()
  575         self.assertEqual('LVM', self._driver._stats['share_backend_name'])
  576         self.assertEqual('NFS_CIFS', self._driver._stats['storage_protocol'])
  577         self.assertEqual(50, self._driver._stats['reserved_percentage'])
  578         self.assertTrue(self._driver._stats['snapshot_support'])
  579         self.assertEqual('LVMShareDriver', self._driver._stats['driver_name'])
  580         self.assertEqual('test-pool', self._driver._stats['pools'])
  581         self.assertEqual(version == 4, self._driver._stats['ipv4_support'])
  582         self.assertEqual(version == 6, self._driver._stats['ipv6_support'])
  583 
  584     def test_revert_to_snapshot(self):
  585         mock_update_access = self.mock_object(self._helper_nfs,
  586                                               'update_access')
  587         self._driver.revert_to_snapshot(self._context, self.snapshot,
  588                                         [], [], self.share_server)
  589         snap_lv = "%s/fakesnapshotname" % (CONF.lvm_share_volume_group)
  590         share_lv = "%s/fakename" % (CONF.lvm_share_volume_group)
  591         share_mount_path = self._get_mount_path(self.snapshot['share'])
  592         snapshot_mount_path = self._get_mount_path(self.snapshot)
  593         expected_exec = [
  594             ('umount -f %s' % snapshot_mount_path),
  595             ("rmdir %s" % snapshot_mount_path),
  596             ("umount -f %s" % share_mount_path),
  597             ("rmdir %s" % share_mount_path),
  598             ("lvconvert --merge %s" % snap_lv),
  599             ("lvcreate -L 1G --name fakesnapshotname --snapshot %s" %
  600                 share_lv),
  601             ("e2fsck -y -f /dev/mapper/%s-fakesnapshotname" %
  602                 CONF.lvm_share_volume_group),
  603             ("tune2fs -U random /dev/mapper/%s-fakesnapshotname" %
  604                 CONF.lvm_share_volume_group),
  605             ("mkdir -p %s" % share_mount_path),
  606             ("mount /dev/mapper/%s-fakename %s" %
  607                 (CONF.lvm_share_volume_group, share_mount_path)),
  608             ("chmod 777 %s" % share_mount_path),
  609             ("mkdir -p %s" % snapshot_mount_path),
  610             ("mount /dev/mapper/fakevg-fakesnapshotname "
  611              "%s" % snapshot_mount_path),
  612             ("chmod 777 %s" % snapshot_mount_path),
  613         ]
  614         self.assertEqual(expected_exec, fake_utils.fake_execute_get_log())
  615         self.assertEqual(4, mock_update_access.call_count)
  616 
  617     def test_snapshot_update_access(self):
  618         access_rules = [{
  619             'access_type': 'ip',
  620             'access_to': '1.1.1.1',
  621             'access_level': 'ro',
  622         }]
  623 
  624         add_rules = [{
  625             'access_type': 'ip',
  626             'access_to': '2.2.2.2',
  627             'access_level': 'ro',
  628         }]
  629 
  630         delete_rules = [{
  631             'access_type': 'ip',
  632             'access_to': '3.3.3.3',
  633             'access_level': 'ro',
  634         }]
  635 
  636         self._driver.snapshot_update_access(self._context, self.snapshot,
  637                                             access_rules, add_rules,
  638                                             delete_rules)
  639 
  640         (self._driver._helpers[self.snapshot['share']['share_proto']].
  641             update_access.assert_called_once_with(
  642             self.server, self.snapshot['name'],
  643             access_rules, add_rules=add_rules, delete_rules=delete_rules))
  644 
  645     @mock.patch.object(timeutils, 'utcnow', mock.Mock(
  646                        return_value='fake_date'))
  647     def test_update_share_usage_size(self):
  648         mount_path = self._get_mount_path(self.share)
  649         self._os.path.exists.return_value = True
  650         self.mock_object(
  651             self._driver,
  652             '_execute',
  653             mock.Mock(return_value=(
  654                 "Mounted on                    Used "
  655                 + mount_path + "               1G", None)))
  656 
  657         update_shares = self._driver.update_share_usage_size(
  658             self._context, [self.share, ])
  659         self._os.path.exists.assert_called_with(mount_path)
  660         self.assertEqual(
  661             [{'id': 'fakeid', 'used_size': '1',
  662               'gathered_at': 'fake_date'}],
  663             update_shares)
  664         self._driver._execute.assert_called_once_with(
  665             'df', '-l', '--output=target,used',
  666             '--block-size=g')
  667 
  668     @mock.patch.object(timeutils, 'utcnow', mock.Mock(
  669                        return_value='fake_date'))
  670     def test_update_share_usage_size_multiple_share(self):
  671         share1 = fake_share(id='fakeid_get_fail', name='get_fail')
  672         share2 = fake_share(id='fakeid_success', name='get_success')
  673         share3 = fake_share(id='fakeid_not_exist', name='get_not_exist')
  674 
  675         mount_path2 = self._get_mount_path(share2)
  676         mount_path3 = self._get_mount_path(share3)
  677         self._os.path.exists.side_effect = [True, True, False]
  678         self.mock_object(
  679             self._driver,
  680             '_execute',
  681             mock.Mock(return_value=(
  682                 "Mounted on                    Used "
  683                 + mount_path2 + "               1G", None)))
  684 
  685         update_shares = self._driver.update_share_usage_size(
  686             self._context, [share1, share2, share3])
  687         self._os.path.exists.assert_called_with(mount_path3)
  688         self.assertEqual(
  689             [{'gathered_at': 'fake_date',
  690               'id': 'fakeid_success', 'used_size': '1'}],
  691             update_shares)
  692         self._driver._execute.assert_called_with(
  693             'df', '-l', '--output=target,used',
  694             '--block-size=g')
  695 
  696     def test_update_share_usage_size_fail(self):
  697         def _fake_exec(*args, **kwargs):
  698             raise exception.ProcessExecutionError(stderr="error")
  699 
  700         self.mock_object(self._driver, '_execute', _fake_exec)
  701         self.assertRaises(exception.ProcessExecutionError,
  702                           self._driver.update_share_usage_size,
  703                           self._context,
  704                           [self.share])
  705 
  706     def test_get_backend_info(self):
  707         backend_info = self._driver.get_backend_info(self._context)
  708 
  709         self.assertEqual(
  710             {'export_ips': ','.join(self.server['public_addresses']),
  711              'db_version': mock.ANY},
  712             backend_info)