"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

Home  |  About  |  Features  |  All  |  Newest  |  Dox  |  Diffs  |  RSS Feeds  |  Screenshots  |  Comments  |  Imprint  |  Privacy  |  HTTP(S)