From 377a13308a43bb7caf2635bf0da0c7e9ab05800e Mon Sep 17 00:00:00 2001 From: Steve Baker Date: Thu, 20 Sep 2012 12:05:48 +1200 Subject: [PATCH] Unit tests for volumes and attachments Change-Id: Id8f1bab9d4cb782830e38a30a6411095d3710273 --- heat/engine/volume.py | 1 - heat/tests/test_volume.py | 223 ++++++++++++++++++++++++++++++++++++++ 2 files changed, 223 insertions(+), 1 deletion(-) create mode 100644 heat/tests/test_volume.py diff --git a/heat/engine/volume.py b/heat/engine/volume.py index 7b07599451..bd78bf2cb0 100644 --- a/heat/engine/volume.py +++ b/heat/engine/volume.py @@ -15,7 +15,6 @@ import eventlet from heat.openstack.common import log as logging -import re from heat.common import exception from heat.engine.resources import Resource diff --git a/heat/tests/test_volume.py b/heat/tests/test_volume.py new file mode 100644 index 0000000000..0bd1e36233 --- /dev/null +++ b/heat/tests/test_volume.py @@ -0,0 +1,223 @@ +# vim: tabstop=4 shiftwidth=4 softtabstop=4 + +# Licensed under the Apache License, Version 2.0 (the "License"); you may +# not use this file except in compliance with the License. You may obtain +# a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +# License for the specific language governing permissions and limitations +# under the License. + + +import sys +import os + +import json +import nose +import mox +import unittest + +from nose.plugins.attrib import attr + +from heat.engine import parser +from heat.engine import volume as vol +from heat.tests.v1_1 import fakes + + +@attr(tag=['unit', 'resource']) +@attr(speed='fast') +class VolumeTest(unittest.TestCase): + def setUp(self): + self.m = mox.Mox() + self.fc = fakes.FakeClient() + self.m.StubOutWithMock(vol.Volume, 'nova') + self.m.StubOutWithMock(vol.VolumeAttachment, 'nova') + self.m.StubOutWithMock(self.fc.volumes, 'create') + self.m.StubOutWithMock(self.fc.volumes, 'get') + self.m.StubOutWithMock(self.fc.volumes, 'delete') + self.m.StubOutWithMock(self.fc.volumes, 'create_server_volume') + self.m.StubOutWithMock(self.fc.volumes, 'delete_server_volume') + + def tearDown(self): + self.m.UnsetStubs() + print "VolumeTest teardown complete" + + def load_template(self): + self.path = os.path.dirname(os.path.realpath(__file__)).\ + replace('heat/tests', 'templates') + f = open("%s/WordPress_2_Instances_With_EBS.template" % self.path) + t = json.loads(f.read()) + f.close() + return t + + def parse_stack(self, t): + class DummyContext(): + tenant = 'test_tenant' + username = 'test_username' + password = 'password' + auth_url = 'http://localhost:5000/v2.0' + t['Parameters']['KeyName']['Value'] = 'test' + stack = parser.Stack(DummyContext(), 'test_stack', parser.Template(t), + 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) + 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) + return resource + + def test_volume(self): + + fv = FakeVolume('creating', 'available') + + # 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) + + # delete script + 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.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) + + resource = self.create_volume(t, stack, 'DataVolume') + self.assertEqual('available', fv.status) + + self.assertEqual(vol.Volume.UPDATE_REPLACE, + resource.handle_update()) + + fv.status = 'in-use' + self.assertEqual('Volume in use', resource.delete()) + fv.status = 'available' + self.assertEqual(None, resource.delete()) + + self.m.VerifyAll() + + def test_volume_create_error(self): + + fv = FakeVolume('creating', 'error') + + # 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) + + self.m.ReplayAll() + + t = self.load_template() + stack = self.parse_stack(t) + + resource = vol.Volume('DataVolume', + t['Resources']['DataVolume'], + stack) + self.assertEqual('error', resource.create()) + + self.m.VerifyAll() + + def test_volume_attachment_error(self): + + fv = FakeVolume('attaching', 'error') + + # create script + vol.VolumeAttachment.nova().AndReturn(self.fc) + self.fc.volumes.create_server_volume(device=u'/dev/vdc', + server_id=u'WikiDatabase', + volume_id=u'vol-123').AndReturn(fv) + + vol.VolumeAttachment.nova('volume').AndReturn(self.fc) + self.fc.volumes.get('vol-123').AndReturn(fv) + + self.m.ReplayAll() + + t = self.load_template() + stack = self.parse_stack(t) + + resource = vol.VolumeAttachment('MountPoint', + t['Resources']['MountPoint'], + stack) + self.assertEqual('error', resource.create()) + + self.m.VerifyAll() + + def test_volume_attachment(self): + + fv = FakeVolume('attaching', 'in-use') + + # create script + vol.VolumeAttachment.nova().AndReturn(self.fc) + self.fc.volumes.create_server_volume(device=u'/dev/vdc', + server_id=u'WikiDatabase', + volume_id=u'vol-123').AndReturn(fv) + + vol.VolumeAttachment.nova('volume').AndReturn(self.fc) + self.fc.volumes.get('vol-123').AndReturn(fv) + + # delete script + + fv = FakeVolume('in-use', 'available') + vol.VolumeAttachment.nova().AndReturn(self.fc) + 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) + + self.m.ReplayAll() + + t = self.load_template() + stack = self.parse_stack(t) + + resource = self.create_attachment(t, stack, 'MountPoint') + + self.assertEqual(vol.Volume.UPDATE_REPLACE, + resource.handle_update()) + + self.assertEqual(None, resource.delete()) + + self.m.VerifyAll() + + # allows testing of the test directly, shown below + if __name__ == '__main__': + sys.argv.append(__file__) + nose.main() + + +class FakeVolume: + status = 'attaching' + id = 'vol-123' + + def __init__(self, initial_status, final_status): + self.status = initial_status + self.final_status = final_status + + def get(self): + self.status = self.final_status