/
condition.py
131 lines (105 loc) · 4.13 KB
/
condition.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
import functools
import unittest
import six
try:
import _pytest.outcomes
_error = None
except ImportError as e:
_error = e
class QuietTestRunner(object):
def run(self, suite):
result = unittest.TestResult()
suite(result)
return result
def repeat_with_success_at_least(times, min_success):
"""Decorator for multiple trial of the test case.
The decorated test case is launched multiple times.
The case is judged as passed at least specified number of trials.
If the number of successful trials exceeds `min_success`,
the remaining trials are skipped.
Args:
times(int): The number of trials.
min_success(int): Threshold that the decorated test
case is regarded as passed.
"""
assert times >= min_success
def _repeat_with_success_at_least(f):
@functools.wraps(f)
def wrapper(*args, **kwargs):
assert len(args) > 0
instance = args[0]
assert isinstance(instance, unittest.TestCase)
success_counter = 0
failure_counter = 0
results = []
def fail():
msg = '\nFail: {0}, Success: {1}'.format(
failure_counter, success_counter)
if results:
first = results[0]
errs = first.failures + first.errors
if errs:
err_msg = '\n'.join(fail[1] for fail in errs)
msg += '\n\nThe first error message:\n' + err_msg
instance.fail(msg)
# Wrapper to convert pytest.skip() to unittest.SkipTest
def f_wrap(ins, args, kwargs):
try:
f(ins, *args[1:], **kwargs)
except _pytest.outcomes.Skipped as e:
ins.skipTest(e.msg)
for _ in six.moves.range(times):
suite = unittest.TestSuite()
# Create new instance to call the setup and the teardown only
# once.
ins = type(instance)(instance._testMethodName)
suite.addTest(
unittest.FunctionTestCase(
lambda: f_wrap(ins, args, kwargs),
setUp=ins.setUp,
tearDown=ins.tearDown))
result = QuietTestRunner().run(suite)
if len(result.skipped) == 1:
# "Skipped" is a special case of "Successful".
# When the test has been skipped, immediately quit the
# test regardless of `times` and `min_success` by raising
# SkipTest exception using the original reason.
instance.skipTest(result.skipped[0][1])
elif result.wasSuccessful():
success_counter += 1
else:
results.append(result)
failure_counter += 1
if success_counter >= min_success:
instance.assertTrue(True)
return
if failure_counter > times - min_success:
fail()
return
fail()
return wrapper
return _repeat_with_success_at_least
def repeat(times):
"""Decorator that imposes the test to be successful in a row.
Decorated test case is launched multiple times.
The case is regarded as passed only if it is successful
specified times in a row.
.. note::
In current implementation, this decorator grasps the
failure information of each trial.
Args:
times(int): The number of trials.
"""
return repeat_with_success_at_least(times, times)
def retry(times):
"""Decorator that imposes the test to be successful at least once.
Decorated test case is launched multiple times.
The case is regarded as passed if it is successful
at least once.
.. note::
In current implementation, this decorator grasps the
failure information of each trial.
Args:
times(int): The number of trials.
"""
return repeat_with_success_at_least(times, 1)