Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def _expect_test_not_flaky(self):
self._expect_call_test_address()
for test_object in (
self._mock_test,
getattr(self._mock_test, self._mock_test_method_name)
):
for flaky_attr in FlakyNames():
setattr(test_object, flaky_attr, None)
def _get_flaky_attributes(test):
actual_flaky_attributes = {
attr: getattr(
test,
attr,
None,
) for attr in FlakyNames()
}
return actual_flaky_attributes
if is_test_method:
mock_test_method = getattr(
self._mock_test,
self._mock_test_method_name
)
for flaky_attr in FlakyNames():
setattr(self._mock_test, flaky_attr, None)
setattr(mock_test_method, flaky_attr, None)
flaky(max_runs, min_passes)(mock_test_method)
else:
flaky(max_runs, min_passes)(self._mock_test)
mock_test_method = getattr(
self._mock_test,
self._mock_test_method_name
)
for flaky_attr in FlakyNames():
setattr(mock_test_method, flaky_attr, None)
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)
if current_errors is None:
current_errors = [self._mock_error]
else:
current_errors.append(self._mock_error)
self._set_flaky_attribute(
FlakyNames.CURRENT_ERRORS,
current_errors,
)
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 < min_passes
expected_plugin_handles_failure = too_few_passes and retries_remaining
did_plugin_retry_test = max_runs > 1
self._flaky_plugin.prepareTestCase(self._mock_test_case)
if is_failure:
actual_plugin_handles_failure = self._flaky_plugin.handleFailure(
self._mock_test_case,
self._assert_flaky_plugin_configured()
self._expect_test_flaky(is_test_method, max_runs, min_passes)
if current_errors is None:
current_errors = [self._mock_error]
else:
current_errors.append(self._mock_error)
self._set_flaky_attribute(
FlakyNames.CURRENT_ERRORS,
current_errors,
)
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 < min_passes
expected_plugin_handles_failure = too_few_passes and retries_remaining
did_plugin_retry_test = max_runs > 1
self._flaky_plugin.prepareTestCase(self._mock_test_case)
if is_failure:
actual_plugin_handles_failure = self._flaky_plugin.handleFailure(
self._mock_test_case,
self._mock_error,
)
else:
actual_plugin_handles_failure = self._flaky_plugin.handleError(
This method may be called multiple times for the same test run, so it has no side effects.
:param test:
The test that has raised an error
:type test:
:class:`nose.case.Test` or :class:`Function`
:return:
True, if the test needs to be rerun; False, otherwise.
:rtype:
`bool`
"""
if not self._has_flaky_attributes(test):
return False
flaky_attributes = self._get_flaky_attributes(test)
flaky_attributes[FlakyNames.CURRENT_RUNS] += 1
has_failed = self._has_flaky_test_failed(flaky_attributes)
return not has_failed
:type test:
:class:`nose.case.Test` or :class:`Function`
:param name:
The test name
:type name:
`unicode`
:param err:
Information about the test failure (from sys.exc_info())
:type err:
`tuple` of `class`, :class:`Exception`, `traceback`
:return:
Whether flaky should rerun this test.
:rtype:
`bool`
"""
rerun_filter = self._get_flaky_attribute(test, FlakyNames.RERUN_FILTER)
return rerun_filter(err, name, test, self)
By default, this means that the test has failed twice.
:param err:
Information about the test failure (from sys.exc_info())
:type err:
`tuple` of `class`, :class:`Exception`, `traceback`
:param flaky:
Dictionary of flaky attributes
:type flaky:
`dict` of `unicode` to varies
:param name:
The test name
:type name:
`unicode`
"""
min_passes = flaky[FlakyNames.MIN_PASSES]
current_passes = flaky[FlakyNames.CURRENT_PASSES]
message = self._failure_message.format(
current_passes,
min_passes,
)
self._log_test_failure(name, err, message)