from heat.tests.v1_1 import fakes
-@attr(tag=['unit', 'resource'])
+@attr(tag=['unit', 'resource', 'volume'])
@attr(speed='fast')
class VolumeTest(unittest.TestCase):
def setUp(self):
f.close()
return t
- def parse_stack(self, t):
+ def parse_stack(self, t, stack_name):
ctx = context.RequestContext.from_dict({
'tenant': 'test_tenant',
'username': 'test_username',
'password': 'password',
'auth_url': 'http://localhost:5000/v2.0'})
template = parser.Template(t)
- params = parser.Parameters('test_stack', template, {'KeyName': 'test'})
- stack = parser.Stack(ctx, 'test_stack', template,
- params, stack_id=-1)
+ params = parser.Parameters(stack_name, template, {'KeyName': 'test'})
+ stack = parser.Stack(ctx, stack_name, template, params, stack_id=-1)
return stack
def create_volume(self, t, stack, resource_name):
resource = vol.Volume(resource_name,
- t['Resources'][resource_name],
- stack)
- self.assertEqual(None, resource.validate())
- self.assertEqual(None, resource.create())
- self.assertEqual(vol.Volume.CREATE_COMPLETE, resource.state)
+ t['Resources'][resource_name],
+ stack)
+ self.assertEqual(resource.validate(), None)
+ self.assertEqual(resource.create(), None)
+ self.assertEqual(resource.state, vol.Volume.CREATE_COMPLETE)
return resource
def create_attachment(self, t, stack, resource_name):
resource = vol.VolumeAttachment(resource_name,
- t['Resources'][resource_name],
- stack)
- self.assertEqual(None, resource.validate())
- self.assertEqual(None, resource.create())
- self.assertEqual(vol.VolumeAttachment.CREATE_COMPLETE,
- resource.state)
+ t['Resources'][resource_name],
+ stack)
+ self.assertEqual(resource.validate(), None)
+ self.assertEqual(resource.create(), None)
+ self.assertEqual(resource.state, vol.VolumeAttachment.CREATE_COMPLETE)
return resource
def test_volume(self):
-
fv = FakeVolume('creating', 'available')
+ stack_name = 'test_volume_stack'
# create script
- vol.Volume.nova('volume').AndReturn(self.fc)
+ vol.Volume.nova('volume').MultipleTimes().AndReturn(self.fc)
self.fc.volumes.create(u'1',
- display_description='test_stack.DataVolume',
- display_name='test_stack.DataVolume').AndReturn(fv)
+ display_description='%s.DataVolume' % stack_name,
+ display_name='%s.DataVolume' % stack_name).AndReturn(fv)
# delete script
- vol.Volume.nova('volume').AndReturn(self.fc)
self.fc.volumes.get('vol-123').AndReturn(fv)
eventlet.sleep(1).AndReturn(None)
- vol.Volume.nova('volume').AndReturn(self.fc)
self.fc.volumes.get('vol-123').AndReturn(fv)
- vol.Volume.nova('volume').AndReturn(self.fc)
self.fc.volumes.delete('vol-123').AndReturn(None)
self.m.ReplayAll()
t = self.load_template()
- stack = self.parse_stack(t)
+ stack = self.parse_stack(t, stack_name)
resource = self.create_volume(t, stack, 'DataVolume')
- self.assertEqual('available', fv.status)
+ self.assertEqual(fv.status, 'available')
- self.assertEqual(vol.Volume.UPDATE_REPLACE,
- resource.handle_update())
+ self.assertEqual(resource.handle_update(), vol.Volume.UPDATE_REPLACE)
fv.status = 'in-use'
- self.assertEqual('Volume in use', resource.delete())
+ self.assertEqual(resource.delete(), 'Volume in use')
fv.status = 'available'
- self.assertEqual(None, resource.delete())
+ self.assertEqual(resource.delete(), None)
self.m.VerifyAll()
def test_volume_create_error(self):
-
fv = FakeVolume('creating', 'error')
+ stack_name = 'test_volume_create_error_stack'
# create script
vol.Volume.nova('volume').AndReturn(self.fc)
self.fc.volumes.create(u'1',
- display_description='test_stack.DataVolume',
- display_name='test_stack.DataVolume').AndReturn(fv)
+ display_description='%s.DataVolume' % stack_name,
+ display_name='%s.DataVolume' % stack_name).AndReturn(fv)
eventlet.sleep(1).AndReturn(None)
self.m.ReplayAll()
t = self.load_template()
- stack = self.parse_stack(t)
+ stack = self.parse_stack(t, stack_name)
resource = vol.Volume('DataVolume',
- t['Resources']['DataVolume'],
- stack)
- self.assertEqual('error', resource.create())
+ t['Resources']['DataVolume'],
+ stack)
+ self.assertEqual(resource.create(), 'error')
self.m.VerifyAll()
def test_volume_attachment_error(self):
-
- fv = FakeVolume('attaching', 'error')
+ fva = FakeVolume('attaching', 'error')
+ stack_name = 'test_volume_attach_error_stack'
# create script
- vol.VolumeAttachment.nova().AndReturn(self.fc)
+ vol.VolumeAttachment.nova().MultipleTimes().AndReturn(self.fc)
+ vol.VolumeAttachment.nova('volume').MultipleTimes().AndReturn(self.fc)
+ eventlet.sleep(1).MultipleTimes().AndReturn(None)
self.fc.volumes.create_server_volume(device=u'/dev/vdc',
server_id=u'WikiDatabase',
- volume_id=u'vol-123').AndReturn(fv)
+ volume_id=u'vol-123').AndReturn(fva)
- vol.VolumeAttachment.nova('volume').AndReturn(self.fc)
- self.fc.volumes.get('vol-123').AndReturn(fv)
- eventlet.sleep(1).AndReturn(None)
+ self.fc.volumes.get('vol-123').AndReturn(fva)
self.m.ReplayAll()
t = self.load_template()
- stack = self.parse_stack(t)
+ stack = self.parse_stack(t, stack_name)
resource = vol.VolumeAttachment('MountPoint',
- t['Resources']['MountPoint'],
- stack)
- self.assertEqual('error', resource.create())
+ t['Resources']['MountPoint'],
+ stack)
+ self.assertEqual(resource.create(), 'error')
self.m.VerifyAll()
def test_volume_attachment(self):
-
- fv = FakeVolume('attaching', 'in-use')
+ fva = FakeVolume('attaching', 'in-use')
+ stack_name = 'test_volume_attach_stack'
# create script
- vol.VolumeAttachment.nova().AndReturn(self.fc)
+ vol.VolumeAttachment.nova().MultipleTimes().AndReturn(self.fc)
+ vol.VolumeAttachment.nova('volume').MultipleTimes().AndReturn(self.fc)
+ eventlet.sleep(1).MultipleTimes().AndReturn(None)
self.fc.volumes.create_server_volume(device=u'/dev/vdc',
server_id=u'WikiDatabase',
- volume_id=u'vol-123').AndReturn(fv)
+ volume_id=u'vol-123').AndReturn(fva)
- vol.VolumeAttachment.nova('volume').AndReturn(self.fc)
- self.fc.volumes.get('vol-123').AndReturn(fv)
- eventlet.sleep(1).AndReturn(None)
+ self.fc.volumes.get('vol-123').AndReturn(fva)
# delete script
-
- fv = FakeVolume('in-use', 'available')
- vol.VolumeAttachment.nova().AndReturn(self.fc)
+ fva = FakeVolume('in-use', 'available')
self.fc.volumes.delete_server_volume('WikiDatabase',
'vol-123').AndReturn(None)
- vol.VolumeAttachment.nova('volume').AndReturn(self.fc)
- self.fc.volumes.get('vol-123').AndReturn(fv)
- eventlet.sleep(1).AndReturn(None)
+ self.fc.volumes.get('vol-123').AndReturn(fva)
self.m.ReplayAll()
t = self.load_template()
- stack = self.parse_stack(t)
+ stack = self.parse_stack(t, stack_name)
resource = self.create_attachment(t, stack, 'MountPoint')
- self.assertEqual(vol.Volume.UPDATE_REPLACE,
- resource.handle_update())
+ self.assertEqual(resource.handle_update(), vol.Volume.UPDATE_REPLACE)
- self.assertEqual(None, resource.delete())
+ self.assertEqual(resource.delete(), None)
self.m.VerifyAll()