stub = mox_lib.MockObject(attr_to_replace)
self.stubs.Set(obj, attr_name, stub)
- def test_path_exists_should_return_true(self):
- """_path_exists should return True if stat returns 0."""
- mox = self._mox
- drv = self._driver
-
- mox.StubOutWithMock(drv, '_execute')
- drv._execute('stat', self.TEST_FILE_NAME, run_as_root=True)
-
- mox.ReplayAll()
-
- self.assertTrue(drv._path_exists(self.TEST_FILE_NAME))
-
- mox.VerifyAll()
-
- def test_path_exists_should_return_false(self):
- """_path_exists should return True if stat doesn't return 0."""
- mox = self._mox
- drv = self._driver
-
- mox.StubOutWithMock(drv, '_execute')
- drv._execute(
- 'stat',
- self.TEST_FILE_NAME, run_as_root=True).\
- AndRaise(ProcessExecutionError(
- stderr="stat: cannot stat `test.txt': No such file "
- "or directory"))
-
- mox.ReplayAll()
-
- self.assertFalse(drv._path_exists(self.TEST_FILE_NAME))
-
- mox.VerifyAll()
-
def test_local_path(self):
"""local_path common use case."""
glusterfs.FLAGS.glusterfs_mount_point_base = self.TEST_MNT_POINT_BASE
mox = self._mox
drv = self._driver
- mox.StubOutWithMock(drv, '_path_exists')
- drv._path_exists(self.TEST_MNT_POINT).AndReturn(True)
-
mox.StubOutWithMock(drv, '_execute')
+ drv._execute('mkdir', '-p', self.TEST_MNT_POINT)
drv._execute('mount', '-t', 'glusterfs', self.TEST_EXPORT1,
self.TEST_MNT_POINT, run_as_root=True)
mox = self._mox
drv = self._driver
- mox.StubOutWithMock(drv, '_path_exists')
- drv._path_exists(self.TEST_MNT_POINT).AndReturn(True)
-
mox.StubOutWithMock(drv, '_execute')
+ drv._execute('mkdir', '-p', self.TEST_MNT_POINT)
drv._execute('mount', '-t', 'glusterfs', self.TEST_EXPORT1,
self.TEST_MNT_POINT, run_as_root=True).\
AndRaise(ProcessExecutionError(
mox = self._mox
drv = self._driver
- mox.StubOutWithMock(drv, '_path_exists')
- drv._path_exists(self.TEST_MNT_POINT).AndReturn(True)
-
mox.StubOutWithMock(drv, '_execute')
+ drv._execute('mkdir', '-p', self.TEST_MNT_POINT)
drv._execute(
'mount',
'-t',
mox = self._mox
drv = self._driver
- mox.StubOutWithMock(drv, '_path_exists')
- drv._path_exists(self.TEST_MNT_POINT).AndReturn(False)
-
mox.StubOutWithMock(drv, '_execute')
drv._execute('mkdir', '-p', self.TEST_MNT_POINT)
drv._execute(*([IgnoreArg()] * 5), run_as_root=IgnoreArg())
mox.VerifyAll()
- def test_mount_glusterfs_should_not_create_mountpoint_if_already(self):
- """_mount_glusterfs should not create mountpoint if it already exists.
- """
- mox = self._mox
- drv = self._driver
-
- mox.StubOutWithMock(drv, '_path_exists')
- drv._path_exists(self.TEST_MNT_POINT).AndReturn(True)
-
- mox.StubOutWithMock(drv, '_execute')
- drv._execute(*([IgnoreArg()] * 5), run_as_root=IgnoreArg())
-
- mox.ReplayAll()
-
- drv._mount_glusterfs(self.TEST_EXPORT1, self.TEST_MNT_POINT)
-
- mox.VerifyAll()
-
def test_get_hash_str(self):
"""_get_hash_str should calculation correct value."""
drv = self._driver
mox.StubOutWithMock(drv, 'local_path')
drv.local_path(volume).AndReturn(self.TEST_LOCAL_PATH)
- mox.StubOutWithMock(drv, '_path_exists')
- drv._path_exists(self.TEST_LOCAL_PATH).AndReturn(True)
-
mox.StubOutWithMock(drv, '_execute')
drv._execute('rm', '-f', self.TEST_LOCAL_PATH, run_as_root=True)
drv.delete_volume(volume)
mox.VerifyAll()
-
- def test_delete_should_not_delete_if_there_is_no_file(self):
- """delete_volume should not try to delete if file missed."""
- mox = self._mox
- drv = self._driver
-
- self.stub_out_not_replaying(drv, '_ensure_share_mounted')
-
- volume = DumbVolume()
- volume['name'] = 'volume-123'
- volume['provider_location'] = self.TEST_EXPORT1
-
- mox.StubOutWithMock(drv, 'local_path')
- drv.local_path(volume).AndReturn(self.TEST_LOCAL_PATH)
-
- mox.StubOutWithMock(drv, '_path_exists')
- drv._path_exists(self.TEST_LOCAL_PATH).AndReturn(False)
-
- mox.StubOutWithMock(drv, '_execute')
-
- mox.ReplayAll()
-
- drv.delete_volume(volume)
-
- mox.VerifyAll()
mox.VerifyAll()
- def test_path_exists_should_return_true(self):
- """_path_exists should return True if stat returns 0."""
- mox = self._mox
- drv = self._driver
-
- mox.StubOutWithMock(drv, '_execute')
- drv._execute('stat', self.TEST_FILE_NAME, run_as_root=True)
-
- mox.ReplayAll()
-
- self.assertTrue(drv._path_exists(self.TEST_FILE_NAME))
-
- mox.VerifyAll()
-
- def test_path_exists_should_return_false(self):
- """_path_exists should return True if stat doesn't return 0."""
- mox = self._mox
- drv = self._driver
-
- mox.StubOutWithMock(drv, '_execute')
- drv._execute(
- 'stat',
- self.TEST_FILE_NAME, run_as_root=True).\
- AndRaise(ProcessExecutionError(
- stderr="stat: cannot stat `test.txt': No such file "
- "or directory"))
-
- mox.ReplayAll()
-
- self.assertFalse(drv._path_exists(self.TEST_FILE_NAME))
-
- mox.VerifyAll()
-
def test_get_hash_str(self):
"""_get_hash_str should calculation correct value."""
drv = self._driver
stub = mox_lib.MockObject(attr_to_replace)
self.stubs.Set(obj, attr_name, stub)
- def test_path_exists_should_return_true(self):
- """_path_exists should return True if stat returns 0."""
- mox = self._mox
- drv = self._driver
-
- mox.StubOutWithMock(drv, '_execute')
- drv._execute('stat', self.TEST_FILE_NAME, run_as_root=True)
-
- mox.ReplayAll()
-
- self.assertTrue(drv._path_exists(self.TEST_FILE_NAME))
-
- mox.VerifyAll()
-
- def test_path_exists_should_return_false(self):
- """_path_exists should return True if stat doesn't return 0."""
- mox = self._mox
- drv = self._driver
-
- mox.StubOutWithMock(drv, '_execute')
- drv._execute(
- 'stat',
- self.TEST_FILE_NAME, run_as_root=True).\
- AndRaise(ProcessExecutionError(
- stderr="stat: cannot stat `test.txt': No such file "
- "or directory"))
-
- mox.ReplayAll()
-
- self.assertFalse(drv._path_exists(self.TEST_FILE_NAME))
-
- mox.VerifyAll()
-
def test_local_path(self):
"""local_path common use case."""
self.configuration.nfs_mount_point_base = self.TEST_MNT_POINT_BASE
mox = self._mox
drv = self._driver
- mox.StubOutWithMock(drv, '_path_exists')
- drv._path_exists(self.TEST_MNT_POINT).AndReturn(True)
-
mox.StubOutWithMock(drv, '_execute')
+ drv._execute('mkdir', '-p', self.TEST_MNT_POINT)
drv._execute('mount', '-t', 'nfs', self.TEST_NFS_EXPORT1,
self.TEST_MNT_POINT, run_as_root=True)
mox = self._mox
drv = self._driver
- mox.StubOutWithMock(drv, '_path_exists')
- drv._path_exists(self.TEST_MNT_POINT).AndReturn(True)
-
mox.StubOutWithMock(drv, '_execute')
+ drv._execute('mkdir', '-p', self.TEST_MNT_POINT)
drv._execute('mount', '-t', 'nfs', self.TEST_NFS_EXPORT1,
self.TEST_MNT_POINT, run_as_root=True).\
AndRaise(ProcessExecutionError(
mox = self._mox
drv = self._driver
- mox.StubOutWithMock(drv, '_path_exists')
- drv._path_exists(self.TEST_MNT_POINT).AndReturn(True)
-
mox.StubOutWithMock(drv, '_execute')
+ drv._execute('mkdir', '-p', self.TEST_MNT_POINT)
drv._execute(
'mount',
'-t',
mox = self._mox
drv = self._driver
- mox.StubOutWithMock(drv, '_path_exists')
- drv._path_exists(self.TEST_MNT_POINT).AndReturn(False)
-
mox.StubOutWithMock(drv, '_execute')
drv._execute('mkdir', '-p', self.TEST_MNT_POINT)
drv._execute(*([IgnoreArg()] * 5), run_as_root=IgnoreArg())
mox = self._mox
drv = self._driver
- mox.StubOutWithMock(drv, '_path_exists')
- drv._path_exists(self.TEST_MNT_POINT).AndReturn(True)
-
mox.StubOutWithMock(drv, '_execute')
+ drv._execute('mkdir', '-p', self.TEST_MNT_POINT)
drv._execute(*([IgnoreArg()] * 5), run_as_root=IgnoreArg())
mox.ReplayAll()
mox.StubOutWithMock(drv, 'local_path')
drv.local_path(volume).AndReturn(self.TEST_LOCAL_PATH)
- mox.StubOutWithMock(drv, '_path_exists')
- drv._path_exists(self.TEST_LOCAL_PATH).AndReturn(True)
-
mox.StubOutWithMock(drv, '_execute')
drv._execute('rm', '-f', self.TEST_LOCAL_PATH, run_as_root=True)
mox.VerifyAll()
- def test_delete_should_not_delete_if_there_is_no_file(self):
- """delete_volume should not try to delete if file missed."""
- mox = self._mox
- drv = self._driver
-
- self.stub_out_not_replaying(drv, '_ensure_share_mounted')
-
- volume = DumbVolume()
- volume['name'] = 'volume-123'
- volume['provider_location'] = self.TEST_NFS_EXPORT1
-
- mox.StubOutWithMock(drv, 'local_path')
- drv.local_path(volume).AndReturn(self.TEST_LOCAL_PATH)
-
- mox.StubOutWithMock(drv, '_path_exists')
- drv._path_exists(self.TEST_LOCAL_PATH).AndReturn(False)
-
- mox.StubOutWithMock(drv, '_execute')
-
- mox.ReplayAll()
-
- drv.delete_volume(volume)
-
- mox.VerifyAll()
-
def test_get_volume_stats(self):
"""get_volume_stats must fill the correct values"""
mox = self._mox
mounted_path = self.local_path(volume)
- if not self._path_exists(mounted_path):
- volume = volume['name']
-
- LOG.warn(_('Trying to delete non-existing volume %(volume)s at '
- 'path %(mounted_path)s') % locals())
- return
-
self._execute('rm', '-f', mounted_path, run_as_root=True)
def ensure_export(self, ctx, volume):
def _mount_glusterfs(self, glusterfs_share, mount_path, ensure=False):
"""Mount GlusterFS share to mount path."""
- if not self._path_exists(mount_path):
- self._execute('mkdir', '-p', mount_path)
+ self._execute('mkdir', '-p', mount_path)
try:
self._execute('mount', '-t', 'glusterfs', glusterfs_share,
return os.path.join(self._get_mount_point_for_share(nfs_share),
volume['name'])
- def _path_exists(self, path):
- """Check for existence of given path."""
- try:
- self._execute('stat', path, run_as_root=True)
- return True
- except exception.ProcessExecutionError as exc:
- if 'No such file or directory' in exc.stderr:
- return False
- else:
- raise
-
def _get_hash_str(self, base_str):
"""returns string that represents hash of base_str
(in a hex format)."""
mounted_path = self.local_path(volume)
- if not self._path_exists(mounted_path):
- volume = volume['name']
-
- LOG.warn(_('Trying to delete non-existing volume %(volume)s at '
- 'path %(mounted_path)s') % locals())
- return
-
self._execute('rm', '-f', mounted_path, run_as_root=True)
def ensure_export(self, ctx, volume):
def _mount_nfs(self, nfs_share, mount_path, ensure=False):
"""Mount NFS share to mount path"""
- if not self._path_exists(mount_path):
- self._execute('mkdir', '-p', mount_path)
+ self._execute('mkdir', '-p', mount_path)
# Construct the NFS mount command.
nfs_cmd = ['mount', '-t', 'nfs']