diff --git a/heat/tests/aws/test_volume.py b/heat/tests/aws/test_volume.py index 9a04bb9f11..978747f8bb 100644 --- a/heat/tests/aws/test_volume.py +++ b/heat/tests/aws/test_volume.py @@ -15,7 +15,6 @@ import copy from cinderclient import exceptions as cinder_exp import mock -import mox from oslo_config import cfg import six @@ -69,26 +68,24 @@ volume_template = ''' ''' -class VolumeTest(vt_base.BaseVolumeTest): +class VolumeTest(vt_base.VolumeTestCase): def setUp(self): super(VolumeTest, self).setUp() self.t = template_format.parse(volume_template) self.use_cinder = False - def _mock_create_volume(self, fv, stack_name, final_status='available'): - cinder.CinderClientPlugin._create().MultipleTimes().AndReturn( - self.cinder_fc) - vol_name = utils.PhysName(stack_name, 'DataVolume') - self.cinder_fc.volumes.create( - size=1, availability_zone='nova', - description=vol_name, - name=vol_name, - metadata={u'Usage': u'Wiki Data Volume'}).AndReturn( - vt_base.FakeVolume(fv)) - self.cinder_fc.volumes.get(fv.id).AndReturn(fv) + def _mock_create_volume(self, fv, stack_name, final_status='available', + mock_attachment=None): + self.vol_name = utils.PhysName(stack_name, 'DataVolume') + self.stack_name = stack_name + self.cinder_fc.volumes.create.return_value = vt_base.FakeVolume(fv) fv_ready = vt_base.FakeVolume(final_status, id=fv.id) - self.cinder_fc.volumes.get(fv.id).AndReturn(fv_ready) + if mock_attachment is not None: + results = [fv, fv_ready, mock_attachment] + else: + results = [fv, fv_ready, vt_base.FakeVolume('in-use')] + self.cinder_fc.volumes.get.side_effect = results return fv_ready def test_volume(self): @@ -97,75 +94,55 @@ class VolumeTest(vt_base.BaseVolumeTest): # create script fv = self._mock_create_volume(vt_base.FakeVolume('creating'), stack_name) - # failed delete due to in-use script - self.cinder_fc.volumes.get(fv.id).AndReturn( - vt_base.FakeVolume('in-use')) # delete script self._mock_delete_volume(fv) - self.m.ReplayAll() - stack = utils.parse_stack(self.t, stack_name=stack_name) rsrc = self.create_volume(self.t, stack, 'DataVolume') + # delete script + self._mock_delete_volume(fv) ex = self.assertRaises(exception.ResourceFailure, scheduler.TaskRunner(rsrc.destroy)) self.assertIn("Volume in use", six.text_type(ex)) + self.cinder_fc.volumes.get.side_effect = [ + vt_base.FakeVolume('available'), cinder_exp.NotFound('Not found')] scheduler.TaskRunner(rsrc.destroy)() - self.m.VerifyAll() - def test_volume_default_az(self): fv = vt_base.FakeVolume('creating') stack_name = 'test_volume_defaultaz_stack' # create script - nova.NovaClientPlugin._create().AndReturn(self.fc) - self.m.StubOutWithMock(instance.Instance, 'handle_create') - self.m.StubOutWithMock(instance.Instance, 'check_create_complete') - self.m.StubOutWithMock(instance.Instance, '_resolve_attribute') - self.m.StubOutWithMock(aws_vol.VolumeAttachment, - 'handle_create') - self.m.StubOutWithMock(aws_vol.VolumeAttachment, - 'check_create_complete') + self.patchobject(instance.Instance, 'handle_create') + self.patchobject(instance.Instance, 'check_create_complete', + return_value=True) + self.patchobject(instance.Instance, '_resolve_attribute', + return_value=None) + self.patchobject(aws_vol.VolumeAttachment, 'handle_create') + self.patchobject(aws_vol.VolumeAttachment, + 'check_create_complete', + return_value=True) - instance.Instance.handle_create().AndReturn(None) - instance.Instance.check_create_complete(None).AndReturn(True) - instance.Instance._resolve_attribute( - 'AvailabilityZone').MultipleTimes().AndReturn(None) - cinder.CinderClientPlugin._create().AndReturn( - self.cinder_fc) + cinder.CinderClientPlugin._create.return_value = self.cinder_fc self.stub_ImageConstraint_validate() self.stub_ServerConstraint_validate() self.stub_VolumeConstraint_validate() vol_name = utils.PhysName(stack_name, 'DataVolume') - self.cinder_fc.volumes.create( - size=1, availability_zone=None, - description=vol_name, - name=vol_name, - metadata={u'Usage': u'Wiki Data Volume'}).AndReturn(fv) - self.cinder_fc.volumes.get(fv.id).AndReturn(fv) + self.cinder_fc.volumes.create.return_value = fv fv_ready = vt_base.FakeVolume('available', id=fv.id) - self.cinder_fc.volumes.get(fv.id).AndReturn(fv_ready) - aws_vol.VolumeAttachment.handle_create().AndReturn(None) - aws_vol.VolumeAttachment.check_create_complete( - None).AndReturn(True) + self.cinder_fc.volumes.get.side_effect = [ + fv, fv_ready, cinder_exp.NotFound('Not found')] # delete script - self.m.StubOutWithMock(instance.Instance, 'handle_delete') - self.m.StubOutWithMock(aws_vol.VolumeAttachment, 'handle_delete') - self.m.StubOutWithMock(aws_vol.VolumeAttachment, - 'check_delete_complete') - instance.Instance.handle_delete().AndReturn(None) - self.cinder_fc.volumes.get('vol-123').AndRaise( - cinder_exp.NotFound('Not found')) cookie = object() - aws_vol.VolumeAttachment.handle_delete().AndReturn(cookie) - aws_vol.VolumeAttachment.check_delete_complete(cookie).AndReturn(True) - - self.m.ReplayAll() + self.patchobject(instance.Instance, 'handle_delete') + self.patchobject(aws_vol.VolumeAttachment, 'handle_delete', + return_value=cookie) + self.patchobject(aws_vol.VolumeAttachment, 'check_delete_complete', + return_value=True) stack = utils.parse_stack(self.t, stack_name=stack_name) stack._update_all_resource_data(True, False) @@ -177,7 +154,16 @@ class VolumeTest(vt_base.BaseVolumeTest): scheduler.TaskRunner(stack.delete)() - self.m.VerifyAll() + instance.Instance._resolve_attribute.assert_called_with( + 'AvailabilityZone') + self.cinder_fc.volumes.create.assert_called_once_with( + size=1, availability_zone=None, + description=vol_name, + name=vol_name, + metadata={u'Usage': u'Wiki Data Volume'}) + self.cinder_fc.volumes.get.assert_called_with('vol-123') + aws_vol.VolumeAttachment.check_delete_complete.assert_called_once_with( + cookie) def test_volume_create_error(self): fv = vt_base.FakeVolume('creating') @@ -186,16 +172,12 @@ class VolumeTest(vt_base.BaseVolumeTest): self._mock_create_volume(fv, stack_name, final_status='error') - self.m.ReplayAll() - stack = utils.parse_stack(self.t, stack_name=stack_name) ex = self.assertRaises(exception.ResourceFailure, self.create_volume, self.t, stack, 'DataVolume') self.assertIn('Went to status error due to "Unknown"', six.text_type(ex)) - self.m.VerifyAll() - def test_volume_bad_tags(self): stack_name = 'test_volume_bad_tags_stack' self.t['Resources']['DataVolume']['Properties'][ @@ -208,19 +190,17 @@ class VolumeTest(vt_base.BaseVolumeTest): "Resources.DataVolume.Properties.Tags[0]: " "Unknown Property Foo", six.text_type(ex)) - self.m.VerifyAll() - def test_volume_attachment_error(self): stack_name = 'test_volume_attach_error_stack' - self._mock_create_volume(vt_base.FakeVolume('creating'), - stack_name) - self._mock_create_server_volume_script( + mock_attachment = self._mock_create_server_volume_script( vt_base.FakeVolume('attaching'), final_status='error') + self._mock_create_volume(vt_base.FakeVolume('creating'), + stack_name, + mock_attachment=mock_attachment + ) self.stub_VolumeConstraint_validate() - self.m.ReplayAll() - stack = utils.parse_stack(self.t, stack_name=stack_name) self.create_volume(self.t, stack, 'DataVolume') @@ -229,239 +209,225 @@ class VolumeTest(vt_base.BaseVolumeTest): self.t, stack, 'MountPoint') self.assertIn("Volume attachment failed - Unknown status error", six.text_type(ex)) - - self.m.VerifyAll() + self.validate_mock_create_server_volume_script() def test_volume_attachment(self): stack_name = 'test_volume_attach_stack' - self._mock_create_volume(vt_base.FakeVolume('creating'), - stack_name) fva = self._mock_create_server_volume_script( vt_base.FakeVolume('attaching')) + self._mock_create_volume(vt_base.FakeVolume('creating'), + stack_name, mock_attachment=fva) self.stub_VolumeConstraint_validate() - # delete script - fva = vt_base.FakeVolume('in-use') - self.fc.volumes.get_server_volume(u'WikiDatabase', - 'vol-123').AndReturn(fva) - self.cinder_fc.volumes.get(fva.id).AndReturn(fva) - self.fc.volumes.delete_server_volume( - 'WikiDatabase', 'vol-123').MultipleTimes().AndReturn(None) - self.cinder_fc.volumes.get(fva.id).AndReturn( - vt_base.FakeVolume('detaching', id=fva.id)) - self.cinder_fc.volumes.get(fva.id).AndReturn( - vt_base.FakeVolume('available', id=fva.id)) - self.fc.volumes.get_server_volume(u'WikiDatabase', - 'vol-123').AndReturn(fva) - self.fc.volumes.get_server_volume( - u'WikiDatabase', 'vol-123').AndRaise(fakes_nova.fake_exception()) - - self.m.ReplayAll() - stack = utils.parse_stack(self.t, stack_name=stack_name) self.create_volume(self.t, stack, 'DataVolume') rsrc = self.create_attachment(self.t, stack, 'MountPoint') - scheduler.TaskRunner(rsrc.delete)() + # delete script + fva = vt_base.FakeVolume('in-use') + self.fc.volumes.get_server_volume.return_value = fva + self.cinder_fc.volumes.get.side_effect = [ + fva, vt_base.FakeVolume('detaching', id=fva.id), + vt_base.FakeVolume('available', id=fva.id) + ] + self.fc.volumes.delete_server_volume.return_value = None + self.fc.volumes.get_server_volume.side_effect = [ + fva, fakes_nova.fake_exception] - self.m.VerifyAll() + scheduler.TaskRunner(rsrc.delete)() + self.fc.volumes.get_server_volume.assert_called_with(u'WikiDatabase', + 'vol-123') + self.fc.volumes.delete_server_volume.assert_called_with( + 'WikiDatabase', 'vol-123') + self.fc.volumes.get_server_volume.assert_called_with( + u'WikiDatabase', 'vol-123') + self.validate_mock_create_server_volume_script() def test_volume_detachment_err(self): stack_name = 'test_volume_detach_err_stack' - self._mock_create_volume(vt_base.FakeVolume('creating'), - stack_name) fva = self._mock_create_server_volume_script( vt_base.FakeVolume('attaching')) + self._mock_create_volume(vt_base.FakeVolume('creating'), + stack_name, mock_attachment=fva) self.stub_VolumeConstraint_validate() - # delete script - fva = vt_base.FakeVolume('in-use') - self.fc.volumes.get_server_volume(u'WikiDatabase', - 'vol-123').AndReturn(fva) - self.cinder_fc.volumes.get(fva.id).AndReturn(fva) - - self.fc.volumes.delete_server_volume( - 'WikiDatabase', 'vol-123').AndRaise(fakes_nova.fake_exception(400)) - self.cinder_fc.volumes.get(fva.id).AndReturn( - vt_base.FakeVolume('available', id=fva.id)) - self.fc.volumes.get_server_volume(u'WikiDatabase', - 'vol-123').AndReturn(fva) - self.fc.volumes.get_server_volume( - u'WikiDatabase', 'vol-123').AndRaise(fakes_nova.fake_exception()) - self.m.ReplayAll() stack = utils.parse_stack(self.t, stack_name=stack_name) self.create_volume(self.t, stack, 'DataVolume') rsrc = self.create_attachment(self.t, stack, 'MountPoint') - scheduler.TaskRunner(rsrc.delete)() + # delete script + fva = vt_base.FakeVolume('in-use') + self.fc.volumes.get_server_volume.side_effect = [ + fva, fva, fakes_nova.fake_exception] + self.cinder_fc.volumes.get.side_effect = [ + fva, vt_base.FakeVolume('available', id=fva.id)] - self.m.VerifyAll() + exc = fakes_nova.fake_exception(400) + self.fc.volumes.delete_server_volume.side_effect = exc + + scheduler.TaskRunner(rsrc.delete)() + self.fc.volumes.get_server_volume.assert_called_with( + u'WikiDatabase', 'vol-123') + self.fc.volumes.delete_server_volume.assert_called_once_with( + 'WikiDatabase', 'vol-123') + self.validate_mock_create_server_volume_script() def test_volume_detach_non_exist(self): fv = vt_base.FakeVolume('creating') fva = vt_base.FakeVolume('in-use') stack_name = 'test_volume_detach_nonexist_stack' - self._mock_create_volume(fv, stack_name) - self._mock_create_server_volume_script(fva) + mock_attachment = self._mock_create_server_volume_script(fva) + self._mock_create_volume(fv, stack_name, + mock_attachment=mock_attachment) self.stub_VolumeConstraint_validate() - # delete script - self.fc.volumes.delete_server_volume(u'WikiDatabase', - 'vol-123').AndReturn(None) - self.cinder_fc.volumes.get(fva.id).AndRaise( - cinder_exp.NotFound('Not found')) - self.fc.volumes.get_server_volume(u'WikiDatabase', 'vol-123' - ).AndRaise( - fakes_nova.fake_exception()) - - self.m.ReplayAll() stack = utils.parse_stack(self.t, stack_name=stack_name) self.create_volume(self.t, stack, 'DataVolume') rsrc = self.create_attachment(self.t, stack, 'MountPoint') + # delete script + self.fc.volumes.delete_server_volume.return_value = None + self.cinder_fc.volumes.get.side_effect = cinder_exp.NotFound( + 'Not found') + + exc = fakes_nova.fake_exception() + self.fc.volumes.get_server_volume.side_effect = exc + scheduler.TaskRunner(rsrc.delete)() - self.m.VerifyAll() + self.fc.volumes.delete_server_volume.assert_called_once_with( + u'WikiDatabase', 'vol-123') + self.fc.volumes.get_server_volume.assert_called_with( + u'WikiDatabase', 'vol-123') + self.validate_mock_create_server_volume_script() def test_volume_detach_deleting_volume(self): fv = vt_base.FakeVolume('creating') fva = vt_base.FakeVolume('deleting') stack_name = 'test_volume_detach_deleting_volume_stack' - self._mock_create_volume(fv, stack_name) - self._mock_create_server_volume_script(fva) + mock_attachment = self._mock_create_server_volume_script(fva) + self._mock_create_volume(fv, stack_name, + mock_attachment=mock_attachment) self.stub_VolumeConstraint_validate() - # delete script - self.fc.volumes.delete_server_volume(u'WikiDatabase', - 'vol-123').AndReturn(None) - self.cinder_fc.volumes.get(fva.id).AndReturn(fva) - self.fc.volumes.get_server_volume(u'WikiDatabase', 'vol-123' - ).AndRaise( - fakes_nova.fake_exception()) - - self.m.ReplayAll() stack = utils.parse_stack(self.t, stack_name=stack_name) self.create_volume(self.t, stack, 'DataVolume') rsrc = self.create_attachment(self.t, stack, 'MountPoint') + # delete script + self.cinder_fc.volumes.get.return_value = fva + exc = fakes_nova.fake_exception + self.fc.volumes.get_server_volume.side_effect = exc + scheduler.TaskRunner(rsrc.delete)() - self.m.VerifyAll() + self.fc.volumes.delete_server_volume.assert_called_once_with( + u'WikiDatabase', 'vol-123') + self.validate_mock_create_server_volume_script() def test_volume_detach_with_latency(self): stack_name = 'test_volume_detach_latency_stack' - self._mock_create_volume(vt_base.FakeVolume('creating'), - stack_name) fva = self._mock_create_server_volume_script( vt_base.FakeVolume('attaching')) + self._mock_create_volume(vt_base.FakeVolume('creating'), + stack_name, mock_attachment=fva) self.stub_VolumeConstraint_validate() - # delete script - self.fc.volumes.get_server_volume(u'WikiDatabase', - 'vol-123').AndReturn(fva) - self.cinder_fc.volumes.get(fva.id).AndReturn(fva) - self.fc.volumes.delete_server_volume( - 'WikiDatabase', 'vol-123').MultipleTimes().AndReturn(None) - self.cinder_fc.volumes.get(fva.id).AndReturn( - vt_base.FakeVolume('in-use', id=fva.id)) - self.cinder_fc.volumes.get(fva.id).AndReturn( - vt_base.FakeVolume('detaching', id=fva.id)) - self.cinder_fc.volumes.get(fva.id).AndReturn( - vt_base.FakeVolume('available', id=fva.id)) - self.fc.volumes.get_server_volume(u'WikiDatabase', - 'vol-123').AndReturn(fva) - self.fc.volumes.get_server_volume( - u'WikiDatabase', 'vol-123').AndRaise(fakes_nova.fake_exception()) - - self.m.ReplayAll() - stack = utils.parse_stack(self.t, stack_name=stack_name) self.create_volume(self.t, stack, 'DataVolume') rsrc = self.create_attachment(self.t, stack, 'MountPoint') + # delete script + self.fc.volumes.get_server_volume.side_effect = [ + fva, fva, fakes_nova.fake_exception] + self.cinder_fc.volumes.get.side_effect = [ + fva, vt_base.FakeVolume('in-use', id=fva.id), + vt_base.FakeVolume('detaching', id=fva.id), + vt_base.FakeVolume('available', id=fva.id)] + scheduler.TaskRunner(rsrc.delete)() - self.m.VerifyAll() + self.fc.volumes.delete_server_volume.assert_called_once_with( + u'WikiDatabase', 'vol-123') + self.fc.volumes.get_server_volume.assert_called_with( + u'WikiDatabase', 'vol-123') + self.validate_mock_create_server_volume_script() def test_volume_detach_with_error(self): stack_name = 'test_volume_detach_werr_stack' - self._mock_create_volume(vt_base.FakeVolume('creating'), - stack_name) fva = self._mock_create_server_volume_script( vt_base.FakeVolume('attaching')) + self._mock_create_volume(vt_base.FakeVolume('creating'), + stack_name, mock_attachment=fva) self.stub_VolumeConstraint_validate() - # delete script - fva = vt_base.FakeVolume('in-use') - self.fc.volumes.delete_server_volume( - 'WikiDatabase', 'vol-123').AndReturn(None) - self.cinder_fc.volumes.get(fva.id).AndReturn( - vt_base.FakeVolume('error', id=fva.id)) - self.m.ReplayAll() - stack = utils.parse_stack(self.t, stack_name=stack_name) self.create_volume(self.t, stack, 'DataVolume') rsrc = self.create_attachment(self.t, stack, 'MountPoint') - detach_task = scheduler.TaskRunner(rsrc.delete) + # delete script + self.fc.volumes.delete_server_volume.return_value = None + fva = vt_base.FakeVolume('in-use') + self.cinder_fc.volumes.get.side_effect = [ + vt_base.FakeVolume('error', id=fva.id)] + + detach_task = scheduler.TaskRunner(rsrc.delete) ex = self.assertRaises(exception.ResourceFailure, detach_task) self.assertIn('Volume detachment failed - Unknown status error', six.text_type(ex)) - self.m.VerifyAll() + self.fc.volumes.delete_server_volume.assert_called_once_with( + u'WikiDatabase', 'vol-123') + self.validate_mock_create_server_volume_script() def test_volume_delete(self): stack_name = 'test_volume_delete_stack' fv = vt_base.FakeVolume('creating') self._mock_create_volume(fv, stack_name) - self.m.ReplayAll() self.t['Resources']['DataVolume']['DeletionPolicy'] = 'Delete' stack = utils.parse_stack(self.t, stack_name=stack_name) rsrc = self.create_volume(self.t, stack, 'DataVolume') - self.m.StubOutWithMock(rsrc, "handle_delete") - rsrc.handle_delete().AndReturn(None) - self.m.StubOutWithMock(rsrc, "check_delete_complete") - rsrc.check_delete_complete(None).AndReturn(True) - self.m.ReplayAll() + m_hd = mock.Mock(return_value=None) + rsrc.handle_delete = m_hd + m_cdc = mock.Mock(return_value=True) + rsrc.check_delete_complete = m_cdc scheduler.TaskRunner(rsrc.destroy)() - - self.m.VerifyAll() + m_cdc.assert_called_with(None) + m_hd.assert_called_once_with() def test_volume_deleting_delete(self): - fv = vt_base.FakeVolume('creating') + vt_base.FakeVolume('creating') stack_name = 'test_volume_deleting_stack' - fv = self._mock_create_volume(vt_base.FakeVolume('creating'), - stack_name) - - self.cinder_fc.volumes.get(fv.id).AndReturn( - vt_base.FakeVolume('deleting')) - self.cinder_fc.volumes.get(fv.id).AndReturn( - vt_base.FakeVolume('deleting')) - self.cinder_fc.volumes.get(fv.id).AndRaise( - cinder_exp.NotFound('NotFound')) - - self.m.ReplayAll() + self._mock_create_volume(vt_base.FakeVolume('creating'), + stack_name) stack = utils.parse_stack(self.t, stack_name=stack_name) rsrc = self.create_volume(self.t, stack, 'DataVolume') - scheduler.TaskRunner(rsrc.destroy)() - self.m.VerifyAll() + self.assertEqual(2, self.cinder_fc.volumes.get.call_count) + # delete script + self.cinder_fc.volumes.get.side_effect = [ + vt_base.FakeVolume('deleting'), + vt_base.FakeVolume('deleting'), + cinder_exp.NotFound('NotFound')] + + scheduler.TaskRunner(rsrc.destroy)() + self.assertEqual(5, self.cinder_fc.volumes.get.call_count) def test_volume_delete_error(self): fv = vt_base.FakeVolume('creating') @@ -470,30 +436,29 @@ class VolumeTest(vt_base.BaseVolumeTest): fv = self._mock_create_volume(vt_base.FakeVolume('creating'), stack_name) - self.cinder_fc.volumes.get(fv.id).AndReturn(fv) - self.cinder_fc.volumes.delete(fv.id).AndReturn(True) - self.cinder_fc.volumes.get(fv.id).AndReturn( - vt_base.FakeVolume('deleting')) - self.cinder_fc.volumes.get(fv.id).AndReturn( - vt_base.FakeVolume('error_deleting')) - - self.m.ReplayAll() - stack = utils.parse_stack(self.t, stack_name=stack_name) rsrc = self.create_volume(self.t, stack, 'DataVolume') + + self.assertEqual(2, self.cinder_fc.volumes.get.call_count) + self.cinder_fc.volumes.get.side_effect = [ + fv, + vt_base.FakeVolume('deleting'), + vt_base.FakeVolume('error_deleting')] + self.cinder_fc.volumes.delete.return_value = True + deleter = scheduler.TaskRunner(rsrc.destroy) self.assertRaisesRegex(exception.ResourceFailure, ".*ResourceInError.*error_deleting.*delete", deleter) - self.m.VerifyAll() + self.cinder_fc.volumes.delete.assert_called_once_with(fv.id) + self.assertEqual(5, self.cinder_fc.volumes.get.call_count) def test_volume_update_not_supported(self): stack_name = 'test_volume_updnotsup_stack' fv = vt_base.FakeVolume('creating') self._mock_create_volume(fv, stack_name) - self.m.ReplayAll() t = template_format.parse(volume_template) stack = utils.parse_stack(t, stack_name=stack_name) @@ -561,20 +526,17 @@ class VolumeTest(vt_base.BaseVolumeTest): def test_snapshot(self): stack_name = 'test_volume_snapshot_stack' - fv = self._mock_create_volume(vt_base.FakeVolume('creating'), - stack_name) + fv = vt_base.FakeVolume('creating') + fv_ready = vt_base.FakeVolume('available', id=fv.id) + fv = self._mock_create_volume(fv, + stack_name, mock_attachment=fv_ready) # snapshot script - self.m.StubOutWithMock(self.cinder_fc.backups, 'create') - self.m.StubOutWithMock(self.cinder_fc.backups, 'get') fb = vt_base.FakeBackup('available') - self.cinder_fc.backups.create(fv.id).AndReturn(fb) - self.cinder_fc.backups.get(fb.id).AndReturn(fb) - self.cinder_fc.volumes.get(fv.id).AndReturn(fv) + self.m_backups.create.return_value = fb + self.m_backups.get.return_value = fb self._mock_delete_volume(fv) - self.m.ReplayAll() - self.t['Resources']['DataVolume']['DeletionPolicy'] = 'Snapshot' stack = utils.parse_stack(self.t, stack_name=stack_name) @@ -582,7 +544,8 @@ class VolumeTest(vt_base.BaseVolumeTest): scheduler.TaskRunner(rsrc.destroy)() - self.m.VerifyAll() + self.m_backups.create.assert_called_once_with(fv.id) + self.m_backups.get.assert_called_once_with(fb.id) def test_snapshot_error(self): stack_name = 'test_volume_snapshot_err_stack' @@ -590,12 +553,9 @@ class VolumeTest(vt_base.BaseVolumeTest): stack_name) # snapshot script - self.m.StubOutWithMock(self.cinder_fc.backups, 'create') - self.m.StubOutWithMock(self.cinder_fc.backups, 'get') fb = vt_base.FakeBackup('error') - self.cinder_fc.backups.create(fv.id).AndReturn(fb) - self.cinder_fc.backups.get(fb.id).AndReturn(fb) - self.m.ReplayAll() + self.m_backups.create.return_value = fb + self.m_backups.get.return_value = fb self.t['Resources']['DataVolume']['DeletionPolicy'] = 'Snapshot' stack = utils.parse_stack(self.t, stack_name=stack_name) @@ -606,18 +566,20 @@ class VolumeTest(vt_base.BaseVolumeTest): scheduler.TaskRunner(rsrc.destroy)) self.assertIn('Unknown status error', six.text_type(ex)) - self.m.VerifyAll() + self.m_backups.create.assert_called_once_with(fv.id) + self.m_backups.get.assert_called_once_with(fb.id) def test_snapshot_no_volume(self): """Test that backup does not start for failed resource.""" stack_name = 'test_volume_snapshot_novol_stack' cfg.CONF.set_override('action_retry_limit', 0) + fva = vt_base.FakeVolume('error') fv = self._mock_create_volume(vt_base.FakeVolume('creating'), stack_name, - final_status='error') + final_status='error', + mock_attachment=fva) self._mock_delete_volume(fv) - self.m.ReplayAll() self.t['Resources']['DataVolume']['DeletionPolicy'] = 'Snapshot' self.t['Resources']['DataVolume']['Properties'][ @@ -635,56 +597,42 @@ class VolumeTest(vt_base.BaseVolumeTest): scheduler.TaskRunner(rsrc.destroy)() - self.m.VerifyAll() - def test_create_from_snapshot(self): stack_name = 'test_volume_create_from_snapshot_stack' fv = vt_base.FakeVolume('restoring-backup') fvbr = vt_base.FakeBackupRestore('vol-123') # create script - cinder.CinderClientPlugin._create().AndReturn( - self.cinder_fc) - self.patchobject(self.cinder_fc.backups, 'get') - self.m.StubOutWithMock(self.cinder_fc.restores, 'restore') - self.cinder_fc.restores.restore('backup-123').AndReturn(fvbr) - self.cinder_fc.volumes.get('vol-123').AndReturn(fv) + cinder.CinderClientPlugin._create.return_value = self.cinder_fc + self.m_restore.return_value = fvbr + fv2 = vt_base.FakeVolume('available') + self.cinder_fc.volumes.get.side_effect = [fv, fv2] vol_name = utils.PhysName(stack_name, 'DataVolume') - self.cinder_fc.volumes.update('vol-123', - description=vol_name, name=vol_name) - fv.status = 'available' - self.cinder_fc.volumes.get('vol-123').AndReturn(fv) - - self.m.ReplayAll() self.t['Resources']['DataVolume']['Properties'][ 'SnapshotId'] = 'backup-123' stack = utils.parse_stack(self.t, stack_name=stack_name) - self.create_volume(self.t, stack, 'DataVolume') + self.create_volume(self.t, stack, 'DataVolume', no_create=True) - self.m.VerifyAll() + cinder.CinderClientPlugin._create.assert_called_once_with() + self.m_restore.assert_called_once_with('backup-123') + self.cinder_fc.volumes.get.assert_called_with('vol-123') + self.cinder_fc.volumes.update.assert_called_once_with( + 'vol-123', description=vol_name, name=vol_name) def test_create_from_snapshot_error(self): stack_name = 'test_volume_create_from_snap_err_stack' cfg.CONF.set_override('action_retry_limit', 0) fv = vt_base.FakeVolume('restoring-backup') + fv2 = vt_base.FakeVolume('error') fvbr = vt_base.FakeBackupRestore('vol-123') # create script - cinder.CinderClientPlugin._create().AndReturn( - self.cinder_fc) - self.patchobject(self.cinder_fc.backups, 'get') - self.m.StubOutWithMock(self.cinder_fc.restores, 'restore') - self.cinder_fc.restores.restore('backup-123').AndReturn(fvbr) - self.cinder_fc.volumes.get('vol-123').AndReturn(fv) + cinder.CinderClientPlugin._create.return_value = self.cinder_fc + self.m_restore.return_value = fvbr + self.cinder_fc.volumes.get.side_effect = [fv, fv2] vol_name = utils.PhysName(stack_name, 'DataVolume') - self.cinder_fc.volumes.update(fv.id, - description=vol_name, name=vol_name) - fv.status = 'error' - self.cinder_fc.volumes.get('vol-123').AndReturn(fv) - - self.m.ReplayAll() self.t['Resources']['DataVolume']['Properties'][ 'SnapshotId'] = 'backup-123' @@ -695,7 +643,10 @@ class VolumeTest(vt_base.BaseVolumeTest): self.assertIn('Went to status error due to "Unknown"', six.text_type(ex)) - self.m.VerifyAll() + cinder.CinderClientPlugin._create.assert_called_once_with() + self.m_restore.assert_called_once_with('backup-123') + self.cinder_fc.volumes.update.assert_called_once_with( + fv.id, description=vol_name, name=vol_name) def test_volume_size_constraint(self): self.t['Resources']['DataVolume']['Properties']['Size'] = '0' @@ -708,19 +659,16 @@ class VolumeTest(vt_base.BaseVolumeTest): "0 is out of range (min: 1, max: None)", six.text_type(error)) def test_volume_attachment_updates_not_supported(self): - self.m.StubOutWithMock(nova.NovaClientPlugin, 'get_server') - nova.NovaClientPlugin.get_server(mox.IgnoreArg()).AndReturn( - mox.MockAnything()) + self.patchobject(nova.NovaClientPlugin, 'get_server') fv = vt_base.FakeVolume('creating') fva = vt_base.FakeVolume('attaching') stack_name = 'test_volume_attach_updnotsup_stack' - self._mock_create_volume(fv, stack_name) - self._mock_create_server_volume_script(fva) + mock_create_server_volume = self._mock_create_server_volume_script(fva) + self._mock_create_volume(fv, stack_name, + mock_attachment=mock_create_server_volume) self.stub_VolumeConstraint_validate() - self.m.ReplayAll() - stack = utils.parse_stack(self.t, stack_name=stack_name) self.create_volume(self.t, stack, 'DataVolume') @@ -739,7 +687,7 @@ class VolumeTest(vt_base.BaseVolumeTest): 'VolumeId of MountPoint (AWS::EC2::VolumeAttachment)', six.text_type(ex)) self.assertEqual((rsrc.UPDATE, rsrc.FAILED), rsrc.state) - self.m.VerifyAll() + self.validate_mock_create_server_volume_script() def test_validate_deletion_policy(self): cfg.CONF.set_override('backups_enabled', False, group='volumes') diff --git a/heat/tests/openstack/cinder/test_volume_utils.py b/heat/tests/openstack/cinder/test_volume_utils.py index ce4df1b905..177d038723 100644 --- a/heat/tests/openstack/cinder/test_volume_utils.py +++ b/heat/tests/openstack/cinder/test_volume_utils.py @@ -13,6 +13,7 @@ from cinderclient import exceptions as cinder_exp from cinderclient.v2 import client as cinderclient +import mock import six from heat.engine.clients.os import cinder @@ -23,6 +24,96 @@ from heat.engine import scheduler from heat.engine import stk_defn from heat.tests import common from heat.tests.openstack.nova import fakes as fakes_nova +from heat.tests import utils + + +class VolumeTestCase(common.HeatTestCase): + def setUp(self): + super(VolumeTestCase, self).setUp() + self.fc = fakes_nova.FakeClient() + self.cinder_fc = cinderclient.Client('username', 'password') + self.cinder_fc.volume_api_version = 2 + self.patchobject(cinder.CinderClientPlugin, '_create', + return_value=self.cinder_fc) + self.patchobject(nova.NovaClientPlugin, '_create', + return_value=self.fc) + self.cinder_fc.volumes = mock.Mock(spec=self.cinder_fc.volumes) + self.fc.volumes = mock.Mock() + self.use_cinder = False + self.m_backups = mock.Mock(spec=self.cinder_fc.backups) + self.m_restore = mock.Mock(spec=self.cinder_fc.restores.restore) + self.cinder_fc.backups = self.m_backups + self.cinder_fc.restores.restore = self.m_restore + + def _mock_delete_volume(self, fv): + self.cinder_fc.volumes.delete.return_value = True + + def validate_mock_create_server_volume_script(self): + self.fc.volumes.create_server_volume.assert_called_once_with( + device=u'/dev/vdc', server_id=u'WikiDatabase', volume_id='vol-123') + + def _mock_create_server_volume_script(self, fva, + final_status='in-use', + update=False, + extra_create_server_volume_mocks=[]): + if not update: + nova.NovaClientPlugin._create.return_value = self.fc + + result = [fva] + for m in extra_create_server_volume_mocks: + result.append(m) + self.fc.volumes.create_server_volume.side_effect = result + fv_ready = FakeVolume(final_status, id=fva.id) + return fv_ready + + def get_volume(self, t, stack, resource_name): + if self.use_cinder: + Volume = os_vol.CinderVolume + else: + data = t['Resources'][resource_name] + data['Properties']['AvailabilityZone'] = 'nova' + Volume = aws_vol.Volume + vol = Volume(resource_name, + stack.defn.resource_definition(resource_name), + stack) + return vol + + def create_volume(self, t, stack, resource_name, az='nova', + no_create=False): + rsrc = self.get_volume(t, stack, resource_name) + if isinstance(rsrc, os_vol.CinderVolume): + self.patchobject(rsrc, '_store_config_default_properties') + + self.assertIsNone(rsrc.validate()) + scheduler.TaskRunner(rsrc.create)() + self.assertEqual((rsrc.CREATE, rsrc.COMPLETE), rsrc.state) + stk_defn.update_resource_data(stack.defn, resource_name, + rsrc.node_data()) + if no_create: + self.cinder_fc.volumes.create.assert_not_called() + else: + self.vol_name = utils.PhysName(self.stack_name, 'DataVolume') + self.cinder_fc.volumes.create.assert_called_once_with( + size=1, availability_zone=az, + description=self.vol_name, + name=self.vol_name, + metadata={u'Usage': u'Wiki Data Volume'}) + return rsrc + + def create_attachment(self, t, stack, resource_name): + if self.use_cinder: + Attachment = os_vol.CinderVolumeAttachment + else: + Attachment = aws_vol.VolumeAttachment + rsrc = Attachment(resource_name, + stack.defn.resource_definition(resource_name), + stack) + self.assertIsNone(rsrc.validate()) + scheduler.TaskRunner(rsrc.create)() + self.assertEqual((rsrc.CREATE, rsrc.COMPLETE), rsrc.state) + stk_defn.update_resource_data(stack.defn, resource_name, + rsrc.node_data()) + return rsrc class BaseVolumeTest(common.HeatTestCase):