openstack
/
trove
Code Issues Proposed changes

Replace assertRaisesRegexp with assertRaisesRegex 

This replaces the deprecated (in python 3.2) unittest.TestCase
method assertRaisesRegexp() with assertRaisesRegex()

Added hacking rule 'assert_raises_regexp'

Change-Id: Ie427bfd5af8474119a9e20cc025cd28abfa42cc2
This commit is contained in:
Valencia Serrao 2017-11-22 15:49:27 +05:30
parent ccb6752f69
commit e2afdadb68
14 changed files with 170 additions and 114 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
trove/hacking/checks.py
View File

@ -115,8 +115,23 @@ def check_no_basestring(logical_line):
yield(0, msg)
asse_raises_regexp = re.compile(r"assertRaisesRegexp\(")
def assert_raises_regexp(logical_line):
"""Check for usage of deprecated assertRaisesRegexp
N335
"""
res = asse_raises_regexp.search(logical_line)
if res:
yield (0, "N335: assertRaisesRegex must be used instead "
"of assertRaisesRegexp")
def factory(register):
register(validate_log_translations)
register(no_translate_debug_logs)
register(check_raised_localized_exceptions)
register(check_no_basestring)
register(assert_raises_regexp)

16
trove/tests/unittests/cluster/test_cluster_controller.py
View File

@ -418,10 +418,10 @@ class TestClusterControllerWithStrategy(trove_testtools.TestCase):
cluster.datastore_version.manager = 'test_dsv'
mock_cluster_load.return_value = cluster
self.assertRaisesRegexp(exception.TroveError,
'should have exactly one action specified',
self.controller.action, req,
body, tenant_id, cluster_id)
self.assertRaisesRegex(exception.TroveError,
'should have exactly one action specified',
self.controller.action, req,
body, tenant_id, cluster_id)
@patch.object(models.Cluster, 'load')
def test_controller_action_no_strategy(self,
@ -442,10 +442,10 @@ class TestClusterControllerWithStrategy(trove_testtools.TestCase):
datastore_version='test_dsv')
mock_cluster_load.return_value = cluster
self.assertRaisesRegexp(exception.TroveError,
'Action do_stuff2 not supported',
self.controller.action, req,
body, tenant_id, cluster_id)
self.assertRaisesRegex(exception.TroveError,
'Action do_stuff2 not supported',
self.controller.action, req,
body, tenant_id, cluster_id)
@patch.object(strategy, 'load_api_strategy')
@patch.object(models.Cluster, 'load')

24
trove/tests/unittests/cluster/test_cluster_redis_controller.py
View File

@ -124,12 +124,12 @@ class TestClusterController(trove_testtools.TestCase):
datastore_version.manager = 'mysql'
mock_get_datastore_version.return_value = (Mock(), datastore_version)
self.assertRaisesRegexp(exception.ClusterDatastoreNotSupported,
"Clusters not supported for",
self.controller.create,
req,
body,
tenant_id)
self.assertRaisesRegex(exception.ClusterDatastoreNotSupported,
"Clusters not supported for",
self.controller.create,
req,
body,
tenant_id)
@patch.object(Cluster, 'create')
@patch.object(utils, 'get_id_from_href')
@ -307,12 +307,12 @@ class TestClusterControllerWithStrategy(trove_testtools.TestCase):
datastore_version.manager = 'redis'
mock_get_datastore_version.return_value = (Mock(), datastore_version)
self.assertRaisesRegexp(exception.TroveError,
"Clusters not supported for",
self.controller.create,
req,
body,
tenant_id)
self.assertRaisesRegex(exception.TroveError,
"Clusters not supported for",
self.controller.create,
req,
body,
tenant_id)
@patch.object(views.ClusterView, 'data', return_value={})
@patch.object(datastore_models, 'get_datastore_version')

32
trove/tests/unittests/common/test_notification.py
View File

