openstack
/
heat
Code Issues Proposed changes

Use unittest assert*() methods instead of assert keyword 

This gives much more useful messages when something fails

Change-Id: Ibdf22996e674beef6edabfd3f7ef7d2213b6f26c
Signed-off-by: Zane Bitter <zbitter@redhat.com>
This commit is contained in:
Zane Bitter 2012-07-03 11:25:33 +02:00
parent 82c61467a6
commit ab2ee441df
5 changed files with 55 additions and 48 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
heat/tests/test_resources.py
View File

@ -73,7 +73,7 @@ class instancesTest(unittest.TestCase):
instance.create() instance.create()
# this makes sure the auto increment worked on instance creation # this makes sure the auto increment worked on instance creation
assert(instance.id > 0) self.assertTrue(instance.id > 0)
def test_instance_create_delete(self): def test_instance_create_delete(self):
f = open("%s/WordPress_Single_Instance_gold.template" % self.path) f = open("%s/WordPress_Single_Instance_gold.template" % self.path)
@ -120,11 +120,11 @@ class instancesTest(unittest.TestCase):
instance.create() instance.create()
# this makes sure the auto increment worked on instance creation # this makes sure the auto increment worked on instance creation
assert(instance.id > 0) self.assertTrue(instance.id > 0)
instance.delete() instance.delete()
assert(instance.instance_id is None) self.assertTrue(instance.instance_id is None)
assert(instance.state == instance.DELETE_COMPLETE) self.assertEqual(instance.state, instance.DELETE_COMPLETE)
# allows testing of the test directly, shown below # allows testing of the test directly, shown below
if __name__ == '__main__': if __name__ == '__main__':

55
heat/tests/test_stacks.py
View File

