* PosixBackupDriver: open file in binary mode (to read/write).
Update test_backup_posix.py for that.
* test_backup_nfs: buffer() doesn't exist and is no more needed on
Python 3, only use buffer() on Python 2.
* tests-py3.txt: add cinder.tests.unit.backup
Partial-Implements: blueprint cinder-python3
Change-Id: I250d7378547df474f3c78024a737a3e2fa9bbaf4
def get_object_writer(self, container, object_name, extra_metadata=None):
path = os.path.join(self.backup_path, container, object_name)
- f = open(path, 'w')
+ f = open(path, 'wb')
permissions = (
stat.S_IRUSR |
stat.S_IWUSR |
def get_object_reader(self, container, object_name, extra_metadata=None):
path = os.path.join(self.backup_path, container, object_name)
- return open(path, 'r')
+ return open(path, 'rb')
def delete_object(self, container, object_name):
# TODO(tbarron): clean up the container path if it is empty
import mock
from os_brick.remotefs import remotefs as remotefs_brick
from oslo_config import cfg
+import six
from cinder.backup.drivers import nfs
from cinder import context
self.assertEqual(compressor, bz2)
self.assertRaises(ValueError, service._get_compressor, 'fake')
+ def create_buffer(self, size):
+ # Set up buffer of zeroed bytes
+ fake_data = bytearray(size)
+ if six.PY2:
+ # On Python 2, zlib.compressor() accepts buffer, but not bytearray
+ fake_data = buffer(fake_data)
+ return fake_data
+
def test_prepare_output_data_effective_compression(self):
service = nfs.NFSBackupDriver(self.ctxt)
- # Set up buffer of 128 zeroed bytes
- fake_data = buffer(bytearray(128))
-
+ fake_data = self.create_buffer(128)
result = service._prepare_output_data(fake_data)
self.assertEqual('zlib', result[0])
def test_prepare_output_data_no_compresssion(self):
self.flags(backup_compression_algorithm='none')
service = nfs.NFSBackupDriver(self.ctxt)
- # Set up buffer of 128 zeroed bytes
- fake_data = buffer(bytearray(128))
+ fake_data = self.create_buffer(128)
result = service._prepare_output_data(fake_data)
def test_prepare_output_data_ineffective_compression(self):
service = nfs.NFSBackupDriver(self.ctxt)
- # Set up buffer of 128 zeroed bytes
- fake_data = buffer(bytearray(128))
+ fake_data = self.create_buffer(128)
+
# Pre-compress so that compression in the driver will be ineffective.
already_compressed_data = service.compressor.compress(fake_data)
self.driver.get_object_writer(FAKE_CONTAINER, FAKE_OBJECT_NAME)
os.chmod.assert_called_once_with(FAKE_OBJECT_PATH, 0o660)
- builtins.open.assert_called_once_with(FAKE_OBJECT_PATH, 'w')
+ builtins.open.assert_called_once_with(FAKE_OBJECT_PATH, 'wb')
def test_get_object_reader(self):
self.mock_object(builtins, 'open', mock.mock_open())
self.driver.get_object_reader(FAKE_CONTAINER, FAKE_OBJECT_NAME)
- builtins.open.assert_called_once_with(FAKE_OBJECT_PATH, 'r')
+ builtins.open.assert_called_once_with(FAKE_OBJECT_PATH, 'rb')
def test_delete_object(self):
self.mock_object(os, 'remove')
cinder.tests.unit.api.test_versions
cinder.tests.unit.api.test_xmlutil
cinder.tests.unit.api.v2.test_volumes
+cinder.tests.unit.backup.drivers.test_backup_glusterfs
+cinder.tests.unit.backup.drivers.test_backup_nfs
+cinder.tests.unit.backup.drivers.test_backup_posix
+cinder.tests.unit.backup.test_rpcapi
cinder.tests.unit.image.test_cache
cinder.tests.unit.image.test_glance
cinder.tests.unit.keymgr.test_barbican