@ -305,30 +305,30 @@ class TestDBaaSNotification(trove_testtools.TestCase):
self.test_n = DBaaSTestNotification(Mock(), request=Mock())
def test_missing_required_start_traits(self):
self.assertRaisesRegexp(exception.TroveError,
self.test_n.required_start_traits()[0],
self.test_n.notify_start)
self.assertRaisesRegex(exception.TroveError,
self.test_n.required_start_traits()[0],
self.test_n.notify_start)
def test_invalid_start_traits(self):
self.assertRaisesRegexp(exception.TroveError,
"The following required keys",
self.test_n.notify_start, foo='bar')
self.assertRaisesRegex(exception.TroveError,
"The following required keys",
self.test_n.notify_start, foo='bar')
def test_missing_required_end_traits(self):
self.assertRaisesRegexp(exception.TroveError,
self.test_n.required_end_traits()[0],
self.test_n.notify_end)
self.assertRaisesRegex(exception.TroveError,
self.test_n.required_end_traits()[0],
self.test_n.notify_end)
def test_invalid_end_traits(self):
self.assertRaisesRegexp(exception.TroveError,
"The following required keys",
self.test_n.notify_end, foo='bar')
self.assertRaisesRegex(exception.TroveError,
"The following required keys",
self.test_n.notify_end, foo='bar')
def test_missing_required_error_traits(self):
self.assertRaisesRegexp(exception.TroveError,
self.test_n.required_error_traits()[0],
self.test_n._notify, 'error',
self.test_n.required_error_traits(), [])
self.assertRaisesRegex(exception.TroveError,
self.test_n.required_error_traits()[0],
self.test_n._notify, 'error',
self.test_n.required_error_traits(), [])
@patch.object(rpc, 'get_notifier')
def test_start_event(self, notifier):

14
trove/tests/unittests/datastore/test_datastore.py
View File

@ -36,13 +36,13 @@ class TestDatastore(TestDatastoreBase):
@patch.object(datastore_models, 'CONF')
def test_create_failure_with_datastore_default(self, mock_conf):
mock_conf.default_datastore = 'bad_ds'
self.assertRaisesRegexp(exception.DatastoreDefaultDatastoreNotFound,
"Default datastore 'bad_ds' cannot be found",
datastore_models.get_datastore_version)
self.assertRaisesRegexp(exception.DatastoreNotFound,
"Datastore 'my_ds' cannot be found",
datastore_models.get_datastore_version,
'my_ds')
self.assertRaisesRegex(exception.DatastoreDefaultDatastoreNotFound,
"Default datastore 'bad_ds' cannot be found",
datastore_models.get_datastore_version)
self.assertRaisesRegex(exception.DatastoreNotFound,
"Datastore 'my_ds' cannot be found",
datastore_models.get_datastore_version,
'my_ds')
def test_get_datastore_or_version(self):
# datastore, datastore_version, valid, exception

4
trove/tests/unittests/datastore/test_datastore_version_metadata.py
View File

