x/gantt
Code Issues Proposed changes

assertRaises(Exception, ...) considered harmful

Browse Source 
Expecting that Exception is raised can end up passing an a test when an
unexpected error occurs. For instance, errors in the unit test itself
can be masked:

https://review.openstack.org/4848
https://review.openstack.org/4873
https://review.openstack.org/4874

Change a variety of unit tests to expect a more specific exception so
we don't run into false positive tests in the future.

Change-Id: Ibc0c63b1f6b5574a3ce93d9f02c9d1ff5ac4a8b0
This commit is contained in:
Johannes Erdfelt
2012-03-03 17:53:41 +00:00
parent 60105a66d5
commit c4802fa465
9 changed files with 78 additions and 53 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
HACKING.rst
View File

@@ -181,6 +181,18 @@ For more information on creating unit tests and utilizing the testing
infrastructure in OpenStack Nova, please read nova/testing/README.rst.
Unit Tests and assertRaises
---------------------------
When asserting that a test should raise an exception, test against the
most specific exception possible. An overly broad exception type (like
Exception) can mask errors in the unit test itself.
Example::
self.assertRaises(exception.InstanceNotFound, db.instance_get_by_uuid,
elevated, instance_uuid)
openstack-common
----------------

4
nova/exception.py
View File

@@ -554,6 +554,10 @@ class NetworkHostNotSet(NovaException):
message = _("Host is not set to the network (%(network_id)s).")
class NetworkBusy(NovaException):
message = _("Network %(network)s has active ports, cannot delete.")
class DatastoreNotFound(NotFound):
message = _("Could not find the datastore reference(s) which the VM uses.")

4
nova/test.py
View File

@@ -115,6 +115,10 @@ def skip_if_fake(func):
return _skipper
class TestingException(Exception):
pass
class TestCase(unittest.TestCase):
"""Test case base class for all unit tests."""

4
nova/tests/fakelibvirt.py
View File

