self._as_root([ip_version], tuple(args))
except RuntimeError as rte:
with (excutils.save_and_reraise_exception()) as ctx:
- if "Cannot find device" in rte.message:
+ if "Cannot find device" in str(rte):
ctx.reraise = False
raise exceptions.DeviceNotFoundError(
device_name=self.name)
# Unexpected errors are the responsibility of the caller
with excutils.save_and_reraise_exception() as ctxt:
# Exception has already been logged by execute
- no_children_found = 'Exit code: 1' in e.message
+ no_children_found = 'Exit code: 1' in str(e)
if no_children_found:
ctxt.reraise = False
return []
elif isinstance(translatable, webob.exc.HTTPError):
translatable.detail = localize(translatable.detail, locale)
elif isinstance(translatable, Exception):
- translatable.message = localize(translatable.message, locale)
+ translatable.message = localize(translatable, locale)
else:
return localize(translatable, locale)
return translatable
except ImportError as ex:
LOG.error(_LE("Failed to import required modules. Ensure that the "
"python-openvswitch package is installed. Error: %s"),
- ex.message)
+ ex)
except Exception as ex:
LOG.exception(six.text_type(ex))
h_exc.BrokenInterface, h_exc.DvaCreationFailed,
h_exc.DvaCreationPending, h_exc.BrokenDva,
h_exc.ConfigurationFailed) as ex:
- LOG.warning(p_con.error_map[type(ex)], ex.message)
+ LOG.warning(p_con.error_map[type(ex)], ex)
transient_state = p_con.Status.ERROR
except h_exc.DvaDeleteFailed as ex:
- LOG.warning(p_con.error_map[type(ex)], ex.message)
+ LOG.warning(p_con.error_map[type(ex)], ex)
transient_state = p_con.Status.DELETED
finally:
# if the returned transient state is None, no operations
neutron_router["admin_state_up"],
ip_allocation_info)
except h_exc.PreliminaryOperationsFailed as ex:
- raise h_exc.BrokenInterface(err_msg=ex.message)
+ raise h_exc.BrokenInterface(err_msg=str(ex))
state = api.extract_dva_state(dva)
return state
neutron_router["admin_state_up"],
ip_allocation_info)
except h_exc.PreliminaryOperationsFailed as ex:
- raise h_exc.BrokenInterface(err_msg=ex.message)
+ raise h_exc.BrokenInterface(err_msg=str(ex))
state = api.extract_dva_state(dva)
return state
return (p_con.Status.ACTIVE if neutron_router["admin_state_up"] else
p_con.Status.READY)
except h_exc.PreliminaryOperationsFailed as ex:
- raise h_exc.BrokenInterface(err_msg=ex.message)
+ raise h_exc.BrokenInterface(err_msg=str(ex))
state = api.extract_dva_state(dva)
return state
except RuntimeError as e:
# If the namespace didn't exist when delete was attempted, mission
# accomplished. Otherwise, re-raise the exception
- if 'No such file or directory' not in e.message:
+ if 'No such file or directory' not in str(e):
raise e
def _namespace_exists(self, namespace):
return getattr(user, cred_arg) != getattr(alt_user, cred_arg)
except exceptions.InvalidCredentials as ic:
msg = "At least one of the configured credentials is " \
- "not valid: %s" % ic.message
+ "not valid: %s" % ic
raise exceptions.InvalidConfiguration(msg)
else:
# TODO(andreaf) Add a uniqueness check here
'dev', self.parent.name,
'table', self.table))
- def test_del_gateway(self):
+ def test_del_gateway_success(self):
self.route_cmd.delete_gateway(self.gateway, table=self.table)
self._assert_sudo([self.ip_version],
('del', 'default',
'dev', self.parent.name,
'table', self.table))
+ def test_del_gateway_cannot_find_device(self):
+ self.parent._as_root.side_effect = RuntimeError("Cannot find device")
+
+ exc = self.assertRaises(exceptions.DeviceNotFoundError,
+ self.route_cmd.delete_gateway,
+ self.gateway, table=self.table)
+ self.assertIn(self.parent.name, str(exc))
+
+ def test_del_gateway_other_error(self):
+ self.parent._as_root.side_effect = RuntimeError()
+
+ self.assertRaises(RuntimeError, self.route_cmd.delete_gateway,
+ self.gateway, table=self.table)
+
def test_get_gateway(self):
for test_case in self.test_cases:
self.parent._run = mock.Mock(return_value=test_case['sample'])
# Shouldn't be reached
self.assertTrue(False)
except webtest.AppError as e:
- self.assertIn('501', e.message)
+ self.assertIn('501', str(e))
def test_resource_can_be_added_as_extension(self):
res_ext = extensions.ResourceExtension(
"DummyServicePlugin"])
cfg.CONF.set_override("core_plugin", DB_PLUGIN_KLASS)
e = self.assertRaises(ValueError, manager.NeutronManager.get_instance)
- self.assertIn(constants.DUMMY, e.message)
+ self.assertIn(constants.DUMMY, str(e))
def test_multiple_plugins_by_name_specified_for_service_type(self):
cfg.CONF.set_override("service_plugins", ["dummy", "dummy"])
"neutron.tests.unit.test_manager."
"MultiServiceCorePlugin")
e = self.assertRaises(ValueError, manager.NeutronManager.get_instance)
- self.assertIn(constants.DUMMY, e.message)
+ self.assertIn(constants.DUMMY, str(e))
def test_core_plugin_supports_services(self):
cfg.CONF.set_override("core_plugin",