openstack/taskflow
Code Issues Proposed changes

Replace assertRaisesRegexp with assertRaisesRegex

Browse Source 
assertRaisesRegexp was renamed to assertRaisesRegex in Py3.2
For more details, please check:
https://docs.python.org/3/library/
unittest.html#unittest.TestCase.assertRaisesRegex

Change-Id: I89cce19e80b04074aab9f49a76c7652acace78b3
Closes-Bug: #1436957
This commit is contained in:
Vu Cong Tuan 2017-06-03 12:28:52 +07:00
parent cf0ee0b0c7
commit db61fad6b2
13 changed files with 102 additions and 102 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
taskflow/test.py
View File

@ -122,7 +122,7 @@ class TestCase(base.BaseTestCase):
self.assertRaises(exc_class, access_func)
def assertRaisesRegexp(self, exc_class, pattern, callable_obj,
def assertRaisesRegex(self, exc_class, pattern, callable_obj,
*args, **kwargs):
# TODO(harlowja): submit a pull/review request to testtools to add
# this method to there codebase instead of having it exist in ours

6
taskflow/tests/unit/action_engine/test_compile.py
View File

@ -54,7 +54,7 @@ class PatternCompileTest(test.TestCase):
def test_wrong_object(self):
msg_regex = '^Unknown object .* requested to compile'
self.assertRaisesRegexp(TypeError, msg_regex,
self.assertRaisesRegex(TypeError, msg_regex,
compiler.PatternCompiler(42).compile)
def test_empty(self):
@ -458,7 +458,7 @@ class PatternCompileTest(test.TestCase):
test_utils.DummyTask(name="a")
)
e = engines.load(flo)
self.assertRaisesRegexp(exc.Duplicate,
self.assertRaisesRegex(exc.Duplicate,
'^Atoms with duplicate names',
e.compile)
@ -467,7 +467,7 @@ class PatternCompileTest(test.TestCase):
gf.Flow("int1").add(test_utils.DummyTask(name="a")),
gf.Flow("int2").add(test_utils.DummyTask(name="a")))
e = engines.load(flo)
self.assertRaisesRegexp(exc.Duplicate,
self.assertRaisesRegex(exc.Duplicate,
'^Atoms with duplicate names',
e.compile)

6
taskflow/tests/unit/patterns/test_graph_flow.py
View File

@ -200,14 +200,14 @@ class GraphFlowTest(test.TestCase):
task1 = _task('task1')
task2 = _task('task2')
f = gf.Flow('test').add(task2)
self.assertRaisesRegexp(ValueError, 'Node .* not found to link from',
self.assertRaisesRegex(ValueError, 'Node .* not found to link from',
f.link, task1, task2)
def test_graph_flow_link_to_unknown_node(self):
task1 = _task('task1')
task2 = _task('task2')
f = gf.Flow('test').add(task1)
self.assertRaisesRegexp(ValueError, 'Node .* not found to link to',
self.assertRaisesRegex(ValueError, 'Node .* not found to link to',
f.link, task1, task2)
def test_graph_flow_link_raises_on_cycle(self):
@ -288,7 +288,7 @@ class TargetedGraphFlowTest(test.TestCase):
task1 = _task('task1', provides=['a'], requires=[])
task2 = _task('task2', provides=['b'], requires=['a'])
f.add(task1)
self.assertRaisesRegexp(ValueError, '^Node .* not found',
self.assertRaisesRegex(ValueError, '^Node .* not found',
f.set_target, task2)
def test_targeted_flow_one_node(self):

2
taskflow/tests/unit/test_check_transition.py
View File

@ -33,7 +33,7 @@ class TransitionTest(test.TestCase):
self.assertFalse(self.check_transition(from_state, to_state), msg=msg)
def assertTransitionForbidden(self, from_state, to_state):
self.assertRaisesRegexp(exc.InvalidState,
self.assertRaisesRegex(exc.InvalidState,
self.transition_exc_regexp,
self.check_transition, from_state, to_state)

8
taskflow/tests/unit/test_engine_helpers.py
View File

@ -54,14 +54,14 @@ class FlowFromDetailTestCase(test.TestCase):
def test_no_meta(self):
_lb, flow_detail = p_utils.temporary_flow_detail()
self.assertEqual({}, flow_detail.meta)
self.assertRaisesRegexp(ValueError,
self.assertRaisesRegex(ValueError,
'^Cannot .* no factory information saved.$',
taskflow.engines.flow_from_detail,
flow_detail)
def test_no_factory_in_meta(self):
_lb, flow_detail = p_utils.temporary_flow_detail()
self.assertRaisesRegexp(ValueError,
self.assertRaisesRegex(ValueError,
'^Cannot .* no factory information saved.$',
taskflow.engines.flow_from_detail,
flow_detail)
@ -71,7 +71,7 @@ class FlowFromDetailTestCase(test.TestCase):
flow_detail.meta = dict(factory=dict(
name='you can not import me, i contain spaces'
))
self.assertRaisesRegexp(ImportError,
self.assertRaisesRegex(ImportError,
'^Could not import factory',
taskflow.engines.flow_from_detail,
flow_detail)
@ -110,7 +110,7 @@ class LoadFromFactoryTestCase(test.TestCase):
def factory():
pass
self.assertRaisesRegexp(ValueError,
self.assertRaisesRegex(ValueError,
'Flow factory .* is not reimportable',
taskflow.engines.load_from_factory,
factory)

2
taskflow/tests/unit/test_engines.py
View File

@ -1390,7 +1390,7 @@ class EngineCheckingTaskTest(utils.EngineTestBase):
utils.FailingTask('fail1')
)
engine = self._make_engine(flow)
self.assertRaisesRegexp(RuntimeError, '^Woot', engine.run)
self.assertRaisesRegex(RuntimeError, '^Woot', engine.run)
class SerialEngineTest(EngineTaskTest,

4
taskflow/tests/unit/test_failure.py
View File

@ -89,7 +89,7 @@ class CaptureFailureTestCase(test.TestCase, GeneralFailureObjTestsMixin):
self.assertIs(exc_info[1], self.fail_obj.exception)
def test_reraises(self):
self.assertRaisesRegexp(RuntimeError, '^Woot!$', self.fail_obj.reraise)
self.assertRaisesRegex(RuntimeError, '^Woot!$', self.fail_obj.reraise)
class ReCreatedFailureTestCase(test.TestCase, GeneralFailureObjTestsMixin):
@ -209,7 +209,7 @@ class FailureObjectTestCase(test.TestCase):
def test_reraises_one(self):
fls = [_captured_failure('Woot!')]
self.assertRaisesRegexp(RuntimeError, '^Woot!$',
self.assertRaisesRegex(RuntimeError, '^Woot!$',
failure.Failure.reraise_if_any, fls)
def test_reraises_several(self):

2
taskflow/tests/unit/test_flow_dependencies.py
View File

@ -233,7 +233,7 @@ class FlowDependenciesTest(test.TestCase):
def test_graph_cyclic_dependency(self):
flow = gf.Flow('g-3-cyclic')
self.assertRaisesRegexp(exceptions.DependencyFailure, '^No path',
self.assertRaisesRegex(exceptions.DependencyFailure, '^No path',
flow.add,
utils.TaskOneArgOneReturn(provides='a',
requires=['b']),

2
taskflow/tests/unit/test_functor_task.py
View File

@ -65,7 +65,7 @@ class FunctorTaskTest(test.TestCase):
t(bof.run_one, revert=bof.revert_one),
t(bof.run_fail)
)
self.assertRaisesRegexp(RuntimeError, '^Woot',
self.assertRaisesRegex(RuntimeError, '^Woot',
taskflow.engines.run, flow)
self.assertEqual(['one', 'fail', 'revert one'], values)

38
taskflow/tests/unit/test_retries.py
View File

@ -103,7 +103,7 @@ class RetryTest(utils.EngineTestBase):
engine = self._make_engine(flow)
engine.storage.inject({'y': 4})
with utils.CaptureListener(engine) as capturer:
self.assertRaisesRegexp(RuntimeError, '^Woot', engine.run)
self.assertRaisesRegex(RuntimeError, '^Woot', engine.run)
self.assertEqual({'y': 4}, engine.storage.fetch_all())
expected = ['flow-1.f RUNNING',
'r1.r RUNNING',
@ -142,7 +142,7 @@ class RetryTest(utils.EngineTestBase):
engine = self._make_engine(flow)
engine.storage.inject({'y': 4})
with utils.CaptureListener(engine) as capturer:
self.assertRaisesRegexp(RuntimeError, '^Gotcha', engine.run)
self.assertRaisesRegex(RuntimeError, '^Gotcha', engine.run)
self.assertEqual({'y': 4, 'x': 1}, engine.storage.fetch_all())
expected = ['flow-1.f RUNNING',
'r1.r RUNNING',
@ -468,7 +468,7 @@ class RetryTest(utils.EngineTestBase):
engine = self._make_engine(flow)
engine.storage.inject({'y': 2})
with utils.CaptureListener(engine) as capturer:
self.assertRaisesRegexp(RuntimeError, '^Woot', engine.run)
self.assertRaisesRegex(RuntimeError, '^Woot', engine.run)
self.assertEqual({'y': 2}, engine.storage.fetch_all())
expected = ['flow-1.f RUNNING',
'r1.r RUNNING',
@ -494,8 +494,8 @@ class RetryTest(utils.EngineTestBase):
flow = lf.Flow('test', retry=utils.OneReturnRetry(provides='x')).add(
utils.FailingTask('fail'))
engine = self._make_engine(flow)
self.assertRaisesRegexp(RuntimeError, '^Woot', engine.run)
self.assertRaisesRegexp(RuntimeError, '^Woot', engine.run)
self.assertRaisesRegex(RuntimeError, '^Woot', engine.run)
self.assertRaisesRegex(RuntimeError, '^Woot', engine.run)
def test_run_just_retry(self):
flow = utils.OneReturnRetry(provides='x')
@ -596,7 +596,7 @@ class RetryTest(utils.EngineTestBase):
utils.FailingTask('t2'))
engine = self._make_engine(flow)
with utils.CaptureListener(engine) as capturer:
self.assertRaisesRegexp(RuntimeError, '^Woot', engine.run)
self.assertRaisesRegex(RuntimeError, '^Woot', engine.run)
expected = ['flow-1.f RUNNING',
'r1.r RUNNING',
'r1.r SUCCESS(1)',
@ -645,7 +645,7 @@ class RetryTest(utils.EngineTestBase):
flow = lf.Flow('flow-1', retry1).add(utils.FailingTaskWithOneArg('t1'))
engine = self._make_engine(flow)
with utils.CaptureListener(engine) as capturer:
self.assertRaisesRegexp(RuntimeError, '^Woot', engine.run)
self.assertRaisesRegex(RuntimeError, '^Woot', engine.run)
expected = ['flow-1.f RUNNING',
'r1.r RUNNING',
'r1.r SUCCESS(3)',
@ -688,7 +688,7 @@ class RetryTest(utils.EngineTestBase):
flow = lf.Flow('flow-1', retry1).add(utils.FailingTaskWithOneArg('t1'))
engine = self._make_engine(flow)
with utils.CaptureListener(engine) as capturer:
self.assertRaisesRegexp(RuntimeError, '^Woot', engine.run)
self.assertRaisesRegex(RuntimeError, '^Woot', engine.run)
expected = ['flow-1.f RUNNING',
'r1.r RUNNING',
'r1.r SUCCESS(2)',
@ -728,7 +728,7 @@ class RetryTest(utils.EngineTestBase):
)
engine = self._make_engine(flow)
with utils.CaptureListener(engine) as capturer:
self.assertRaisesRegexp(RuntimeError, '^Woot', engine.run)
self.assertRaisesRegex(RuntimeError, '^Woot', engine.run)
expected = ['flow-1.f RUNNING',
'task1.t RUNNING',
'task1.t SUCCESS(5)',
@ -778,7 +778,7 @@ class RetryTest(utils.EngineTestBase):
)
engine = self._make_engine(flow)
with utils.CaptureListener(engine) as capturer:
self.assertRaisesRegexp(RuntimeError, '^Woot', engine.run)
self.assertRaisesRegex(RuntimeError, '^Woot', engine.run)
expected = ['flow-1.f RUNNING',
'task1.t RUNNING',
'task1.t SUCCESS(5)',
@ -825,7 +825,7 @@ class RetryTest(utils.EngineTestBase):
flow = lf.Flow('flow-1', retry1).add(utils.ConditionalTask('t1'))
engine = self._make_engine(flow)
engine.storage.inject({'y': 1})
self.assertRaisesRegexp(exc.NotFound, '^No elements left', engine.run)
self.assertRaisesRegex(exc.NotFound, '^No elements left', engine.run)
def test_parameterized_for_each_with_list(self):
values = [3, 2, 5]
@ -834,7 +834,7 @@ class RetryTest(utils.EngineTestBase):
engine = self._make_engine(flow)
engine.storage.inject({'values': values, 'y': 1})
with utils.CaptureListener(engine) as capturer:
self.assertRaisesRegexp(RuntimeError, '^Woot', engine.run)
self.assertRaisesRegex(RuntimeError, '^Woot', engine.run)
expected = ['flow-1.f RUNNING',
'r1.r RUNNING',
'r1.r SUCCESS(3)',
@ -870,7 +870,7 @@ class RetryTest(utils.EngineTestBase):
engine = self._make_engine(flow)
engine.storage.inject({'values': values, 'y': 1})
with utils.CaptureListener(engine) as capturer:
self.assertRaisesRegexp(RuntimeError, '^Woot', engine.run)
self.assertRaisesRegex(RuntimeError, '^Woot', engine.run)
expected = ['flow-1.f RUNNING',
'r1.r RUNNING',
'r1.r SUCCESS(3)',
@ -911,7 +911,7 @@ class RetryTest(utils.EngineTestBase):
engine = self._make_engine(flow)
engine.storage.inject({'values': values, 'y': 1})
with utils.CaptureListener(engine) as capturer:
self.assertRaisesRegexp(RuntimeError, '^Woot', engine.run)
self.assertRaisesRegex(RuntimeError, '^Woot', engine.run)
expected = ['flow-1.f RUNNING',
'task-1.t RUNNING',
'task-1.t SUCCESS(5)',
@ -955,7 +955,7 @@ class RetryTest(utils.EngineTestBase):
engine = self._make_engine(flow)
engine.storage.inject({'values': values, 'y': 1})
with utils.CaptureListener(engine) as capturer:
self.assertRaisesRegexp(RuntimeError, '^Woot', engine.run)
self.assertRaisesRegex(RuntimeError, '^Woot', engine.run)
expected = ['flow-1.f RUNNING',
'task-1.t RUNNING',
'task-1.t SUCCESS(5)',
@ -994,7 +994,7 @@ class RetryTest(utils.EngineTestBase):
flow = lf.Flow('flow-1', retry1).add(utils.ConditionalTask('t1'))
engine = self._make_engine(flow)
engine.storage.inject({'values': values, 'y': 1})
self.assertRaisesRegexp(exc.NotFound, '^No elements left', engine.run)
self.assertRaisesRegex(exc.NotFound, '^No elements left', engine.run)
def _pretend_to_run_a_flow_and_crash(self, when):
flow = uf.Flow('flow-1', retry.Times(3, provides='x')).add(
@ -1109,7 +1109,7 @@ class RetryTest(utils.EngineTestBase):
r = FailingRetry()
flow = lf.Flow('testflow', r)
engine = self._make_engine(flow)
self.assertRaisesRegexp(ValueError, '^OMG', engine.run)
self.assertRaisesRegex(ValueError, '^OMG', engine.run)
self.assertEqual(1, len(engine.storage.get_retry_histories()))
self.assertEqual(0, len(r.history))
self.assertEqual([], list(r.history.outcomes_iter()))
@ -1120,7 +1120,7 @@ class RetryTest(utils.EngineTestBase):
r = NastyFailingRetry()
flow = lf.Flow('testflow', r)
engine = self._make_engine(flow)
self.assertRaisesRegexp(ValueError, '^WOOT', engine.run)
self.assertRaisesRegex(ValueError, '^WOOT', engine.run)
def test_nested_provides_graph_reverts_correctly(self):
flow = gf.Flow("test").add(
@ -1135,7 +1135,7 @@ class RetryTest(utils.EngineTestBase):
engine.storage.save('b', 11)
engine.storage.save('a', 10)
with utils.CaptureListener(engine, capture_flow=False) as capturer:
self.assertRaisesRegexp(RuntimeError, '^Woot', engine.run)
self.assertRaisesRegex(RuntimeError, '^Woot', engine.run)
expected = ['c.t RUNNING',
'c.t FAILURE(Failure: RuntimeError: Woot!)',
'a.t REVERTING',

16
taskflow/tests/unit/test_storage.py
View File

@ -191,7 +191,7 @@ class StorageTestMixin(object):
def test_fetch_unknown_name(self):
s = self._get_storage()
self.assertRaisesRegexp(exceptions.NotFound,
self.assertRaisesRegex(exceptions.NotFound,
"^Name 'xxx' is not mapped",
s.fetch, 'xxx')
@ -375,7 +375,7 @@ class StorageTestMixin(object):
def test_get_state_of_unknown_task(self):
s = self._get_storage()
self.assertRaisesRegexp(exceptions.NotFound, '^Unknown',
self.assertRaisesRegex(exceptions.NotFound, '^Unknown',
s.get_atom_state, 'my task')
def test_task_by_name(self):
@ -414,7 +414,7 @@ class StorageTestMixin(object):
def test_unknown_task_by_name(self):
s = self._get_storage()
self.assertRaisesRegexp(exceptions.NotFound,
self.assertRaisesRegex(exceptions.NotFound,
'^Unknown atom',
s.get_atom_uuid, '42')
@ -439,14 +439,14 @@ class StorageTestMixin(object):
s = self._get_storage()
s.ensure_atom(test_utils.NoopTask('my task', provides=set(['result'])))
s.save('my task', {})
self.assertRaisesRegexp(exceptions.NotFound,
self.assertRaisesRegex(exceptions.NotFound,
'^Unable to find result', s.fetch, 'result')
def test_empty_result_is_checked(self):
s = self._get_storage()
s.ensure_atom(test_utils.NoopTask('my task', provides=['a']))
s.save('my task', ())
self.assertRaisesRegexp(exceptions.NotFound,
self.assertRaisesRegex(exceptions.NotFound,
'^Unable to find result', s.fetch, 'a')
def test_short_result_is_checked(self):
@ -454,7 +454,7 @@ class StorageTestMixin(object):
s.ensure_atom(test_utils.NoopTask('my task', provides=['a', 'b']))
s.save('my task', ['result'])
self.assertEqual('result', s.fetch('a'))
self.assertRaisesRegexp(exceptions.NotFound,
self.assertRaisesRegex(exceptions.NotFound,
'^Unable to find result', s.fetch, 'b')
def test_ensure_retry(self):
@ -466,7 +466,7 @@ class StorageTestMixin(object):
def test_ensure_retry_and_task_with_same_name(self):
s = self._get_storage()
s.ensure_atom(test_utils.NoopTask('my retry'))
self.assertRaisesRegexp(exceptions.Duplicate,
self.assertRaisesRegex(exceptions.Duplicate,
'^Atom detail', s.ensure_atom,
test_utils.NoopRetry('my retry'))
@ -516,7 +516,7 @@ class StorageTestMixin(object):
self.assertEqual({'my retry': a_failure}, s.get_failures())
def test_logbook_get_unknown_atom_type(self):
self.assertRaisesRegexp(TypeError,
self.assertRaisesRegex(TypeError,
'Unknown atom',
models.atom_detail_class, 'some_detail')

6
taskflow/tests/unit/test_suspend.py
View File

@ -102,7 +102,7 @@ class SuspendTest(utils.EngineTestBase):
'b.t REVERTED(None)']
self.assertEqual(expected, capturer.values)
with utils.CaptureListener(engine, capture_flow=False) as capturer:
self.assertRaisesRegexp(RuntimeError, '^Woot', engine.run)
self.assertRaisesRegex(RuntimeError, '^Woot', engine.run)
self.assertEqual(states.REVERTED, engine.storage.get_flow_state())
expected = ['a.t REVERTING', 'a.t REVERTED(None)']
self.assertEqual(expected, capturer.values)
@ -132,7 +132,7 @@ class SuspendTest(utils.EngineTestBase):
# pretend we are resuming
engine2 = self._make_engine(flow, engine.storage._flowdetail)
with utils.CaptureListener(engine2, capture_flow=False) as capturer2:
self.assertRaisesRegexp(RuntimeError, '^Woot', engine2.run)
self.assertRaisesRegex(RuntimeError, '^Woot', engine2.run)
self.assertEqual(states.REVERTED, engine2.storage.get_flow_state())
expected = ['a.t REVERTING',
'a.t REVERTED(None)']
@ -169,7 +169,7 @@ class SuspendTest(utils.EngineTestBase):
)
engine2 = self._make_engine(flow2, engine.storage._flowdetail)
with utils.CaptureListener(engine2, capture_flow=False) as capturer2:
self.assertRaisesRegexp(RuntimeError, '^Woot', engine2.run)
self.assertRaisesRegex(RuntimeError, '^Woot', engine2.run)
self.assertEqual(states.REVERTED, engine2.storage.get_flow_state())
expected = ['a.t REVERTING', 'a.t REVERTED(None)']
self.assertEqual(expected, capturer2.values)

10
taskflow/tests/unit/test_task.py
View File

@ -109,7 +109,7 @@ class TaskTest(test.TestCase):
self.assertEqual({'food': 0}, my_task.save_as)
def test_bad_provides(self):
self.assertRaisesRegexp(TypeError, '^Atom provides',
self.assertRaisesRegex(TypeError, '^Atom provides',
MyTask, provides=object())
def test_requires_by_default(self):
@ -144,7 +144,7 @@ class TaskTest(test.TestCase):
self.assertEqual(expected, my_task.rebind)
def test_requires_explicit_not_enough(self):
self.assertRaisesRegexp(ValueError, '^Missing arguments',
self.assertRaisesRegex(ValueError, '^Missing arguments',
MyTask,
auto_extract=False, requires=('spam', 'eggs'))
@ -189,7 +189,7 @@ class TaskTest(test.TestCase):
my_task.requires)
def test_rebind_unknown(self):
self.assertRaisesRegexp(ValueError, '^Extra arguments',
self.assertRaisesRegex(ValueError, '^Extra arguments',
MyTask, rebind={'foo': 'bar'})
def test_rebind_unknown_kwargs(self):
@ -223,7 +223,7 @@ class TaskTest(test.TestCase):
my_task.requires)
def test_rebind_list_more(self):
self.assertRaisesRegexp(ValueError, '^Extra arguments',
self.assertRaisesRegex(ValueError, '^Extra arguments',
MyTask, rebind=('a', 'b', 'c', 'd'))
def test_rebind_list_more_kwargs(self):
@ -238,7 +238,7 @@ class TaskTest(test.TestCase):
my_task.requires)
def test_rebind_list_bad_value(self):
self.assertRaisesRegexp(TypeError, '^Invalid rebind value',
self.assertRaisesRegex(TypeError, '^Invalid rebind value',
MyTask, rebind=object())
def test_default_provides(self):