"Fossies" - the Fresh Open Source Software Archive  

Source code changes of the file "cinder/tests/unit/volume/test_volume.py" between
cinder-17.0.0.tar.gz and cinder-17.0.1.tar.gz

About: OpenStack Cinder (Core Service: Block Storage) provides persistent block storage to running instances. Its pluggable driver architecture facilitates the creation and management of block storage devices.
The "Victoria" series (latest release).

test_volume.py  (cinder-17.0.0)	:	test_volume.py  (cinder-17.0.1)
skipping to change at line 23		skipping to change at line 23
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the		# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations		# License for the specific language governing permissions and limitations
# under the License.		# under the License.
"""Tests for Volume Code."""		"""Tests for Volume Code."""
import datetime		import datetime
import enum		import enum
import time		import time
from unittest import mock		from unittest import mock
		import castellan
from castellan.common import exception as castellan_exception		from castellan.common import exception as castellan_exception
from castellan import key_manager		from castellan import key_manager
import ddt		import ddt
import eventlet		import eventlet
import os_brick.initiator.connectors.iscsi		import os_brick.initiator.connectors.iscsi
from oslo_concurrency import processutils		from oslo_concurrency import processutils
from oslo_config import cfg		from oslo_config import cfg
from oslo_utils import imageutils		from oslo_utils import imageutils
import six		import six
from taskflow.engines.action_engine import engine		from taskflow.engines.action_engine import engine
skipping to change at line 88		skipping to change at line 89
snap.project_id = fake.PROJECT_ID		snap.project_id = fake.PROJECT_ID
snap.volume_id = volume_id		snap.volume_id = volume_id
snap.status = fields.SnapshotStatus.CREATING		snap.status = fields.SnapshotStatus.CREATING
if metadata is not None:		if metadata is not None:
snap.metadata = metadata		snap.metadata = metadata
snap.update(kwargs)		snap.update(kwargs)
snap.create()		snap.create()
return snap		return snap
		class KeyObject(object):
		def get_encoded(arg):
		return "asdf".encode('utf-8')
		class KeyObject2(object):
		def get_encoded(arg):
		return "qwert".encode('utf-8')
@ddt.ddt		@ddt.ddt
class VolumeTestCase(base.BaseVolumeTestCase):		class VolumeTestCase(base.BaseVolumeTestCase):
def setUp(self):		def setUp(self):
super(VolumeTestCase, self).setUp()		super(VolumeTestCase, self).setUp()
self.patch('cinder.volume.volume_utils.clear_volume', autospec=True)		self.patch('cinder.volume.volume_utils.clear_volume', autospec=True)
self.expected_status = 'available'		self.expected_status = 'available'
self.service_id = 1		self.service_id = 1
self.user_context = context.RequestContext(user_id=fake.USER_ID,		self.user_context = context.RequestContext(user_id=fake.USER_ID,
project_id=fake.PROJECT_ID)		project_id=fake.PROJECT_ID)
skipping to change at line 1766		skipping to change at line 1775
mock_del_enc_key.assert_called_once_with(mock.ANY, # context		mock_del_enc_key.assert_called_once_with(mock.ANY, # context
mock.ANY, # keymgr		mock.ANY, # keymgr
fake.ENCRYPTION_KEY2_ID)		fake.ENCRYPTION_KEY2_ID)
else:		else:
mock_setup_enc_keys.assert_not_called()		mock_setup_enc_keys.assert_not_called()
mock_execute.assert_not_called()		mock_execute.assert_not_called()
mock_del_enc_key.assert_not_called()		mock_del_enc_key.assert_not_called()
mock_at.assert_called()		mock_at.assert_called()
mock_det.assert_called()		mock_det.assert_called()
		@mock.patch('cinder.db.sqlalchemy.api.volume_encryption_metadata_get')
		def test_setup_encryption_keys(self, mock_enc_metadata_get):
		key_mgr = fake_keymgr.fake_api()
		self.mock_object(castellan.key_manager, 'API', return_value=key_mgr)
		key_id = key_mgr.store(self.context, KeyObject())
		key2_id = key_mgr.store(self.context, KeyObject2())
		params = {'status': 'creating',
		'size': 1,
		'host': CONF.host,
		'encryption_key_id': key_id}
		vol = tests_utils.create_volume(self.context, **params)
		self.volume.create_volume(self.context, vol)
		db.volume_update(self.context,
		vol['id'],
		{'encryption_key_id': key_id})
		mock_enc_metadata_get.return_value = {'cipher': 'aes-xts-plain64',
		'key_size': 256,
		'provider': 'luks'}
		ctxt = context.get_admin_context()
		enc_info = {'encryption_key_id': key_id}
		with mock.patch('cinder.volume.volume_utils.create_encryption_key',
		return_value=key2_id):
		r = cinder.volume.flows.manager.create_volume.\
		CreateVolumeFromSpecTask._setup_encryption_keys(ctxt,
		vol,
		enc_info)
		(source_pass, new_pass, new_key_id) = r
		self.assertNotEqual(source_pass, new_pass)
		self.assertEqual(new_key_id, key2_id)
@mock.patch.object(key_manager, 'API', fake_keymgr.fake_api)		@mock.patch.object(key_manager, 'API', fake_keymgr.fake_api)
def test_create_volume_from_snapshot_with_encryption(self):		def test_create_volume_from_snapshot_with_encryption(self):
"""Test volume can be created from a snapshot of an encrypted volume"""		"""Test volume can be created from a snapshot of an encrypted volume"""
ctxt = context.get_admin_context()		ctxt = context.get_admin_context()
cipher = 'aes-xts-plain64'		cipher = 'aes-xts-plain64'
key_size = 256		key_size = 256
db.volume_type_create(ctxt,		db.volume_type_create(ctxt,
{'id': '61298380-0c12-11e3-bfd6-4b48424183be',		{'id': '61298380-0c12-11e3-bfd6-4b48424183be',
'name': 'LUKS'})		'name': 'LUKS'})
End of changes. 3 change blocks.
0 lines changed or deleted		43 lines changed or added

---