Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
actual_plugin_handles_success = self._flaky_plugin.addSuccess(
self._mock_test_case,
)
self.assertEqual(
expected_plugin_handles_success or None,
actual_plugin_handles_success,
'Expected plugin{} to handle the test run, but it did{}.'.format(
' not' if expected_plugin_handles_success else '',
'' if actual_plugin_handles_success else ' not'
),
)
self._assert_flaky_attributes_contains(
{
FlakyNames.CURRENT_RUNS: current_runs + 1,
FlakyNames.CURRENT_PASSES: current_passes + 1,
},
)
expected_test_case_calls = [mock.call.address(), mock.call.address()]
expected_stream_calls = [mock.call.writelines([
self._mock_test_method_name,
" passed {} out of the required {} times. ".format(
current_passes + 1,
min_passes,
),
])]
if expected_plugin_handles_success:
_rerun_text = 'Running test again until it passes {0} times.\n'
expected_test_case_calls.append(('__hash__',))
expected_stream_calls.append(
mock.call.write(_rerun_text.format(min_passes)),
)
min_passes=1,
rerun_filter=None,
):
flaky(max_runs, min_passes, rerun_filter)(test)
if current_errors is None:
current_errors = [None]
else:
current_errors.append(None)
setattr(
test,
FlakyNames.CURRENT_ERRORS,
current_errors,
)
setattr(
test,
FlakyNames.CURRENT_PASSES,
current_passes,
)
setattr(
test,
FlakyNames.CURRENT_RUNS,
current_runs,
)
too_few_passes = current_passes < min_passes
retries_remaining = current_runs + 1 < max_runs
expected_plugin_handles_failure = too_few_passes and retries_remaining
info.when = 'call'
actual_plugin_handles_failure = plugin.add_failure(
test,
mock_error,
def _test_flaky_plugin_handles_success(
self,
current_passes=0,
current_runs=0,
is_test_method=True,
max_runs=2,
min_passes=1
):
self._assert_flaky_plugin_configured()
self._expect_test_flaky(is_test_method, max_runs, min_passes)
self._set_flaky_attribute(
FlakyNames.CURRENT_PASSES,
current_passes,
)
self._set_flaky_attribute(
FlakyNames.CURRENT_RUNS,
current_runs,
)
retries_remaining = current_runs + 1 < max_runs
too_few_passes = current_passes + 1 < min_passes
expected_plugin_handles_success = too_few_passes and retries_remaining
self._flaky_plugin.prepareTestCase(self._mock_test_case)
actual_plugin_handles_success = self._flaky_plugin.addSuccess(
self._mock_test_case,
)
test,
FlakyNames.CURRENT_RUNS,
current_runs,
)
too_few_passes = current_passes + 1 < min_passes
retries_remaining = current_runs + 1 < max_runs
expected_plugin_handles_success = too_few_passes and retries_remaining
info.when = 'call'
actual_plugin_handles_success = plugin.add_success(test)
assert expected_plugin_handles_success == actual_plugin_handles_success
self._assert_flaky_attributes_contains(
{
FlakyNames.CURRENT_PASSES: current_passes + 1,
FlakyNames.CURRENT_RUNS: current_runs + 1,
},
test,
)
stream.writelines([
self._test_method_name,
" passed {} out of the required {} times. ".format(
current_passes + 1, min_passes,
),
])
if expected_plugin_handles_success:
stream.write(
'Running test again until it passes {} times.\n'.format(
min_passes,
),
)
pass
flaky_attribute = dict((
(attr, getattr(
test_something,
attr,
None
)) for attr in FlakyNames()
))
self.assertIsNotNone(flaky_attribute)
self.assertDictContainsSubset(
{
FlakyNames.MIN_PASSES: min_passes,
FlakyNames.MAX_RUNS: max_runs,
FlakyNames.CURRENT_PASSES: 0,
FlakyNames.CURRENT_RUNS: 0,
FlakyNames.CURRENT_ERRORS: None
},
flaky_attribute
)
:return:
True, if the test will be rerun; False, if the test runner should handle it.
:rtype:
`bool`
"""
try:
name = self._get_test_callable_name(test)
except AttributeError:
return False
need_reruns = self._should_handle_test_success(test)
if self._has_flaky_attributes(test):
self._had_flaky_tests = True
flaky = self._get_flaky_attributes(test)
min_passes = flaky[FlakyNames.MIN_PASSES]
passes = flaky[FlakyNames.CURRENT_PASSES] + 1
self._set_flaky_attribute(test, FlakyNames.CURRENT_PASSES, passes)
self._increment_flaky_attribute(test, FlakyNames.CURRENT_RUNS)
if self._flaky_success_report:
self._stream.writelines([
ensure_unicode_string(name),
' passed {} out of the required {} times. '.format(
passes,
min_passes,
),
])
if need_reruns:
self._stream.write(
'Running test again until it passes {} times.\n'.format(
min_passes,
)
True, if the test will be rerun; False, if the test runner should handle it.
:rtype:
`bool`
"""
try:
name = self._get_test_callable_name(test)
except AttributeError:
return False
need_reruns = self._should_handle_test_success(test)
if self._has_flaky_attributes(test):
self._had_flaky_tests = True
flaky = self._get_flaky_attributes(test)
min_passes = flaky[FlakyNames.MIN_PASSES]
passes = flaky[FlakyNames.CURRENT_PASSES] + 1
self._set_flaky_attribute(test, FlakyNames.CURRENT_PASSES, passes)
self._increment_flaky_attribute(test, FlakyNames.CURRENT_RUNS)
if self._flaky_success_report:
self._stream.writelines([
ensure_unicode_string(name),
' passed {} out of the required {} times. '.format(
passes,
min_passes,
),
])
if need_reruns:
self._stream.write(
'Running test again until it passes {} times.\n'.format(
min_passes,
)
)
`dict`
"""
if max_runs is None:
max_runs = 2
if min_passes is None:
min_passes = 1
if min_passes <= 0:
raise ValueError('min_passes must be positive')
if max_runs < min_passes:
raise ValueError('min_passes cannot be greater than max_runs!')
return {
FlakyNames.MAX_RUNS: max_runs,
FlakyNames.MIN_PASSES: min_passes,
FlakyNames.CURRENT_RUNS: 0,
FlakyNames.CURRENT_PASSES: 0,
FlakyNames.RERUN_FILTER: FilterWrapper(rerun_filter or _true),
}