# License for the specific language governing permissions and limitations
# under the License.
-import base64
-import binascii
-
from oslo_config import cfg
from oslo_log import log as logging
+from oslo_serialization import base64
from oslo_serialization import jsonutils
from oslo_utils import versionutils
from oslo_versionedobjects import fields
-import six
from cinder import db
from cinder import exception
:raises: InvalidInput
"""
try:
- return jsonutils.loads(base64.decodestring(backup_url))
- except binascii.Error:
+ return jsonutils.loads(base64.decode_as_text(backup_url))
+ except TypeError:
msg = _("Can't decode backup record.")
except ValueError:
msg = _("Can't parse backup record.")
# We must update kwargs instead of record to ensure we don't overwrite
# "real" data from the backup
kwargs.update(record)
- retval = jsonutils.dumps(kwargs)
- if six.PY3:
- retval = retval.encode('utf-8')
- return base64.encodestring(retval)
+ retval = jsonutils.dump_as_bytes(kwargs)
+ return base64.encode_as_text(retval)
@base.CinderObjectRegistry.register
import ddt
import mock
from oslo_utils import timeutils
+import six
import webob
# needed for stubs to work
_mock_list_services.return_value = [backup_service]
req = webob.Request.blank('/v2/fake/backups/import_record')
+ if six.PY2:
+ backup_url = backup_url.encode('utf-8')
req.body = ('<backup-record backup_service="%(backup_service)s" '
'backup_url="%(backup_url)s"/>') \
% {'backup_url': backup_url,
from oslo_utils import timeutils
from cinder import context
+from cinder import exception
from cinder.objects import base as obj_base
from cinder import test
@staticmethod
def _compare(test, db, obj):
for field, value in db.items():
- if not hasattr(obj, field):
+ try:
+ getattr(obj, field)
+ except (AttributeError, exception.CinderException,
+ NotImplementedError):
+ # NotImplementedError: ignore "Cannot load 'projects' in the
+ # base class" error
continue
if field in ('modified_at', 'created_at',
# under the License.
import mock
+import six
from cinder.db.sqlalchemy import models
from cinder import exception
# for that field
backup.refresh()
self._compare(self, db_backup2, backup)
+ if six.PY3:
+ call_bool = mock.call.__bool__()
+ else:
+ call_bool = mock.call.__nonzero__()
backup_get.assert_has_calls([mock.call(self.context, '1'),
- mock.call.__nonzero__(),
+ call_bool,
mock.call(self.context, '1')])
self.assertEqual(status, volume.status)
self.assertEqual(size, volume.size)
dirty = volume.cinder_obj_get_changes()
- self.assertEqual(list(dirty_keys), dirty.keys())
+ self.assertEqual(list(dirty_keys), list(dirty.keys()))
for key, value in kwargs.items():
self.assertEqual(value, getattr(volume, key))
# under the License.
import mock
+import six
from cinder import exception
from cinder import objects
# value for that field
cgsnapshot.refresh()
self._compare(self, db_cgsnapshot2, cgsnapshot)
+ if six.PY3:
+ call_bool = mock.call.__bool__()
+ else:
+ call_bool = mock.call.__nonzero__()
cgsnapshot_get.assert_has_calls([mock.call(self.context, '1'),
- mock.call.__nonzero__(),
+ call_bool,
mock.call(self.context, '1')])
# under the License.
import mock
+import six
from cinder import exception
from cinder import objects
# new value for that field
cg.refresh()
self._compare(self, db_cg2, cg)
+ if six.PY3:
+ call_bool = mock.call.__bool__()
+ else:
+ call_bool = mock.call.__nonzero__()
consistencygroup_get.assert_has_calls([mock.call(self.context, '1'),
- mock.call.__nonzero__(),
+ call_bool,
mock.call(self.context, '1')])
# under the License.
import mock
+import six
from cinder import objects
from cinder.tests.unit import fake_service
# new value for that field
service.refresh()
self._compare(self, db_service2, service)
+ if six.PY3:
+ call_bool = mock.call.__bool__()
+ else:
+ call_bool = mock.call.__nonzero__()
service_get.assert_has_calls([mock.call(self.context, 123),
- mock.call.__nonzero__(),
+ call_bool,
mock.call(self.context, 123)])
@mock.patch('cinder.db.service_get_all_by_binary')
import copy
import mock
+import six
from oslo_log import log as logging
# value for that field
snapshot.refresh()
self._compare(self, db_snapshot2, snapshot)
+ if six.PY3:
+ call_bool = mock.call.__bool__()
+ else:
+ call_bool = mock.call.__nonzero__()
snapshot_get.assert_has_calls([mock.call(self.context, '1'),
- mock.call.__nonzero__(),
+ call_bool,
mock.call(self.context, '1')])
# under the License.
import mock
+import six
from cinder import context
from cinder import exception
# for that field
volume.refresh()
self._compare(self, db_volume2, volume)
+ if six.PY3:
+ call_bool = mock.call.__bool__()
+ else:
+ call_bool = mock.call.__nonzero__()
volume_get.assert_has_calls([mock.call(self.context, '1'),
- mock.call.__nonzero__(),
+ call_bool,
mock.call(self.context, '1')])
def test_metadata_aliases(self):
# under the License.
import mock
+import six
from cinder import objects
from cinder.tests.unit import fake_volume
# new value for that field
attachment.refresh()
self._compare(self, db_attachment2, attachment)
+ if six.PY3:
+ call_bool = mock.call.__bool__()
+ else:
+ call_bool = mock.call.__nonzero__()
attachment_get.assert_has_calls([mock.call(self.context, '1'),
- mock.call.__nonzero__(),
+ call_bool,
mock.call(self.context, '1')])
# under the License.
import mock
+import six
from cinder import objects
from cinder.tests.unit import fake_volume
# value for that field
volume_type.refresh()
self._compare(self, db_type2, volume_type)
+ if six.PY3:
+ call_bool = mock.call.__bool__()
+ else:
+ call_bool = mock.call.__nonzero__()
volume_type_get.assert_has_calls([mock.call(self.context, '1'),
- mock.call.__nonzero__(),
+ call_bool,
mock.call(self.context, '1')])
cinder.tests.unit.keymgr.test_key_mgr
cinder.tests.unit.keymgr.test_mock_key_mgr
cinder.tests.unit.keymgr.test_not_implemented_key_mgr
+cinder.tests.unit.objects.test_backup
+cinder.tests.unit.objects.test_base
+cinder.tests.unit.objects.test_cgsnapshot
+cinder.tests.unit.objects.test_consistencygroup
+cinder.tests.unit.objects.test_fields
+cinder.tests.unit.objects.test_objects
+cinder.tests.unit.objects.test_service
+cinder.tests.unit.objects.test_snapshot
+cinder.tests.unit.objects.test_volume
+cinder.tests.unit.objects.test_volume_attachment
+cinder.tests.unit.objects.test_volume_type
cinder.tests.unit.scheduler.test_allocated_capacity_weigher
cinder.tests.unit.scheduler.test_capacity_weigher
cinder.tests.unit.scheduler.test_chance_weigher