@ -9,7 +9,7 @@ import sqlalchemy
from nose.plugins.attrib import attr from nose.plugins.attrib import attr
from nose import with_setup from nose import with_setup
from heat. common import context from heat.common import context
from heat.tests.v1_1 import fakes from heat.tests.v1_1 import fakes
from heat.engine import instance as instances from heat.engine import instance as instances
import heat.db as db_api import heat.db as db_api
@ -61,9 +61,10 @@ class stacksTest(unittest.TestCase):
stack = self.start_wordpress_stack('test_stack') stack = self.start_wordpress_stack('test_stack')
self.m.ReplayAll() self.m.ReplayAll()
stack.create() stack.create()
assert(stack.resources['WebServer'] is not None)
assert(stack.resources['WebServer'].instance_id > 0) self.assertNotEqual(stack.resources['WebServer'], None)
assert(stack.resources['WebServer'].ipaddress != '0.0.0.0') self.assertTrue(stack.resources['WebServer'].instance_id > 0)
self.assertNotEqual(stack.resources['WebServer'].ipaddress, '0.0.0.0')
def test_wordpress_single_instance_stack_delete(self): def test_wordpress_single_instance_stack_delete(self):
stack = self.start_wordpress_stack('test_stack') stack = self.start_wordpress_stack('test_stack')
@ -87,11 +88,13 @@ class stacksTest(unittest.TestCase):
pt['raw_template_id'] = new_rt.id pt['raw_template_id'] = new_rt.id
new_pt = db_api.parsed_template_create(None, pt) new_pt = db_api.parsed_template_create(None, pt)
stack.create() stack.create()
assert(stack.resources['WebServer'] is not None) self.assertNotEqual(stack.resources['WebServer'], None)
assert(stack.resources['WebServer'].instance_id > 0) self.assertTrue(stack.resources['WebServer'].instance_id > 0)
stack.delete() stack.delete()
assert(stack.resources['WebServer'].state == 'DELETE_COMPLETE')
assert(new_s.status == 'DELETE_COMPLETE') self.assertEqual(stack.resources['WebServer'].state, 'DELETE_COMPLETE')
self.assertEqual(new_s.status, 'DELETE_COMPLETE')
def test_stack_event_list(self): def test_stack_event_list(self):
stack = self.start_wordpress_stack('test_event_list_stack') stack = self.start_wordpress_stack('test_event_list_stack')
@ -115,27 +118,29 @@ class stacksTest(unittest.TestCase):
pt['raw_template_id'] = new_rt.id pt['raw_template_id'] = new_rt.id
new_pt = db_api.parsed_template_create(None, pt) new_pt = db_api.parsed_template_create(None, pt)
stack.create() stack.create()
assert(stack.resources['WebServer'] is not None)
assert(stack.resources['WebServer'].instance_id > 0) self.assertNotEqual(stack.resources['WebServer'], None)
self.assertTrue(stack.resources['WebServer'].instance_id > 0)
m = manager.EngineManager() m = manager.EngineManager()
events = db_api.event_get_all_by_stack(None, stack.id) events = db_api.event_get_all_by_stack(None, stack.id)
for ev in events: for ev in events:
result = m.parse_event(ev) result = m.parse_event(ev)
assert(result['EventId'] > 0) self.assertTrue(result['EventId'] > 0)
assert(result['StackName'] == "test_event_list_stack") self.assertEqual(result['StackName'], "test_event_list_stack")
assert(result['ResourceStatus'] in ('IN_PROGRESS', self.assertTrue(result['ResourceStatus'] in ('IN_PROGRESS',
'CREATE_COMPLETE')) 'CREATE_COMPLETE'))
assert(result['ResourceType'] == 'AWS::EC2::Instance') self.assertEqual(result['ResourceType'], 'AWS::EC2::Instance')
assert(result['ResourceStatusReason'] == 'state changed') self.assertEqual(result['ResourceStatusReason'], 'state changed')
assert(result['LogicalResourceId'] == 'WebServer') self.assertEqual(result['LogicalResourceId'], 'WebServer')
# Big long user data field.. it mentions 'wordpress' # Big long user data field.. it mentions 'wordpress'
# a few times so this should work. # a few times so this should work.
assert(result['ResourceProperties']['UserData'].find('wordpress') user_data = result['ResourceProperties']['UserData']
!= -1) self.assertNotEqual(user_data.find('wordpress'), -1)
assert(result['ResourceProperties']['ImageId'] self.assertEqual(result['ResourceProperties']['ImageId'],
== 'F16-x86_64-gold') 'F16-x86_64-gold')
assert(result['ResourceProperties']['InstanceType'] == 'm1.large') self.assertEqual(result['ResourceProperties']['InstanceType'],
'm1.large')
def test_stack_list(self): def test_stack_list(self):
stack = self.start_wordpress_stack('test_stack_list') stack = self.start_wordpress_stack('test_stack_list')
@ -178,10 +183,10 @@ class stacksTest(unittest.TestCase):
man = manager.EngineManager() man = manager.EngineManager()
sl = man.list_stacks(ctx, params) sl = man.list_stacks(ctx, params)
assert(len(sl) > 0) self.assertTrue(len(sl['stacks']) > 0)
for s in sl['stacks']: for s in sl['stacks']:
assert(s['StackId'] > 0) self.assertTrue(s['StackId'] > 0)
assert(s['TemplateDescription'].find('WordPress') != -1) self.assertNotEqual(s['TemplateDescription'].find('WordPress'), -1)
# allows testing of the test directly # allows testing of the test directly
if __name__ == '__main__': if __name__ == '__main__':

10
heat/tests/test_validate.py
View File

@ -239,7 +239,7 @@ class validateTest(unittest.TestCase):
self.m.ReplayAll() self.m.ReplayAll()
volumeattach = stack.resources['MountPoint'] volumeattach = stack.resources['MountPoint']
assert(volumeattach.validate()) self.assertTrue(volumeattach.validate())
def test_validate_ref_valid(self): def test_validate_ref_valid(self):
t = json.loads(test_template_ref % 'WikiDatabase') t = json.loads(test_template_ref % 'WikiDatabase')
@ -256,7 +256,7 @@ class validateTest(unittest.TestCase):
res = dict(manager. res = dict(manager.
validate_template(None, t, params)['ValidateTemplateResult']) validate_template(None, t, params)['ValidateTemplateResult'])
print 'res %s' % res print 'res %s' % res
assert (res['Description'] == 'Successfully validated') self.assertEqual(res['Description'], 'Successfully validated')
def test_validate_ref_invalid(self): def test_validate_ref_invalid(self):
t = json.loads(test_template_ref % 'WikiDatabasez') t = json.loads(test_template_ref % 'WikiDatabasez')
@ -272,7 +272,7 @@ class validateTest(unittest.TestCase):
manager = managers.EngineManager() manager = managers.EngineManager()
res = dict(manager. res = dict(manager.
validate_template(None, t, params)['ValidateTemplateResult']) validate_template(None, t, params)['ValidateTemplateResult'])
assert (res['Description'] != 'Successfully validated') self.assertNotEqual(res['Description'], 'Successfully validated')
def test_validate_findinmap_valid(self): def test_validate_findinmap_valid(self):
t = json.loads(test_template_findinmap_valid) t = json.loads(test_template_findinmap_valid)
@ -288,7 +288,7 @@ class validateTest(unittest.TestCase):
manager = managers.EngineManager() manager = managers.EngineManager()
res = dict(manager. res = dict(manager.
validate_template(None, t, params)['ValidateTemplateResult']) validate_template(None, t, params)['ValidateTemplateResult'])
assert (res['Description'] == 'Successfully validated') self.assertEqual(res['Description'], 'Successfully validated')
def test_validate_findinmap_invalid(self): def test_validate_findinmap_invalid(self):
t = json.loads(test_template_findinmap_invalid) t = json.loads(test_template_findinmap_invalid)
@ -304,7 +304,7 @@ class validateTest(unittest.TestCase):
manager = managers.EngineManager() manager = managers.EngineManager()
res = dict(manager. res = dict(manager.
validate_template(None, t, params)['ValidateTemplateResult']) validate_template(None, t, params)['ValidateTemplateResult'])
assert (res['Description'] != 'Successfully validated') self.assertNotEqual(res['Description'], 'Successfully validated')
# allows testing of the test directly, shown below # allows testing of the test directly, shown below
if __name__ == '__main__': if __name__ == '__main__':