@ -57,7 +57,7 @@ class TestDatastoreVersionMetadata(TestDatastoreBase):
def test_add_existing_flavor_associations(self):
dsmetadata = datastore_models.DatastoreVersionMetadata
self.assertRaisesRegexp(
self.assertRaisesRegex(
exception.DatastoreFlavorAssociationAlreadyExists,
"Flavor %s is already associated with datastore %s version %s"
% (self.flavor_id, self.ds_name, self.ds_version),
@ -73,7 +73,7 @@ class TestDatastoreVersionMetadata(TestDatastoreBase):
def test_delete_nonexistent_flavor_mapping(self):
dsmeta = datastore_models.DatastoreVersionMetadata
self.assertRaisesRegexp(
self.assertRaisesRegex(
exception.DatastoreFlavorAssociationNotFound,
"Flavor 2 is not supported for datastore %s version %s"
% (self.ds_name, self.ds_version),

58
trove/tests/unittests/guestagent/test_dbaas.py
View File

@ -214,8 +214,8 @@ class DbaasTest(trove_testtools.TestCase):
@patch.object(operating_system, 'read_file',
side_effect=RuntimeError('read_file error'))
def test_get_auth_password_error(self, _, get_cnf_mock):
self.assertRaisesRegexp(RuntimeError, "read_file error",
MySqlApp.get_auth_password)
self.assertRaisesRegex(RuntimeError, "read_file error",
MySqlApp.get_auth_password)
def test_service_discovery(self):
with patch.object(os.path, 'isfile', return_value=True):
@ -713,8 +713,8 @@ class MySqlAdminTest(trove_testtools.TestCase):
def test_fail_get_user(self, *args):
username = "os_admin"
hostname = "host"
self.assertRaisesRegexp(BadRequest, "Username os_admin is not valid",
self.mySqlAdmin.get_user, username, hostname)
self.assertRaisesRegex(BadRequest, "Username os_admin is not valid",
self.mySqlAdmin.get_user, username, hostname)
def test_grant_access(self):
user = MagicMock()
@ -926,8 +926,8 @@ class MySqlAppTest(trove_testtools.TestCase):
mock_logging):
with patch.object(BaseDbStatus, 'prepare_completed') as patch_pc:
patch_pc.__get__ = Mock(return_value=True)
self.assertRaisesRegexp(RuntimeError, 'Service is not discovered.',
self.mySqlApp.stop_db)
self.assertRaisesRegex(RuntimeError, 'Service is not discovered.',
self.mySqlApp.stop_db)
self.assertEqual(0, mock_execute.call_count)
def test_restart_is_successful(self):
@ -1097,8 +1097,8 @@ class MySqlAppTest(trove_testtools.TestCase):
@patch.object(utils, 'execute_with_timeout', return_value=('0', ''))
def test_fail__enable_mysql_on_boot(self, mock_execute, mock_service,
mock_logging):
self.assertRaisesRegexp(RuntimeError, 'Service is not discovered.',
self.mySqlApp._enable_mysql_on_boot)
self.assertRaisesRegex(RuntimeError, 'Service is not discovered.',
self.mySqlApp._enable_mysql_on_boot)
self.assertEqual(0, mock_execute.call_count)
@patch.object(utils, 'execute_with_timeout', return_value=('0', ''))
@ -1116,8 +1116,8 @@ class MySqlAppTest(trove_testtools.TestCase):
@patch.object(utils, 'execute_with_timeout', return_value=('0', ''))
def test_fail__disable_mysql_on_boot(self, mock_execute, mock_service,
mock_logging):
self.assertRaisesRegexp(RuntimeError, 'Service is not discovered.',
self.mySqlApp._disable_mysql_on_boot)
self.assertRaisesRegex(RuntimeError, 'Service is not discovered.',
self.mySqlApp._disable_mysql_on_boot)
self.assertEqual(0, mock_execute.call_count)
def test_update_overrides(self):
@ -1298,10 +1298,10 @@ class MySqlAppTest(trove_testtools.TestCase):
return_value=MagicMock(name='get_engine'))
@patch.object(utils, 'poll_until', side_effect=PollTimeOut)
def test_fail__wait_for_slave_status(self, *args):
self.assertRaisesRegexp(RuntimeError,
"Replication is not on after 5 seconds.",
self.mySqlApp._wait_for_slave_status, 'ON',
Mock(), 5)
self.assertRaisesRegex(RuntimeError,
"Replication is not on after 5 seconds.",
self.mySqlApp._wait_for_slave_status, 'ON',
Mock(), 5)
@patch.object(dbaas, 'get_engine',
return_value=MagicMock(name='get_engine'))
@ -2107,7 +2107,7 @@ class BaseDbStatusTest(trove_testtools.TestCase):
return_value=False) as service_call:
with patch.multiple(operating_system, start_service=DEFAULT,
enable_service_on_boot=DEFAULT) as os_cmd:
self.assertRaisesRegexp(
self.assertRaisesRegex(
RuntimeError, "Database failed to start.",
status.start_db_service,
service_candidates, 10, enable_on_boot=True)
@ -2156,7 +2156,7 @@ class BaseDbStatusTest(trove_testtools.TestCase):
return_value=False) as service_call:
with patch.multiple(operating_system, stop_service=DEFAULT,
disable_service_on_boot=DEFAULT) as os_cmd:
self.assertRaisesRegexp(
self.assertRaisesRegex(
RuntimeError, "Database failed to stop.",
status.stop_db_service,
service_candidates, 10, disable_on_boot=True)
@ -2189,7 +2189,7 @@ class BaseDbStatusTest(trove_testtools.TestCase):
side_effect=Exception("Error in database start.")),
stop_db_service=DEFAULT, begin_restart=DEFAULT,
end_restart=DEFAULT):
self.assertRaisesRegexp(
self.assertRaisesRegex(
RuntimeError, "Database restart failed.",
status.restart_db_service, service_candidates, 10)
status.begin_restart.assert_called_once_with()
@ -2865,9 +2865,9 @@ class VerticaAppTest(trove_testtools.TestCase):
def test_failure_install_vertica(self, *args):
with patch.object(vertica_system, 'shell_execute',
side_effect=ProcessExecutionError('some exception')):
self.assertRaisesRegexp(RuntimeError, 'install_vertica failed.',
self.app.install_vertica,
members='10.0.0.2')
self.assertRaisesRegex(RuntimeError, 'install_vertica failed.',
self.app.install_vertica,
members='10.0.0.2')
def test_create_db(self):
with patch.object(self.app, 'read_config',
@ -2884,9 +2884,9 @@ class VerticaAppTest(trove_testtools.TestCase):
def test_failure_create_db(self, *args):
with patch.object(self.app, 'read_config',
side_effect=RuntimeError('Error')):
self.assertRaisesRegexp(RuntimeError,
'Vertica database create failed.',
self.app.create_db)
self.assertRaisesRegex(RuntimeError,
'Vertica database create failed.',
self.app.create_db)
# Because of an exception in read_config there was no shell execution.
self.assertEqual(0, vertica_system.shell_execute.call_count)
@ -3165,9 +3165,9 @@ class VerticaAppTest(trove_testtools.TestCase):
@patch('trove.guestagent.datastore.experimental.vertica.service.LOG')
def test_failure__enable_db_on_boot(self, *args):
with patch.object(subprocess, 'Popen', side_effect=OSError):
self.assertRaisesRegexp(RuntimeError,
'Could not enable database on boot.',
self.app._enable_db_on_boot)
self.assertRaisesRegex(RuntimeError,
'Could not enable database on boot.',
self.app._enable_db_on_boot)
def test__disable_db_on_boot(self):
self.app._disable_db_on_boot()
@ -3187,9 +3187,9 @@ class VerticaAppTest(trove_testtools.TestCase):
def test_failure__disable_db_on_boot(self, *args):
with patch.object(vertica_system, 'shell_execute',
side_effect=ProcessExecutionError('Error')):
self.assertRaisesRegexp(RuntimeError,
'Could not disable database on boot.',
self.app._disable_db_on_boot)
self.assertRaisesRegex(RuntimeError,
'Could not disable database on boot.',
self.app._disable_db_on_boot)
def test_read_config(self):
with patch.object(configparser, 'ConfigParser',

38
trove/tests/unittests/guestagent/test_manager.py
View File

@ -141,28 +141,28 @@ class ManagerTest(trove_testtools.TestCase):
return "".join(sorted(value))
def test_guest_log_action_enable_disable(self):
self.assertRaisesRegexp(exception.BadRequest,
"Cannot enable and disable",
self.manager.guest_log_action,
self.context,
self.log_name_sys,
True, True, False, False)
self.assertRaisesRegex(exception.BadRequest,
"Cannot enable and disable",
self.manager.guest_log_action,
self.context,
self.log_name_sys,
True, True, False, False)
def test_guest_log_action_enable_sys(self):
self.assertRaisesRegexp(exception.BadRequest,
"Cannot enable a SYSTEM log",
self.manager.guest_log_action,
self.context,
self.log_name_sys,
True, False, False, False)
self.assertRaisesRegex(exception.BadRequest,
"Cannot enable a SYSTEM log",
self.manager.guest_log_action,
self.context,
self.log_name_sys,
True, False, False, False)
def test_guest_log_action_disable_sys(self):
self.assertRaisesRegexp(exception.BadRequest,
"Cannot disable a SYSTEM log",
self.manager.guest_log_action,
self.context,
self.log_name_sys,
False, True, False, False)
self.assertRaisesRegex(exception.BadRequest,
"Cannot disable a SYSTEM log",
self.manager.guest_log_action,
self.context,
self.log_name_sys,
False, True, False, False)
def test_guest_log_action_publish_sys(self):
with patch.object(os.path, 'isfile', return_value=True):
@ -472,7 +472,7 @@ class ManagerTest(trove_testtools.TestCase):
side_effect=expected_failure
)):
expected_msg = encodeutils.exception_to_unicode(expected_failure)
self.assertRaisesRegexp(
self.assertRaisesRegex(
Exception, expected_msg,
self.manager.prepare,
self.context, packages, databases, memory_mb, users,

41
trove/tests/unittests/hacking/test_check.py
View File

@ -10,6 +10,10 @@
# License for the specific language governing permissions and limitations
# under the License.
import mock
import pep8
import textwrap
from trove.hacking import checks as tc
from trove.tests.unittests import trove_testtools
@ -95,3 +99,40 @@ class HackingTestCase(trove_testtools.TestCase):
self.assertEqual(
0,
len(list(tc.check_no_basestring("this basestring is good)"))))
# We are patching pep8 so that only the check under test is actually
# installed.
@mock.patch('pep8._checks',
{'physical_line': {}, 'logical_line': {}, 'tree': {}})
def _run_check(self, code, checker, filename=None):
pep8.register_check(checker)
lines = textwrap.dedent(code).strip().splitlines(True)
checker = pep8.Checker(filename=filename, lines=lines)
# NOTE(sdague): the standard reporter has printing to stdout
# as a normal part of check_all, which bleeds through to the
# test output stream in an unhelpful way. This blocks that printing.
with mock.patch('pep8.StandardReport.get_file_results'):
checker.check_all()
checker.report._deferred_print.sort()
return checker.report._deferred_print
def _assert_has_errors(self, code, checker, expected_errors=None,
filename=None):
actual_errors = [e[:3] for e in
self._run_check(code, checker, filename)]
self.assertEqual(expected_errors or [], actual_errors)
def _assert_has_no_errors(self, code, checker, filename=None):
self._assert_has_errors(code, checker, filename=filename)
def test_oslo_assert_raises_regexp(self):
code = """
self.assertRaisesRegexp(ValueError,
"invalid literal for.*XYZ'$",
int,
'XYZ')
"""
self._assert_has_errors(code, tc.assert_raises_regexp,
expected_errors=[(1, 0, "N335")])

4
trove/tests/unittests/mgmt/test_clusters.py
View File

@ -83,7 +83,7 @@ class TestClusterController(trove_testtools.TestCase):
tenant_id = Mock()
id = Mock()
self.assertRaisesRegexp(
self.assertRaisesRegex(
exception.BadRequest, 'Invalid request body.',
self.controller.action, self.req, None, tenant_id, id)
@ -94,6 +94,6 @@ class TestClusterController(trove_testtools.TestCase):
id = Mock()
mock_cluster_load.return_value = self.mock_clusters[0]
self.assertRaisesRegexp(
self.assertRaisesRegex(
exception.BadRequest, 'Invalid cluster action requested.',
self.controller.action, self.req, body, tenant_id, id)

8
trove/tests/unittests/mgmt/test_datastores.py
View File

@ -74,7 +74,7 @@ class TestDatastoreVersion(trove_testtools.TestCase):
"packages": "test-pkg",
"active": True,
"default": True}}
self.assertRaisesRegexp(
self.assertRaisesRegex(
exception.DatastoreVersionAlreadyExists,
"A datastore version with the name 'test_new_vr' already exists",
self.version_controller.create, self.req, body, self.tenant_id)
@ -91,7 +91,7 @@ class TestDatastoreVersion(trove_testtools.TestCase):
"packages": "test-pkg",
"active": True,
"default": True}}
self.assertRaisesRegexp(
self.assertRaisesRegex(
exception.ImageNotFound,
"Image image-id cannot be found.",
self.version_controller.create, self.req, body, self.tenant_id)
@ -108,7 +108,7 @@ class TestDatastoreVersion(trove_testtools.TestCase):
self.assertEqual(202, output.status)
# Try to find deleted version, this should raise exception.
self.assertRaisesRegexp(
self.assertRaisesRegex(
exception.DatastoreVersionNotFound,
err_msg, models.DatastoreVersion.load_by_uuid, ds_version1.id)
@ -131,7 +131,7 @@ class TestDatastoreVersion(trove_testtools.TestCase):
side_effect=glance_exceptions.HTTPNotFound())
body = {"image": "non-existent-image-id"}
self.assertRaisesRegexp(
self.assertRaisesRegex(
exception.ImageNotFound,
"Image non-existent-image-id cannot be found.",
self.version_controller.edit, self.req, body,

6
trove/tests/unittests/taskmanager/test_api.py
View File

@ -118,9 +118,9 @@ class ApiTest(trove_testtools.TestCase):
def test_transform_obj(self):
flavor = Mock()
self.assertRaisesRegexp(ValueError,
('Could not transform %s' % flavor),
self.api._transform_obj, flavor)
self.assertRaisesRegex(ValueError,
('Could not transform %s' % flavor),
self.api._transform_obj, flavor)
def test_upgrade(self):
self.api.upgrade('some-instance-id', 'some-datastore-version')

18
trove/tests/unittests/taskmanager/test_manager.py
View File

@ -63,8 +63,8 @@ class TestManager(trove_testtools.TestCase):
result = self.manager._most_current_replica(master, None)
assert_equal(result, selected_master)
with self.assertRaisesRegexp(TroveError,
'not all replicating from same'):
with self.assertRaisesRegex(TroveError,
'not all replicating from same'):
test_case([['a', '2a99e-32bf', 2], ['b', '2a', 1]], None)
test_case([['a', '2a99e-32bf', 2]], 'a')
@ -158,9 +158,9 @@ class TestManager(trove_testtools.TestCase):
with patch.object(models.BuiltInstanceTasks, 'load',
side_effect=[self.mock_slave1, self.mock_old_master,
self.mock_slave2]):
self.assertRaisesRegexp(RuntimeError, 'Error',
self.manager.promote_to_replica_source,
self.context, 'some-inst-id')
self.assertRaisesRegex(RuntimeError, 'Error',
self.manager.promote_to_replica_source,
self.context, 'some-inst-id')
@patch('trove.taskmanager.manager.LOG')
def test_error_demote_replication_master_promote_to_replica_source(
@ -185,9 +185,9 @@ class TestManager(trove_testtools.TestCase):
with patch.object(models.BuiltInstanceTasks, 'load',
side_effect=[self.mock_master, self.mock_slave1,
self.mock_slave2]):
self.assertRaisesRegexp(RuntimeError, 'Error',
self.manager.eject_replica_source,
self.context, 'some-inst-id')
self.assertRaisesRegex(RuntimeError, 'Error',
self.manager.eject_replica_source,
self.context, 'some-inst-id')
@patch.object(Backup, 'delete')
@patch.object(models.BuiltInstanceTasks, 'load')
@ -225,7 +225,7 @@ class TestManager(trove_testtools.TestCase):
Mock(), 'some-master-id', None, None, None, None)
def test_AttributeError_create_instance(self):
self.assertRaisesRegexp(
self.assertRaisesRegex(
AttributeError, 'Cannot create multiple non-replica instances.',
self.manager.create_instance, self.context, ['id1', 'id2'],
Mock(), Mock(), Mock(), None, None, 'mysql', 'mysql-server', 2,

6
trove/tests/unittests/taskmanager/test_models.py
View File

@ -324,7 +324,7 @@ class FreshInstanceTasksTest(BaseFreshInstanceTasksTest):
@patch('trove.taskmanager.models.LOG')
def test_error_create_dns_entry_create_instance(self, *args):
mock_flavor = {'id': 6, 'ram': 512, 'name': 'big_flavor'}
self.assertRaisesRegexp(
self.assertRaisesRegex(
TroveError,
'Error creating DNS entry for instance',
self.freshinstancetasks.create_instance, mock_flavor,
@ -406,7 +406,7 @@ class FreshInstanceTasksTest(BaseFreshInstanceTasksTest):
mock_flavor = {'id': 8, 'ram': 768, 'name': 'bigger_flavor'}
snapshot = {'replication_strategy': 'MysqlGTIDReplication',
'master': {'id': 'master-id'}}
self.assertRaisesRegexp(
self.assertRaisesRegex(
TroveError, 'Error attaching instance',
self.freshinstancetasks.attach_replication_slave,
snapshot, mock_flavor)
@ -475,7 +475,7 @@ class InstanceSecurityGroupRuleTests(BaseFreshInstanceTasksTest):
mock_conf.get = Mock(
return_value=FakeOptGroup(tcp_ports=['3306', '-3306']))
mock_flavor = {'id': 7, 'ram': 256, 'name': 'smaller_flavor'}
self.assertRaisesRegexp(
self.assertRaisesRegex(
TroveError,
'Error creating security group for instance',
self.freshinstancetasks.create_instance, mock_flavor,