@@ -420,8 +420,8 @@ class Connection(object):
if allow_default_uri_connection:
uri = 'qemu:///session'
else:
raise Exception("URI was None, but fake libvirt is configured"
" to not accept this.")
raise ValueError("URI was None, but fake libvirt is "
"configured to not accept this.")
uri_whitelist = ['qemu:///system',
'qemu:///session',

53
nova/tests/test_api.py
View File

@@ -389,48 +389,45 @@ class ApiEc2TestCase(test.TestCase):
group.authorize('tcp', 80, 81, '0.0.0.0/0')
group.authorize('icmp', -1, -1, '0.0.0.0/0')
group.authorize('udp', 80, 81, '0.0.0.0/0')
def _assert(message, *args):
try:
group.authorize(*args)
except EC2ResponseError as e:
self.assertEqual(e.status, 400, 'Expected status to be 400')
self.assertIn(message, e.error_message, e.error_message)
else:
raise self.failureException, 'EC2ResponseError not raised'
# Invalid CIDR address
self.assertRaises(Exception,
group.authorize, 'tcp', 80, 81, '0.0.0.0/0444')
_assert('Invalid CIDR', 'tcp', 80, 81, '0.0.0.0/0444')
# Missing ports
self.assertRaises(Exception,
group.authorize, 'tcp', '0.0.0.0/0')
_assert('Not enough parameters', 'tcp', '0.0.0.0/0')
# from port cannot be greater than to port
self.assertRaises(Exception,
group.authorize, 'tcp', 100, 1, '0.0.0.0/0')
_assert('Invalid port range', 'tcp', 100, 1, '0.0.0.0/0')
# For tcp, negative values are not allowed
self.assertRaises(Exception,
group.authorize, 'tcp', -1, 1, '0.0.0.0/0')
_assert('Invalid port range', 'tcp', -1, 1, '0.0.0.0/0')
# For tcp, valid port range 1-65535
self.assertRaises(Exception,
group.authorize, 'tcp', 1, 65599, '0.0.0.0/0')
_assert('Invalid port range', 'tcp', 1, 65599, '0.0.0.0/0')
# For icmp, only -1:-1 is allowed for type:code
self.assertRaises(Exception,
group.authorize, 'icmp', -1, 0, '0.0.0.0/0')
_assert('An unknown error has occurred', 'icmp', -1, 0, '0.0.0.0/0')
# Non valid type:code
self.assertRaises(Exception,
group.authorize, 'icmp', 0, 3, '0.0.0.0/0')
_assert('An unknown error has occurred', 'icmp', 0, 3, '0.0.0.0/0')
# Invalid Cidr for ICMP type
self.assertRaises(Exception,
group.authorize, 'icmp', -1, -1, '0.0.444.0/4')
_assert('Invalid CIDR', 'icmp', -1, -1, '0.0.444.0/4')
# Invalid protocol
self.assertRaises(Exception,
group.authorize, 'xyz', 1, 14, '0.0.0.0/0')
_assert('An unknown error has occurred', 'xyz', 1, 14, '0.0.0.0/0')
# Invalid port
self.assertRaises(Exception,
group.authorize, 'tcp', " ", "81", '0.0.0.0/0')
_assert('An unknown error has occurred', 'tcp', " ", "81", '0.0.0.0/0')
# Invalid icmp port
self.assertRaises(Exception,
group.authorize, 'icmp', " ", "81", '0.0.0.0/0')
_assert('An unknown error has occurred', 'icmp', " ", "81",
'0.0.0.0/0')
# Invalid CIDR Address
self.assertRaises(Exception,
group.authorize, 'icmp', -1, -1, '0.0.0.0')
_assert('Invalid CIDR', 'icmp', -1, -1, '0.0.0.0')
# Invalid CIDR Address
self.assertRaises(Exception,
group.authorize, 'icmp', -1, -1, '0.0.0.0/')
_assert('Invalid CIDR', 'icmp', -1, -1, '0.0.0.0/')
# Invalid Cidr ports
self.assertRaises(Exception,
group.authorize, 'icmp', 1, 256, '0.0.0.0/0')
_assert('Invalid port range', 'icmp', 1, 256, '0.0.0.0/0')
self.expect_http()
self.mox.ReplayAll()

27
nova/tests/test_compute.py
View File

@@ -329,11 +329,11 @@ class ComputeTestCase(BaseTestCase):
the instance goes to ERROR state, keeping the task state
"""
def fake(*args, **kwargs):
raise Exception("Failed to block device mapping")
raise test.TestingException()
self.stubs.Set(nova.compute.manager.ComputeManager,
'_setup_block_device_mapping', fake)
instance_uuid = self._create_instance()
self.assertRaises(Exception, self.compute.run_instance,
self.assertRaises(test.TestingException, self.compute.run_instance,
self.context, instance_uuid)
#check state is failed even after the periodic poll
self._assert_state({'vm_state': vm_states.ERROR,
@@ -348,10 +348,10 @@ class ComputeTestCase(BaseTestCase):
Make sure that when there is a spawning problem,
the instance goes to ERROR state, keeping the task state"""
def fake(*args, **kwargs):
raise Exception("Failed to spawn")
raise test.TestingException()
self.stubs.Set(self.compute.driver, 'spawn', fake)
instance_uuid = self._create_instance()
self.assertRaises(Exception, self.compute.run_instance,
self.assertRaises(test.TestingException, self.compute.run_instance,
self.context, instance_uuid)
#check state is failed even after the periodic poll
self._assert_state({'vm_state': vm_states.ERROR,
@@ -718,13 +718,14 @@ class ComputeTestCase(BaseTestCase):
def test_snapshot_fails(self):
"""Ensure task_state is set to None if snapshot fails"""
def fake_snapshot(*args, **kwargs):
raise Exception("I don't want to create a snapshot")
raise test.TestingException()
self.stubs.Set(self.compute.driver, 'snapshot', fake_snapshot)
instance = self._create_fake_instance()
self.compute.run_instance(self.context, instance['uuid'])
self.assertRaises(Exception, self.compute.snapshot_instance,
self.assertRaises(test.TestingException,
self.compute.snapshot_instance,
self.context, instance['uuid'], "failing_snapshot")
self._assert_state({'task_state': None})
self.compute.terminate_instance(self.context, instance['uuid'])
@@ -1050,7 +1051,7 @@ class ComputeTestCase(BaseTestCase):
spectacular=True)
def throw_up(*args, **kwargs):
raise Exception()
raise test.TestingException()
def fake(*args, **kwargs):
pass
@@ -1077,7 +1078,7 @@ class ComputeTestCase(BaseTestCase):
migration_ref = db.migration_get_by_instance_and_status(context,
instance['uuid'], 'pre-migrating')
self.assertRaises(Exception, self.compute.finish_resize,
self.assertRaises(test.TestingException, self.compute.finish_resize,
context, instance['uuid'],
int(migration_ref['id']), {}, {})
@@ -1152,7 +1153,7 @@ class ComputeTestCase(BaseTestCase):
"""Ensure instance status set to Error on resize error"""
def throw_up(*args, **kwargs):
raise Exception()
raise test.TestingException()
self.stubs.Set(self.compute.driver, 'migrate_disk_and_power_off',
throw_up)
@@ -1169,8 +1170,8 @@ class ComputeTestCase(BaseTestCase):
instance_uuid, 'pre-migrating')
#verify
self.assertRaises(Exception, self.compute.resize_instance, context,
instance_uuid, migration_ref['id'], {})
self.assertRaises(test.TestingException, self.compute.resize_instance,
context, instance_uuid, migration_ref['id'], {})
instance = db.instance_get_by_uuid(context, instance_uuid)
self.assertEqual(instance['vm_state'], vm_states.ERROR)
@@ -1289,7 +1290,7 @@ class ComputeTestCase(BaseTestCase):
def test_resize_instance_handles_migration_error(self):
"""Ensure vm_state is ERROR when error occurs"""
def raise_migration_failure(*args):
raise Exception(reason='test failure')
raise test.TestingException()
self.stubs.Set(self.compute.driver,
'migrate_disk_and_power_off',
raise_migration_failure)
@@ -1303,7 +1304,7 @@ class ComputeTestCase(BaseTestCase):
filter_properties={})
migration_ref = db.migration_get_by_instance_and_status(context,
inst_ref['uuid'], 'pre-migrating')
self.assertRaises(Exception, self.compute.resize_instance,
self.assertRaises(test.TestingException, self.compute.resize_instance,
context, inst_ref['uuid'], migration_ref['id'], {})
inst_ref = db.instance_get_by_uuid(context, inst_ref['uuid'])
self.assertEqual(inst_ref['vm_state'], vm_states.ERROR)

2
nova/tests/test_fakelibvirt.py
View File

@@ -89,7 +89,7 @@ class FakeLibvirtTests(test.TestCase):
def _test_connect_method_can_refuse_None_uri(self, conn_method):
libvirt.allow_default_uri_connection = False
self.assertRaises(Exception, conn_method, None)
self.assertRaises(ValueError, conn_method, None)
def test_openReadOnly_can_refuse_None_uri(self):
conn_method = self.get_openReadOnly_curry_func()

5
nova/tests/test_quantum.py
View File

@@ -358,8 +358,9 @@ class QuantumManagerTestCase(QuantumNovaTestCase):
# make sure we aren't allowed to delete network with
# active port
self.assertRaises(Exception, self.net_man.delete_network,
ctx, None, n['uuid'])
self.assertRaises(exception.NetworkBusy,
self.net_man.delete_network,
ctx, None, n['uuid'])
def _check_vifs(self, expect_num_vifs):
ctx = context.RequestContext('user1', "").elevated()

20
nova/tests/test_vmwareapi.py
View File

@@ -21,6 +21,7 @@ Test suite for VMWareAPI.
from nova import context
from nova import db
from nova import exception
from nova import flags
from nova import test
from nova.compute import power_state
@@ -168,8 +169,8 @@ class VMWareAPIVMTestCase(test.TestCase):
def test_snapshot_non_existent(self):
self._create_instance_in_the_db()
self.assertRaises(Exception, self.conn.snapshot, self.context,
self.instance, "Test-Snapshot")
self.assertRaises(exception.InstanceNotFound, self.conn.snapshot,
self.context, self.instance, "Test-Snapshot")
def test_reboot(self):
self._create_vm()
@@ -182,7 +183,8 @@ class VMWareAPIVMTestCase(test.TestCase):
def test_reboot_non_existent(self):
self._create_instance_in_the_db()
self.assertRaises(Exception, self.conn.reboot, self.instance)
self.assertRaises(exception.InstanceNotFound, self.conn.reboot,
self.instance, self.network_info, 'SOFT')
def test_reboot_not_poweredon(self):
self._create_vm()
@@ -191,7 +193,8 @@ class VMWareAPIVMTestCase(test.TestCase):
self.conn.suspend(self.instance)
info = self.conn.get_info({'name': 1})
self._check_vm_info(info, power_state.PAUSED)
self.assertRaises(Exception, self.conn.reboot, self.instance)
self.assertRaises(exception.InstanceRebootFailure, self.conn.reboot,
self.instance, self.network_info, 'SOFT')
def test_suspend(self):
self._create_vm()
@@ -203,7 +206,8 @@ class VMWareAPIVMTestCase(test.TestCase):
def test_suspend_non_existent(self):
self._create_instance_in_the_db()
self.assertRaises(Exception, self.conn.suspend, self.instance)
self.assertRaises(exception.InstanceNotFound, self.conn.suspend,
self.instance)
def test_resume(self):
self._create_vm()
@@ -218,13 +222,15 @@ class VMWareAPIVMTestCase(test.TestCase):
def test_resume_non_existent(self):
self._create_instance_in_the_db()
self.assertRaises(Exception, self.conn.resume, self.instance)
self.assertRaises(exception.InstanceNotFound, self.conn.resume,
self.instance)
def test_resume_not_suspended(self):
self._create_vm()
info = self.conn.get_info({'name': 1})
self._check_vm_info(info, power_state.RUNNING)
self.assertRaises(Exception, self.conn.resume, self.instance)
self.assertRaises(exception.InstanceResumeFailure, self.conn.resume,
self.instance)
def test_get_info(self):
self._create_vm()