openstack/cinder
Code Issues Proposed changes

Fix failure with "None" volume type in Pure drivers

Browse Source 
We were not checking if the type id was None before calling
the util method to get the volume type. Turns out the helper
will raise an exception if the type is None (intentionally).

This fixes the issue by checking first in the driver before
using the util method. Added unit test coverage for this
scenario as well.

Change-Id: Ie3c7b56fafb4a8c07853a9c1cd11e7e26444bdc1
Closes-Bug: #1600362
This commit is contained in:
Patrick East
2016-07-08 13:14:53 -07:00
parent 61602912f4
commit d4a61fcf56
2 changed files with 19 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
cinder/tests/unit/test_pure.py
View File

@@ -1594,9 +1594,19 @@ class PureBaseVolumeDriverTestCase(PureBaseSharedDriverTestCase):
def test_is_vol_replicated_has_repl_extra_specs(self, mock_get_vol_type):
mock_get_vol_type.return_value = REPLICATED_VOL_TYPE
volume = fake_volume.fake_volume_obj(mock.MagicMock())
volume.volume_type_id = REPLICATED_VOL_TYPE['id']
actual = self.driver._is_volume_replicated_type(volume)
self.assertTrue(actual)
@mock.patch('cinder.volume.volume_types.get_volume_type')
def test_is_vol_replicated_none_type(self, mock_get_vol_type):
mock_get_vol_type.side_effect = exception.InvalidVolumeType(reason='')
volume = fake_volume.fake_volume_obj(mock.MagicMock())
volume.volume_type = None
volume.volume_type_id = None
actual = self.driver._is_volume_replicated_type(volume)
self.assertFalse(actual)
@mock.patch('cinder.volume.volume_types.get_volume_type')
def test_is_vol_replicated_has_other_extra_specs(self, mock_get_vol_type):
vtype_test = deepcopy(NON_REPLICATED_VOL_TYPE)

16
cinder/volume/drivers/pure.py
View File

@@ -1423,14 +1423,16 @@ class PureBaseVolumeDriver(san.SanDriver):
def _is_volume_replicated_type(self, volume):
ctxt = context.get_admin_context()
volume_type = volume_types.get_volume_type(ctxt,
volume["volume_type_id"])
replication_flag = False
specs = volume_type.get("extra_specs")
if specs and EXTRA_SPECS_REPL_ENABLED in specs:
replication_capability = specs[EXTRA_SPECS_REPL_ENABLED]
# Do not validate settings, ignore invalid.
replication_flag = (replication_capability == "<is> True")
if volume["volume_type_id"]:
volume_type = volume_types.get_volume_type(
ctxt, volume["volume_type_id"])
specs = volume_type.get("extra_specs")
if specs and EXTRA_SPECS_REPL_ENABLED in specs:
replication_capability = specs[EXTRA_SPECS_REPL_ENABLED]
# Do not validate settings, ignore invalid.
replication_flag = (replication_capability == "<is> True")
return replication_flag
def _find_failover_target(self, secondary):