fileutils is graduated in the oslo.utils library.
Use the open function from builtins instead of fileutils.file_open.
Change the test_backup_volume (test_scality module) and
test_restore_backup (test_scality, test_volume) functions to use
mock library instead of mox.
Implements: blueprint graduate-fileutils[1]
[1] https://blueprints.launchpad.net/oslo-incubator/+spec/graduate-fileutils
Change-Id: I7abf0273bf2b90c72f51ebf4896808eaabb91969
from oslo_concurrency import processutils
from oslo_config import cfg
from oslo_log import log as logging
+from oslo_utils import fileutils
from oslo_utils import timeutils
from oslo_utils import units
from cinder import exception
from cinder.i18n import _, _LI, _LW
-from cinder.openstack.common import fileutils
from cinder.openstack.common import imageutils
from cinder import utils
from cinder.volume import throttling
LOG.debug("%s was %s, no need to convert to %s",
image_id, volume_format, image_meta['disk_format'])
if os.name == 'nt' or os.access(volume_path, os.R_OK):
- with fileutils.file_open(volume_path, 'rb') as image_file:
+ with open(volume_path, 'rb') as image_file:
image_service.update(context, image_id, {}, image_file)
else:
with utils.temporary_chown(volume_path):
- with fileutils.file_open(volume_path) as image_file:
+ with open(volume_path) as image_file:
image_service.update(context, image_id, {}, image_file)
return
reason=_("Converted to %(f1)s, but format is now %(f2)s") %
{'f1': image_meta['disk_format'], 'f2': data.file_format})
- with fileutils.file_open(tmp, 'rb') as image_file:
+ with open(tmp, 'rb') as image_file:
image_service.update(context, image_id, {}, image_file)
+++ /dev/null
-# Copyright 2011 OpenStack Foundation.
-# All Rights Reserved.
-#
-# Licensed under the Apache License, Version 2.0 (the "License"); you may
-# not use this file except in compliance with the License. You may obtain
-# a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
-# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
-# License for the specific language governing permissions and limitations
-# under the License.
-
-import contextlib
-import errno
-import logging
-import os
-import stat
-import tempfile
-
-from oslo_utils import excutils
-
-LOG = logging.getLogger(__name__)
-
-_FILE_CACHE = {}
-DEFAULT_MODE = stat.S_IRWXU | stat.S_IRWXG | stat.S_IRWXO
-
-
-def ensure_tree(path, mode=DEFAULT_MODE):
- """Create a directory (and any ancestor directories required)
-
- :param path: Directory to create
- :param mode: Directory creation permissions
- """
- try:
- os.makedirs(path, mode)
- except OSError as exc:
- if exc.errno == errno.EEXIST:
- if not os.path.isdir(path):
- raise
- else:
- raise
-
-
-def read_cached_file(filename, force_reload=False):
- """Read from a file if it has been modified.
-
- :param force_reload: Whether to reload the file.
- :returns: A tuple with a boolean specifying if the data is fresh
- or not.
- """
- global _FILE_CACHE
-
- if force_reload:
- delete_cached_file(filename)
-
- reloaded = False
- mtime = os.path.getmtime(filename)
- cache_info = _FILE_CACHE.setdefault(filename, {})
-
- if not cache_info or mtime > cache_info.get('mtime', 0):
- LOG.debug("Reloading cached file %s", filename)
- with open(filename) as fap:
- cache_info['data'] = fap.read()
- cache_info['mtime'] = mtime
- reloaded = True
- return (reloaded, cache_info['data'])
-
-
-def delete_cached_file(filename):
- """Delete cached file if present.
-
- :param filename: filename to delete
- """
- global _FILE_CACHE
-
- if filename in _FILE_CACHE:
- del _FILE_CACHE[filename]
-
-
-def delete_if_exists(path, remove=os.unlink):
- """Delete a file, but ignore file not found error.
-
- :param path: File to delete
- :param remove: Optional function to remove passed path
- """
-
- try:
- remove(path)
- except OSError as e:
- if e.errno != errno.ENOENT:
- raise
-
-
-@contextlib.contextmanager
-def remove_path_on_error(path, remove=delete_if_exists):
- """Protect code that wants to operate on PATH atomically.
- Any exception will cause PATH to be removed.
-
- :param path: File to work with
- :param remove: Optional function to remove passed path
- """
-
- try:
- yield
- except Exception:
- with excutils.save_and_reraise_exception():
- remove(path)
-
-
-def file_open(*args, **kwargs):
- """Open file
-
- see built-in open() documentation for more details
-
- Note: The reason this is kept in a separate module is to easily
- be able to provide a stub module that doesn't alter system
- state at all (for unit tests)
- """
- return open(*args, **kwargs)
-
-
-def write_to_tempfile(content, path=None, suffix='', prefix='tmp'):
- """Create temporary file or use existing file.
-
- This util is needed for creating temporary file with
- specified content, suffix and prefix. If path is not None,
- it will be used for writing content. If the path doesn't
- exist it'll be created.
-
- :param content: content for temporary file.
- :param path: same as parameter 'dir' for mkstemp
- :param suffix: same as parameter 'suffix' for mkstemp
- :param prefix: same as parameter 'prefix' for mkstemp
-
- For example: it can be used in database tests for creating
- configuration files.
- """
- if path:
- ensure_tree(path)
-
- (fd, path) = tempfile.mkstemp(suffix=suffix, dir=path, prefix=prefix)
- try:
- os.write(fd, content)
- finally:
- os.close(fd)
- return path
from swiftclient import client as swift
-from cinder.openstack.common import fileutils
-
class FakeSwiftClient2(object):
def __init__(self, *args, **kwargs):
if container == 'socket_error_on_get':
raise socket.error(111, 'ECONNREFUSED')
object_path = tempfile.gettempdir() + '/' + container + '/' + name
- with fileutils.file_open(object_path, 'rb') as object_file:
+ with open(object_path, 'rb') as object_file:
return (None, object_file.read())
def put_object(self, container, name, reader, content_length=None,
etag=None, chunk_size=None, content_type=None,
headers=None, query_string=None):
object_path = tempfile.gettempdir() + '/' + container + '/' + name
- with fileutils.file_open(object_path, 'wb') as object_file:
+ with open(object_path, 'wb') as object_file:
object_file.write(reader.read())
return hashlib.md5(reader.read()).hexdigest()
import tempfile
import mock
+from oslo_utils import fileutils
from oslo_utils import timeutils
-from cinder.openstack.common import fileutils
from cinder import test
from cinder.volume import configuration as conf
assert_called_once_with(snap_ref, volume_ref, volume['size'])
self.assertTrue(mock_delete_snapshot.called)
- @mock.patch('cinder.openstack.common.fileutils.delete_if_exists')
+ @mock.patch('oslo_utils.fileutils.delete_if_exists')
def test_delete_volume(self, mock_delete_if_exists):
volume = self._simple_volume()
volume_filename = 'volume-%s' % self.VOLUME_UUID
self.driver.copy_volume_to_image('', volume, '', '')
@mock.patch('cinder.volume.drivers.ibm.gpfs.GPFSDriver._delete_gpfs_file')
- @mock.patch('cinder.openstack.common.fileutils.file_open')
+ @mock.patch('six.moves.builtins.open')
@mock.patch('cinder.utils.temporary_chown')
@mock.patch('cinder.volume.drivers.ibm.gpfs.GPFSDriver._gpfs_redirect')
@mock.patch('cinder.volume.drivers.ibm.gpfs.GPFSDriver.'
mock_local_path.return_value = self.volumes_path
self.driver.backup_volume('', backup, backup_service)
- @mock.patch('cinder.openstack.common.fileutils.file_open')
+ @mock.patch('six.moves.builtins.open')
@mock.patch('cinder.utils.temporary_chown')
@mock.patch('cinder.volume.drivers.ibm.gpfs.GPFSDriver.local_path')
def test_restore_backup(self,
class TestUploadVolume(test.TestCase):
@mock.patch('cinder.image.image_utils.CONF')
- @mock.patch('cinder.image.image_utils.fileutils.file_open')
+ @mock.patch('six.moves.builtins.open')
@mock.patch('cinder.image.image_utils.qemu_img_info')
@mock.patch('cinder.image.image_utils.convert_image')
@mock.patch('cinder.image.image_utils.temporary_file')
@mock.patch('cinder.image.image_utils.utils.temporary_chown')
@mock.patch('cinder.image.image_utils.CONF')
- @mock.patch('cinder.image.image_utils.fileutils.file_open')
+ @mock.patch('six.moves.builtins.open')
@mock.patch('cinder.image.image_utils.qemu_img_info')
@mock.patch('cinder.image.image_utils.convert_image')
@mock.patch('cinder.image.image_utils.temporary_file')
@mock.patch('cinder.image.image_utils.utils.temporary_chown')
@mock.patch('cinder.image.image_utils.CONF')
- @mock.patch('cinder.image.image_utils.fileutils.file_open')
+ @mock.patch('six.moves.builtins.open')
@mock.patch('cinder.image.image_utils.qemu_img_info')
@mock.patch('cinder.image.image_utils.convert_image')
@mock.patch('cinder.image.image_utils.temporary_file')
mock_open.return_value.__enter__.return_value)
@mock.patch('cinder.image.image_utils.CONF')
- @mock.patch('cinder.image.image_utils.fileutils.file_open')
+ @mock.patch('six.moves.builtins.open')
@mock.patch('cinder.image.image_utils.qemu_img_info')
@mock.patch('cinder.image.image_utils.convert_image')
@mock.patch('cinder.image.image_utils.temporary_file')
volume['size'])
drv._delete_snapshot.assert_called_once_with(mock.ANY)
- @mock.patch('cinder.openstack.common.fileutils.delete_if_exists')
+ @mock.patch('oslo_utils.fileutils.delete_if_exists')
def test_delete_volume(self, mock_delete_if_exists):
volume = self._simple_volume()
volume_filename = 'volume-%s' % self.VOLUME_UUID
from cinder import context
from cinder import exception
from cinder.image import image_utils
-from cinder.openstack.common import fileutils
-from cinder.openstack.common import imageutils
from cinder import test
-from cinder import utils
from cinder.volume import configuration as conf
from cinder.volume.drivers import scality
self._driver.extend_volume(self.TEST_VOLUME, self.TEST_NEWSIZE)
- def test_backup_volume(self):
- self.mox = mox_lib.Mox()
- self._driver.db = self.mox.CreateMockAnything()
- self.mox.StubOutWithMock(self._driver.db, 'volume_get')
-
+ @mock.patch('six.moves.builtins.open')
+ @mock.patch('cinder.utils.temporary_chown')
+ @mock.patch('cinder.image.image_utils.qemu_img_info')
+ def test_backup_volume(self, qemu_img_info, temporary_chown, mock_open):
volume = {'id': '2', 'name': self.TEST_VOLNAME}
- self._driver.db.volume_get(context, volume['id']).AndReturn(volume)
-
- info = imageutils.QemuImgInfo()
- info.file_format = 'raw'
- self.mox.StubOutWithMock(image_utils, 'qemu_img_info')
- image_utils.qemu_img_info(self.TEST_VOLPATH).AndReturn(info)
-
- self.mox.StubOutWithMock(utils, 'temporary_chown')
- mock_tempchown = mock.MagicMock()
- utils.temporary_chown(self.TEST_VOLPATH).AndReturn(mock_tempchown)
-
- self.mox.StubOutWithMock(fileutils, 'file_open')
- mock_fileopen = mock.MagicMock()
- fileutils.file_open(self.TEST_VOLPATH).AndReturn(mock_fileopen)
-
backup = {'volume_id': volume['id']}
- mock_servicebackup = self.mox.CreateMockAnything()
- mock_servicebackup.backup(backup, mox_lib.IgnoreArg())
-
- self.mox.ReplayAll()
-
- self._driver.backup_volume(context, backup, mock_servicebackup)
-
- def test_restore_backup(self):
+ info = mock.Mock()
+ info.file_format = 'raw'
+ info.backing_file = None
+ qemu_img_info.return_value = info
+ backup_service = mock.Mock()
+
+ with mock.patch.object(self._driver, 'db') as db:
+ db.volume_get.return_value = volume
+
+ self._driver.backup_volume(context, backup, backup_service)
+
+ db.volume_get.assert_called_once_with(context, volume['id'])
+ qemu_img_info.assert_called_once_with(self.TEST_VOLPATH)
+ temporary_chown.asser_called_once_with(self.TEST_VOLPATH)
+ mock_open.assert_called_once_with(self.TEST_VOLPATH)
+ backup_service.backup.asser_called_once_with(backup, mock.ANY)
+
+ info.backing_file = 'not None'
+ self.assertRaises(exception.InvalidVolume,
+ self._driver.backup_volume,
+ context, backup, backup_service)
+
+ info.file_format = 'not raw'
+ self.assertRaises(exception.InvalidVolume,
+ self._driver.backup_volume,
+ context, backup, backup_service)
+
+ @mock.patch('six.moves.builtins.open')
+ @mock.patch('cinder.utils.temporary_chown')
+ def test_restore_backup(self, temporary_chown, mock_open):
volume = {'id': '2', 'name': self.TEST_VOLNAME}
-
- self.mox.StubOutWithMock(utils, 'temporary_chown')
- mock_tempchown = mock.MagicMock()
- utils.temporary_chown(self.TEST_VOLPATH).AndReturn(mock_tempchown)
-
- self.mox.StubOutWithMock(fileutils, 'file_open')
- mock_fileopen = mock.MagicMock()
- fileutils.file_open(self.TEST_VOLPATH, 'wb').AndReturn(mock_fileopen)
-
backup = {'id': 123, 'volume_id': volume['id']}
- mock_servicebackup = self.mox.CreateMockAnything()
- mock_servicebackup.restore(backup, volume['id'], mox_lib.IgnoreArg())
-
- self.mox.ReplayAll()
+ backup_service = mock.Mock()
self._driver.restore_backup(context, backup, volume,
- mock_servicebackup)
+ backup_service)
+
+ temporary_chown.asser_called_once_with(self.TEST_VOLPATH)
+ mock_open.assert_called_once_with(self.TEST_VOLPATH, 'wb')
+ backup_service.restore.asser_called_once_with(backup, volume['id'],
+ mock.ANY)
fake_name, fake_size)
@mock.patch.object(image_transfer, 'copy_stream_optimized_disk')
- @mock.patch('cinder.openstack.common.fileutils.file_open')
+ @mock.patch('six.moves.builtins.open')
@mock.patch.object(VMDK_DRIVER, '_temporary_file')
@mock.patch('oslo_utils.uuidutils.generate_uuid')
@mock.patch.object(VMDK_DRIVER, '_create_backing')
@mock.patch.object(VMDK_DRIVER, 'extend_volume')
@mock.patch.object(VMDK_DRIVER, '_restore_backing')
- @mock.patch('cinder.openstack.common.fileutils.file_open')
+ @mock.patch('six.moves.builtins.open')
@mock.patch.object(VMDK_DRIVER, '_temporary_file')
@mock.patch('oslo_utils.uuidutils.generate_uuid')
@mock.patch.object(VMDK_DRIVER, 'volumeops')
@mock.patch.object(VMDK_DRIVER, '_delete_temp_backing')
@mock.patch.object(image_transfer, 'download_stream_optimized_data')
- @mock.patch('cinder.openstack.common.fileutils.file_open')
+ @mock.patch('cinder.volume.drivers.vmware.vmdk.open')
@mock.patch.object(VMDK_DRIVER, 'volumeops')
- @mock.patch(
- 'cinder.volume.drivers.vmware.vmdk.VMwareEsxVmdkDriver._get_disk_type')
+ @mock.patch.object(VMDK_DRIVER, '_get_disk_type')
@mock.patch.object(VMDK_DRIVER, '_get_storage_profile_id')
@mock.patch.object(VMDK_DRIVER, 'session')
@mock.patch.object(VMDK_DRIVER, '_select_ds_for_volume')
_select_ds_for_volume)
@mock.patch.object(image_transfer, 'copy_stream_optimized_disk')
- @mock.patch('cinder.openstack.common.fileutils.file_open')
+ @mock.patch('six.moves.builtins.open')
@mock.patch.object(VMDK_DRIVER, '_temporary_file')
@mock.patch('oslo_utils.uuidutils.generate_uuid')
@mock.patch.object(VMDK_DRIVER, '_create_backing')
@mock.patch.object(VMDK_DRIVER, 'extend_volume')
@mock.patch.object(VMDK_DRIVER, '_restore_backing')
- @mock.patch('cinder.openstack.common.fileutils.file_open')
+ @mock.patch('six.moves.builtins.open')
@mock.patch.object(VMDK_DRIVER, '_temporary_file')
@mock.patch('oslo_utils.uuidutils.generate_uuid')
@mock.patch.object(VMDK_DRIVER, 'volumeops')
@mock.patch.object(VMDK_DRIVER, '_delete_temp_backing')
@mock.patch.object(image_transfer, 'download_stream_optimized_data')
- @mock.patch('cinder.openstack.common.fileutils.file_open')
+ @mock.patch('six.moves.builtins.open')
@mock.patch.object(VMDK_DRIVER, 'volumeops')
@mock.patch(
'cinder.volume.drivers.vmware.vmdk.VMwareEsxVmdkDriver._get_disk_type')
vops.update_backing_disk_uuid.assert_called_once_with(backing,
volume['id'])
- @mock.patch('cinder.openstack.common.fileutils.ensure_tree')
- @mock.patch('cinder.openstack.common.fileutils.delete_if_exists')
+ @mock.patch('oslo_utils.fileutils.ensure_tree')
+ @mock.patch('oslo_utils.fileutils.delete_if_exists')
@mock.patch('tempfile.mkstemp')
@mock.patch('os.close')
def test_temporary_file(
from taskflow.engines.action_engine import engine
from cinder.api import common
-from cinder.backup import driver as backup_driver
from cinder.brick.local_dev import lvm as brick_lvm
from cinder.compute import nova
from cinder import context
from cinder.image import image_utils
from cinder import keymgr
from cinder import objects
-from cinder.openstack.common import fileutils
import cinder.policy
from cinder import quota
from cinder import test
driver_name = "cinder.tests.unit.fake_driver.LoggingVolumeDriver"
@mock.patch.object(utils, 'temporary_chown')
- @mock.patch.object(fileutils, 'file_open')
+ @mock.patch('six.moves.builtins.open')
@mock.patch.object(os_brick.initiator.connector,
'get_connector_properties')
@mock.patch.object(db, 'volume_get')
mock_volume_get.assert_called_with(self.context, vol['id'])
@mock.patch.object(utils, 'temporary_chown')
- @mock.patch.object(fileutils, 'file_open')
+ @mock.patch('six.moves.builtins.open')
@mock.patch.object(os_brick.initiator.connector,
'get_connector_properties')
@mock.patch.object(db, 'volume_get')
'backup_use_temp_snapshot',
return_value=True)
@mock.patch.object(utils, 'temporary_chown')
- @mock.patch.object(fileutils, 'file_open')
+ @mock.patch('six.moves.builtins.open')
@mock.patch.object(os_brick.initiator.connector.LocalConnector,
'connect_volume')
@mock.patch.object(os_brick.initiator.connector.LocalConnector,
self.assertTrue(self.volume.driver.remove_export_snapshot.called)
self.assertTrue(self.volume.driver.delete_snapshot.called)
- def test_restore_backup(self):
+ @mock.patch.object(utils, 'temporary_chown')
+ @mock.patch.object(os_brick.initiator.connector,
+ 'get_connector_properties')
+ @mock.patch('six.moves.builtins.open')
+ def test_restore_backup(self,
+ mock_open,
+ mock_get_connector_properties,
+ mock_temporary_chown):
+ dev_null = '/dev/null'
vol = tests_utils.create_volume(self.context)
- backup = {'volume_id': vol['id'],
- 'id': 'backup-for-%s' % vol['id']}
+ backup = {'volume_id': vol['id'], 'id': 'backup-for-%s' % vol['id']}
properties = {}
- attach_info = {'device': {'path': '/dev/null'}}
+ attach_info = {'device': {'path': dev_null}}
root_helper = 'sudo cinder-rootwrap /etc/cinder/rootwrap.conf'
- backup_service = self.mox.CreateMock(backup_driver.BackupDriver)
- self.mox.StubOutWithMock(os_brick.initiator.connector,
- 'get_connector_properties')
- self.mox.StubOutWithMock(self.volume.driver, '_attach_volume')
- self.mox.StubOutWithMock(os, 'getuid')
- self.mox.StubOutWithMock(utils, 'execute')
- self.mox.StubOutWithMock(fileutils, 'file_open')
- self.mox.StubOutWithMock(self.volume.driver, '_detach_volume')
- self.mox.StubOutWithMock(self.volume.driver, 'terminate_connection')
-
- os_brick.initiator.connector.\
- get_connector_properties(root_helper, CONF.my_ip, False, False).\
- AndReturn(properties)
- self.volume.driver._attach_volume(self.context, vol, properties).\
- AndReturn((attach_info, vol))
- os.getuid()
- utils.execute('chown', None, '/dev/null', run_as_root=True)
- f = fileutils.file_open('/dev/null', 'wb').AndReturn(file('/dev/null'))
- backup_service.restore(backup, vol['id'], f)
- utils.execute('chown', 0, '/dev/null', run_as_root=True)
- self.volume.driver._detach_volume(self.context, attach_info, vol,
- properties)
- self.mox.ReplayAll()
- self.volume.driver.restore_backup(self.context, backup, vol,
- backup_service)
- self.mox.UnsetStubs()
+
+ volume_file = mock.MagicMock()
+ mock_open.return_value.__enter__.return_value = volume_file
+ mock_get_connector_properties.return_value = properties
+
+ self.volume.driver._attach_volume = mock.MagicMock()
+ self.volume.driver._attach_volume.return_value = attach_info, vol
+ self.volume.driver._detach_volume = mock.MagicMock()
+ self.volume.driver.terminate_connection = mock.MagicMock()
+ self.volume.driver.secure_file_operations_enabled = mock.MagicMock()
+ self.volume.driver.secure_file_operations_enabled.side_effect = (False,
+ True)
+ backup_service = mock.MagicMock()
+
+ for i in (1, 2):
+ self.volume.driver.restore_backup(self.context, backup, vol,
+ backup_service)
+
+ mock_get_connector_properties.assert_called_with(root_helper,
+ CONF.my_ip,
+ False, False)
+ self.volume.driver._attach_volume.assert_called_with(
+ self.context, vol, properties)
+ self.assertEqual(i, self.volume.driver._attach_volume.call_count)
+
+ self.volume.driver._detach_volume.assert_called_with(
+ self.context, attach_info, vol, properties)
+ self.assertEqual(i, self.volume.driver._detach_volume.call_count)
+
+ self.volume.driver.secure_file_operations_enabled.\
+ assert_called_with()
+ self.assertEqual(
+ i,
+ self.volume.driver.secure_file_operations_enabled.call_count
+ )
+
+ mock_temporary_chown.assert_called_once_with(dev_null)
+
+ mock_open.assert_called_with(dev_null, 'wb')
+ self.assertEqual(i, mock_open.call_count)
+
+ backup_service.restore.assert_called_with(backup, vol['id'],
+ volume_file)
+ self.assertEqual(i, backup_service.restore.call_count)
class LVMISCSIVolumeDriverTestCase(DriverTestCase):
lvm_driver.check_for_setup_error()
@mock.patch.object(utils, 'temporary_chown')
- @mock.patch.object(fileutils, 'file_open')
+ @mock.patch('six.moves.builtins.open')
@mock.patch.object(os_brick.initiator.connector,
'get_connector_properties')
@mock.patch.object(db, 'volume_get')
update)
@mock.patch.object(utils, 'temporary_chown')
- @mock.patch.object(fileutils, 'file_open')
+ @mock.patch('six.moves.builtins.open')
@mock.patch.object(os_brick.initiator.connector,
'get_connector_properties')
@mock.patch.object(db, 'volume_get')
from mox3 import mox
from oslo_config import cfg
+from oslo_utils import fileutils
from cinder.image import image_utils
-from cinder.openstack.common import fileutils
from cinder import test
from cinder.tests.unit.windows import db_fakes
from cinder.volume import configuration as conf
from cinder.i18n import _, _LE, _LW
from cinder.image import image_utils
from cinder import objects
-from cinder.openstack.common import fileutils
from cinder import utils
from cinder.volume import rpcapi as volume_rpcapi
from cinder.volume import throttling
# Secure network file systems will not chown files.
if self.secure_file_operations_enabled():
- with fileutils.file_open(device_path) as device_file:
+ with open(device_path) as device_file:
backup_service.backup(backup, device_file)
else:
with utils.temporary_chown(device_path):
- with fileutils.file_open(device_path) as device_file:
+ with open(device_path) as device_file:
backup_service.backup(backup, device_file)
finally:
# Secure network file systems will not chown files.
if self.secure_file_operations_enabled():
- with fileutils.file_open(volume_path, 'wb') as volume_file:
+ with open(volume_path, 'wb') as volume_file:
backup_service.restore(backup, volume['id'], volume_file)
else:
with utils.temporary_chown(volume_path):
- with fileutils.file_open(volume_path, 'wb') as volume_file:
+ with open(volume_path, 'wb') as volume_file:
backup_service.restore(backup, volume['id'],
volume_file)
from oslo_concurrency import processutils
from oslo_config import cfg
from oslo_log import log as logging
+from oslo_utils import fileutils
from oslo_utils import units
from cinder import exception
from cinder.i18n import _, _LE, _LI, _LW
from cinder.image import image_utils
-from cinder.openstack.common import fileutils
from cinder import utils
from cinder.volume import driver
from cinder.volume.drivers import remotefs as remotefs_drv
from cinder import exception
from cinder.i18n import _, _LE, _LI
from cinder.image import image_utils
-from cinder.openstack.common import fileutils
from cinder import utils
from cinder.volume import driver
from cinder.volume.drivers import nfs
def _do_backup(self, backup_path, backup, backup_service):
with utils.temporary_chown(backup_path):
- with fileutils.file_open(backup_path) as backup_file:
+ with open(backup_path) as backup_file:
backup_service.backup(backup, backup_file)
def backup_volume(self, context, backup, backup_service):
volume_path = self.local_path(volume)
with utils.temporary_chown(volume_path):
- with fileutils.file_open(volume_path, 'wb') as volume_file:
+ with open(volume_path, 'wb') as volume_file:
backup_service.restore(backup, volume['id'], volume_file)
def _migrate_volume(self, volume, host):
from cinder import exception
from cinder.i18n import _, _LE, _LI, _LW
from cinder.image import image_utils
-from cinder.openstack.common import fileutils
from cinder import utils
from cinder.volume import driver
from cinder.volume import utils as volutils
try:
with utils.temporary_chown(volume_path):
- with fileutils.file_open(volume_path) as volume_file:
+ with open(volume_path) as volume_file:
backup_service.backup(backup, volume_file)
finally:
if temp_snapshot:
"""Restore an existing backup to a new or existing volume."""
volume_path = self.local_path(volume)
with utils.temporary_chown(volume_path):
- with fileutils.file_open(volume_path, 'wb') as volume_file:
+ with open(volume_path, 'wb') as volume_file:
backup_service.restore(backup, volume['id'], volume_file)
def get_volume_stats(self, refresh=False):
from oslo_concurrency import processutils
from oslo_config import cfg
from oslo_log import log as logging
+from oslo_utils import fileutils
from cinder import compute
from cinder import exception
from cinder.i18n import _, _LI, _LW
from cinder.image import image_utils
-from cinder.openstack.common import fileutils
from cinder import utils
from cinder.volume.drivers import remotefs as remotefs_drv
from eventlet import tpool
from oslo_config import cfg
from oslo_log import log as logging
+from oslo_utils import fileutils
from oslo_utils import units
from six.moves import urllib
from cinder import exception
from cinder.i18n import _, _LE, _LI, _LW
from cinder.image import image_utils
-from cinder.openstack.common import fileutils
from cinder import utils
from cinder.volume import driver
from oslo_concurrency import lockutils
from oslo_config import cfg
from oslo_log import log as logging
+from oslo_utils import fileutils
from oslo_utils import units
from six.moves import urllib
from cinder import exception
from cinder.i18n import _, _LI
from cinder.image import image_utils
-from cinder.openstack.common import fileutils
from cinder import utils
from cinder.volume import driver
from cinder.volume import utils as volume_utils
raise exception.InvalidVolume(msg)
with utils.temporary_chown(volume_local_path):
- with fileutils.file_open(volume_local_path) as volume_file:
+ with open(volume_local_path) as volume_file:
backup_service.backup(backup, volume_file)
def restore_backup(self, context, backup, volume, backup_service):
{'backup': backup['id'], 'volume': volume['name']})
volume_local_path = self.local_path(volume)
with utils.temporary_chown(volume_local_path):
- with fileutils.file_open(volume_local_path, 'wb') as volume_file:
+ with open(volume_local_path, 'wb') as volume_file:
backup_service.restore(backup, volume['id'], volume_file)
from cinder import exception
from cinder.i18n import _, _LE
from cinder.image import image_utils
-from cinder.openstack.common import fileutils
from cinder.volume import driver
tmp)
self._try_execute(*cmd)
- with fileutils.file_open(tmp, 'rb') as image_file:
+ with open(tmp, 'rb') as image_file:
image_service.update(context, image_id, {}, image_file)
def create_snapshot(self, snapshot):
from oslo_config import cfg
from oslo_log import log as logging
from oslo_utils import excutils
+from oslo_utils import fileutils
from oslo_utils import units
from oslo_utils import uuidutils
from oslo_vmware import api
from cinder import exception
from cinder.i18n import _, _LE, _LI, _LW
-from cinder.openstack.common import fileutils
from cinder.volume import driver
from cinder.volume.drivers.vmware import datastore as hub
from cinder.volume.drivers.vmware import exceptions as vmdk_exceptions
host_ip = self.configuration.vmware_host_ip
vmdk_ds_file_path = self.volumeops.get_vmdk_path(backing)
- with fileutils.file_open(tmp_file_path, "wb") as tmp_file:
+ with open(tmp_file_path, "wb") as tmp_file:
image_transfer.copy_stream_optimized_disk(
context,
timeout,
{'tmp_path': tmp_file_path,
'backup_id': backup['id']})
self._download_vmdk(context, volume, backing, tmp_file_path)
- with fileutils.file_open(tmp_file_path, "rb") as tmp_file:
+ with open(tmp_file_path, "rb") as tmp_file:
LOG.debug("Calling backup service to backup file: %s.",
tmp_file_path)
backup_service.backup(backup, tmp_file)
timeout = self.configuration.vmware_image_transfer_timeout_secs
host_ip = self.configuration.vmware_host_ip
try:
- with fileutils.file_open(tmp_file_path, "rb") as tmp_file:
+ with open(tmp_file_path, "rb") as tmp_file:
vm_ref = image_transfer.download_stream_optimized_data(
context,
timeout,
"backup: %(backup_id)s.",
{'tmp_path': tmp_file_path,
'backup_id': backup['id']})
- with fileutils.file_open(tmp_file_path, "wb") as tmp_file:
+ with open(tmp_file_path, "wb") as tmp_file:
LOG.debug("Calling backup service to restore backup: "
"%(backup_id)s to file: %(tmp_path)s.",
{'backup_id': backup['id'],
from oslo_config import cfg
from oslo_log import log as logging
+from oslo_utils import fileutils
from oslo_utils import units
from cinder import exception
from cinder.i18n import _, _LI
from cinder.image import image_utils
-from cinder.openstack.common import fileutils
from cinder import utils
from cinder.volume.drivers import smbfs
from cinder.volume.drivers.windows import remotefs
from oslo_config import cfg
from oslo_log import log as logging
+from oslo_utils import fileutils
from cinder.image import image_utils
-from cinder.openstack.common import fileutils
from cinder.volume import driver
from cinder.volume.drivers.windows import constants
from cinder.volume.drivers.windows import vhdutils
from oslo_concurrency import processutils as putils
from oslo_log import log as logging
+from oslo_utils import fileutils
from oslo_utils import netutils
from cinder import exception
-from cinder.openstack.common import fileutils
from cinder.i18n import _LI, _LW, _LE
from cinder import utils
from cinder.volume.targets import iscsi
from oslo_concurrency import processutils as putils
from oslo_log import log as logging
+from oslo_utils import fileutils
from cinder import exception
-from cinder.openstack.common import fileutils
from cinder.i18n import _LI, _LW, _LE
from cinder import utils
from cinder.volume.targets import iscsi
# The list of modules to copy from oslo-incubator
module=config.generator
module=context
-module=fileutils
module=gettextutils
module=imageutils
module=install_venv_common
Code Review / openstack-build / cinder-build.git / commitdiff