8
heat/tests/test_waitcondition.py
View File

@ -82,11 +82,12 @@ class stacksTest(unittest.TestCase):
self.m.ReplayAll() self.m.ReplayAll()
self.greenpool.spawn_n(stack.create) self.greenpool.spawn_n(stack.create)
eventlet.sleep(1) eventlet.sleep(1)
assert(stack.resources['WaitForTheHandle'].state == 'IN_PROGRESS') self.assertEqual(stack.resources['WaitForTheHandle'].state,
'IN_PROGRESS')
r = db_api.resource_get_by_name_and_stack(None, 'WaitHandle', r = db_api.resource_get_by_name_and_stack(None, 'WaitHandle',
stack.id) stack.id)
assert(r.name == 'WaitHandle') self.assertEqual(r.name, 'WaitHandle')
metadata = {"Status": "SUCCESS", metadata = {"Status": "SUCCESS",
"Reason": "woot toot", "Reason": "woot toot",
@ -98,7 +99,8 @@ class stacksTest(unittest.TestCase):
eventlet.sleep(2) eventlet.sleep(2)
logger.debug('state %s' % stack.resources['WaitForTheHandle'].state) logger.debug('state %s' % stack.resources['WaitForTheHandle'].state)
assert(stack.resources['WaitForTheHandle'].state == 'CREATE_COMPLETE') self.assertEqual(stack.resources['WaitForTheHandle'].state,
'CREATE_COMPLETE')
self.greenpool.waitall() self.greenpool.waitall()

22
heat/tests/test_watch.py
View File

@ -46,13 +46,13 @@ class WatchRuleTest(unittest.TestCase):
watcher = watchrule.WatchRule(rule, data, last, now) watcher = watchrule.WatchRule(rule, data, last, now)
new_state = watcher.get_alarm_state() new_state = watcher.get_alarm_state()
logger.info(new_state) logger.info(new_state)
assert(new_state == 'NORMAL') self.assertEqual(new_state, 'NORMAL')
data.append(WatchData(25, now - datetime.timedelta(seconds=250))) data.append(WatchData(25, now - datetime.timedelta(seconds=250)))
watcher = watchrule.WatchRule(rule, data, last, now) watcher = watchrule.WatchRule(rule, data, last, now)
new_state = watcher.get_alarm_state() new_state = watcher.get_alarm_state()
logger.info(new_state) logger.info(new_state)
assert(new_state == 'ALARM') self.assertEqual(new_state, 'ALARM')
@attr(tag=['unit', 'watchrule']) @attr(tag=['unit', 'watchrule'])
@attr(speed='fast') @attr(speed='fast')
@ -74,13 +74,13 @@ class WatchRuleTest(unittest.TestCase):
watcher = watchrule.WatchRule(rule, data, last, now) watcher = watchrule.WatchRule(rule, data, last, now)
new_state = watcher.get_alarm_state() new_state = watcher.get_alarm_state()
logger.info(new_state) logger.info(new_state)
assert(new_state == 'NORMAL') self.assertEqual(new_state, 'NORMAL')
data.append(WatchData(35, now - datetime.timedelta(seconds=150))) data.append(WatchData(35, now - datetime.timedelta(seconds=150)))
watcher = watchrule.WatchRule(rule, data, last, now) watcher = watchrule.WatchRule(rule, data, last, now)
new_state = watcher.get_alarm_state() new_state = watcher.get_alarm_state()
logger.info(new_state) logger.info(new_state)
assert(new_state == 'ALARM') self.assertEqual(new_state, 'ALARM')
@attr(tag=['unit', 'watchrule']) @attr(tag=['unit', 'watchrule'])
@attr(speed='fast') @attr(speed='fast')
@ -103,14 +103,14 @@ class WatchRuleTest(unittest.TestCase):
watcher = watchrule.WatchRule(rule, data, last, now) watcher = watchrule.WatchRule(rule, data, last, now)
new_state = watcher.get_alarm_state() new_state = watcher.get_alarm_state()
logger.info(new_state) logger.info(new_state)
assert(new_state == 'NORMAL') self.assertEqual(new_state, 'NORMAL')
# only 3 samples -> ALARM # only 3 samples -> ALARM
data.append(WatchData(1, now - datetime.timedelta(seconds=200))) data.append(WatchData(1, now - datetime.timedelta(seconds=200)))
watcher = watchrule.WatchRule(rule, data, last, now) watcher = watchrule.WatchRule(rule, data, last, now)
new_state = watcher.get_alarm_state() new_state = watcher.get_alarm_state()
logger.info(new_state) logger.info(new_state)
assert(new_state == 'ALARM') self.assertEqual(new_state, 'ALARM')
# only 3 samples (one old) -> NORMAL # only 3 samples (one old) -> NORMAL
data.pop(0) data.pop(0)
@ -118,7 +118,7 @@ class WatchRuleTest(unittest.TestCase):
watcher = watchrule.WatchRule(rule, data, last, now) watcher = watchrule.WatchRule(rule, data, last, now)
new_state = watcher.get_alarm_state() new_state = watcher.get_alarm_state()
logger.info(new_state) logger.info(new_state)
assert(new_state == 'NORMAL') self.assertEqual(new_state, 'NORMAL')
@attr(tag=['unit', 'watchrule']) @attr(tag=['unit', 'watchrule'])
@attr(speed='fast') @attr(speed='fast')
@ -140,14 +140,14 @@ class WatchRuleTest(unittest.TestCase):
watcher = watchrule.WatchRule(rule, data, last, now) watcher = watchrule.WatchRule(rule, data, last, now)
new_state = watcher.get_alarm_state() new_state = watcher.get_alarm_state()
logger.info(new_state) logger.info(new_state)
assert(new_state == 'NORMAL') self.assertEqual(new_state, 'NORMAL')
# sum > 100 -> ALARM # sum > 100 -> ALARM
data.append(WatchData(85, now - datetime.timedelta(seconds=150))) data.append(WatchData(85, now - datetime.timedelta(seconds=150)))
watcher = watchrule.WatchRule(rule, data, last, now) watcher = watchrule.WatchRule(rule, data, last, now)
new_state = watcher.get_alarm_state() new_state = watcher.get_alarm_state()
logger.info(new_state) logger.info(new_state)
assert(new_state == 'ALARM') self.assertEqual(new_state, 'ALARM')
@attr(tag=['unit', 'watchrule']) @attr(tag=['unit', 'watchrule'])
@attr(speed='fast') @attr(speed='fast')
@ -168,10 +168,10 @@ class WatchRuleTest(unittest.TestCase):
watcher = watchrule.WatchRule(rule, data, last, now) watcher = watchrule.WatchRule(rule, data, last, now)
new_state = watcher.get_alarm_state() new_state = watcher.get_alarm_state()
logger.info(new_state) logger.info(new_state)
assert(new_state == 'NORMAL') self.assertEqual(new_state, 'NORMAL')
data.append(WatchData(195, now - datetime.timedelta(seconds=250))) data.append(WatchData(195, now - datetime.timedelta(seconds=250)))
watcher = watchrule.WatchRule(rule, data, last, now) watcher = watchrule.WatchRule(rule, data, last, now)
new_state = watcher.get_alarm_state() new_state = watcher.get_alarm_state()
logger.info(new_state) logger.info(new_state)
assert(new_state == 'ALARM') self.assertEqual(new_state, 'ALARM')