pool/python-flaky
SHA256
2
0
Fork 1
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Accepting request 1172294 from devel:languages:python

Browse Source 
- Update to 3.8.1: 
  * Support pytest >= 8.1.1, and Python >= 3.12.
- Drop patches, included upstream:
  * remove_genty.patch
  * remove_mock.patch
  * remove_nose.patch
- Switch to pyproject macros.

OBS-URL: https://build.opensuse.org/request/show/1172294
OBS-URL: https://build.opensuse.org/package/show/openSUSE:Factory/python-flaky?expand=0&rev=11
This commit is contained in:
Dominique Leuenberger 2024-05-08 09:38:40 +00:00 committed by Git OBS Bridge
commit 1bd1f66a4b
7 changed files with 22 additions and 1164 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

BIN
flaky-3.7.0.tar.gz (Stored with Git LFS)
View File

Binary file not shown.

BIN
flaky-3.8.1.tar.gz (Stored with Git LFS) Normal file
View File

Binary file not shown.

11
python-flaky.changes
View File

@ -1,3 +1,14 @@
-------------------------------------------------------------------
Tue May 7 05:05:31 UTC 2024 - Steve Kowalik <steven.kowalik@suse.com>
- Update to 3.8.1:
* Support pytest >= 8.1.1, and Python >= 3.12.
- Drop patches, included upstream:
* remove_genty.patch
* remove_mock.patch
* remove_nose.patch
- Switch to pyproject macros.
-------------------------------------------------------------------
Fri Oct 20 10:57:54 UTC 2023 - Matej Cepl <mcepl@cepl.eu>

23
python-flaky.spec
View File

@ -1,7 +1,7 @@
#
# spec file
# spec file for package python-flaky
#
# Copyright (c) 2023 SUSE LLC
# Copyright (c) 2024 SUSE LLC
#
# All modifications and additions to the file contributed by third parties
# remain the property of their copyright owners, unless otherwise agreed
@ -25,22 +25,15 @@
%bcond_with test
%endif
Name: python-flaky%{?psuffix}
Version: 3.7.0
Version: 3.8.1
Release: 0
Summary: Plugin for nose or py.test that automatically reruns flaky tests
License: Apache-2.0
URL: https://github.com/box/flaky
Source: https://files.pythonhosted.org/packages/source/f/flaky/flaky-%{version}.tar.gz
# PATCH-FEATURE-UPSTREAM remove_nose.patch gh#box/flaky#171 mcepl@suse.com
# remove dependency on nose
Patch0: remove_nose.patch
# PATCH-FEATURE-UPSTREAM remove_mock.patch gh#box/flaky!197 mcepl@suse.com
# remove dependency on the external mock package
Patch1: remove_mock.patch
# PATCH-FEATURE-UPSTREAM remove_genty.patch gh#box/flaky!197 mcepl@suse.com
# remove dependency on the external genty package
Patch2: remove_genty.patch
BuildRequires: %{python_module pip}
BuildRequires: %{python_module setuptools}
BuildRequires: %{python_module wheel}
BuildRequires: fdupes
BuildRequires: python-rpm-macros
BuildArch: noarch
@ -67,12 +60,12 @@ For more information about flaky, see `this presentation <http://opensource.box.
%if !%{with test}
%build
%python_build
%pyproject_wheel
%endif
%if !%{with test}
%install
%python_install
%pyproject_install
%python_expand %fdupes %{buildroot}%{$python_sitelib}
%endif
@ -90,7 +83,7 @@ export PYTEST_ADDOPTS="--force-flaky --max-runs 2"
%doc README.rst
%license LICENSE
%{python_sitelib}/flaky
%{python_sitelib}/flaky-%{version}*-info
%{python_sitelib}/flaky-%{version}.dist-info
%endif
%changelog

219
remove_genty.patch
View File

@ -1,219 +0,0 @@
---
test/test_flaky_plugin.py | 83 +++++++++++++++++-------------------
test/test_multiprocess_string_io.py | 51 +++++++++-------------
test/test_utils.py | 17 +++----
3 files changed, 69 insertions(+), 82 deletions(-)
--- a/test/test_flaky_plugin.py
+++ b/test/test_flaky_plugin.py
@@ -2,17 +2,31 @@
from __future__ import unicode_literals
+from collections import namedtuple
from io import StringIO
from unittest import TestCase
from flaky._flaky_plugin import _FlakyPlugin
from flaky.names import FlakyNames
-from genty import genty, genty_dataset
+TestCaseDataset = namedtuple("TestCaseDataset",
+ ['max_runs', 'min_passes', 'current_runs', 'current_passes', 'expect_fail'])
-
-@genty
class TestFlakyPlugin(TestCase):
+ _test_dataset = (
+ "default_not_started": TestCaseDataset(2, 1, 0, 0, False),
+ "default_one_failure": TestCaseDataset(2, 1, 1, 0, False),
+ "default_one_success": TestCaseDataset(2, 1, 1, 1, False),
+ "default_two_failures": TestCaseDataset(2, 1, 2, 0, True),
+ "default_one_failure_one_success": TestCaseDataset(2, 1, 2, 1, False),
+ "three_two_not_started": TestCaseDataset(3, 2, 0, 0, False),
+ "three_two_one_failure": TestCaseDataset(3, 2, 1, 0, False),
+ "three_two_one_success": TestCaseDataset(3, 2, 1, 1, False),
+ "three_two_two_failures": TestCaseDataset(3, 2, 2, 0, True),
+ "three_two_one_failure_one_success": TestCaseDataset(3, 2, 2, 1, False),
+ "three_two_two_successes": TestCaseDataset(3, 2, 2, 2, False),
+ )
+
def setUp(self):
super(TestFlakyPlugin, self).setUp()
self._flaky_plugin = _FlakyPlugin()
@@ -28,43 +42,26 @@ class TestFlakyPlugin(TestCase):
mock_message,
)
- @genty_dataset(
- default_not_started=(2, 1, 0, 0, False),
- default_one_failure=(2, 1, 1, 0, False),
- default_one_success=(2, 1, 1, 1, False),
- default_two_failures=(2, 1, 2, 0, True),
- default_one_failure_one_success=(2, 1, 2, 1, False),
- three_two_not_started=(3, 2, 0, 0, False),
- three_two_one_failure=(3, 2, 1, 0, False),
- three_two_one_success=(3, 2, 1, 1, False),
- three_two_two_failures=(3, 2, 2, 0, True),
- three_two_one_failure_one_success=(3, 2, 2, 1, False),
- three_two_two_successes=(3, 2, 2, 2, False),
- )
- def test_flaky_plugin_identifies_failure(
- self,
- max_runs,
- min_passes,
- current_runs,
- current_passes,
- expect_fail,
- ):
- flaky = {
- FlakyNames.CURRENT_PASSES: current_passes,
- FlakyNames.CURRENT_RUNS: current_runs,
- FlakyNames.MAX_RUNS: max_runs,
- FlakyNames.MIN_PASSES: min_passes,
- }
- # pylint:disable=protected-access
- self.assertEqual(
- self._flaky_plugin._has_flaky_test_failed(flaky),
- expect_fail,
- )
-
- @genty_dataset('ascii stuff', 'ńőń ȁŝćȉȉ ŝƭȕƒƒ')
- def test_write_unicode_to_stream(self, message):
- stream = StringIO()
- stream.write('ascii stuff')
- # pylint:disable=protected-access
- self._flaky_plugin._stream.write(message)
- self._flaky_plugin._add_flaky_report(stream)
+ def test_flaky_plugin_identifies_failure(self):
+ for test in _test_dataset:
+ with self.subTest(test):
+ flaky = {
+ FlakyNames.CURRENT_PASSES: _test_dataset[test].current_passes,
+ FlakyNames.CURRENT_RUNS: _test_dataset[test].current_runs,
+ FlakyNames.MAX_RUNS: _test_dataset[test].max_runs,
+ FlakyNames.MIN_PASSES: _test_dataset[test].min_passes,
+ }
+ # pylint:disable=protected-access
+ self.assertEqual(
+ self._flaky_plugin._has_flaky_test_failed(flaky),
+ _test_dataset[test].expect_fail,
+ )
+
+ def test_write_unicode_to_stream(self):
+ for message in ('ascii stuff', 'ńőń ȁŝćȉȉ ŝƭȕƒƒ'):
+ with self.subTest(message):
+ stream = StringIO()
+ stream.write('ascii stuff')
+ # pylint:disable=protected-access
+ self._flaky_plugin._stream.write(message)
+ self._flaky_plugin._add_flaky_report(stream)
--- a/test/test_multiprocess_string_io.py
+++ b/test/test_multiprocess_string_io.py
@@ -5,13 +5,18 @@ from __future__ import unicode_literals
from io import StringIO
from unittest import TestCase
-from genty import genty, genty_dataset
-
-@genty
class TestMultiprocessStringIO(TestCase):
_unicode_string = 'Plain Hello'
_unicode_string_non_ascii = 'ńőń ȁŝćȉȉ ŝƭȕƒƒ'
+ _test_values = {
+ "no_writes": ([], ''),
+ "one_write": ([_unicode_string], _unicode_string),
+ "two_writes": (
+ [_unicode_string, _unicode_string_non_ascii],
+ '{}{}'.format(_unicode_string, _unicode_string_non_ascii),
+ )
+ }
def setUp(self):
super(TestMultiprocessStringIO, self).setUp()
@@ -21,29 +26,17 @@ class TestMultiprocessStringIO(TestCase)
del self._mp_string_io.proxy[:]
self._string_ios = (self._string_io, self._mp_string_io)
- @genty_dataset(
- no_writes=([], ''),
- one_write=([_unicode_string], _unicode_string),
- two_writes=(
- [_unicode_string, _unicode_string_non_ascii],
- '{}{}'.format(_unicode_string, _unicode_string_non_ascii),
- )
- )
- def test_write_then_read(self, writes, expected_value):
- for string_io in self._string_ios:
- for item in writes:
- string_io.write(item)
- self.assertEqual(string_io.getvalue(), expected_value)
-
- @genty_dataset(
- no_writes=([], ''),
- one_write=([_unicode_string], _unicode_string),
- two_writes=(
- [_unicode_string, _unicode_string_non_ascii],
- '{}{}'.format(_unicode_string, _unicode_string_non_ascii),
- )
- )
- def test_writelines_then_read(self, lines, expected_value):
- for string_io in self._string_ios:
- string_io.writelines(lines)
- self.assertEqual(string_io.getvalue(), expected_value)
+ def test_write_then_read(self):
+ for name in _test_values:
+ with self.subTest(name):
+ for string_io in self._string_ios:
+ for item in _test_values[name][0]:
+ string_io.write(item)
+ self.assertEqual(string_io.getvalue(), _test_values[name][1])
+
+ def test_writelines_then_read(self):
+ for name in _test_values:
+ with self.subTest(name):
+ for string_io in self._string_ios:
+ string_io.writelines(_test_values[name][0])
+ self.assertEqual(string_io.getvalue(), _test_values[name][1])
--- a/test/test_utils.py
+++ b/test/test_utils.py
@@ -7,10 +7,7 @@ from unittest import TestCase
from flaky.utils import ensure_unicode_string, unicode_type
-from genty import genty, genty_dataset
-
-@genty
class TestEnsureUnicodeString(TestCase):
_unicode_string = 'Plain Hello'
_byte_string = b'Plain Hello'
@@ -19,6 +16,13 @@ class TestEnsureUnicodeString(TestCase):
_hello = 'Hèllö'
_mangled_hello = 'H\ufffdll\ufffd'
_byte_string_windows_encoded = _hello.encode('windows-1252')
+ _test_dataset = (
+ (_unicode_string, _unicode_string),
+ (_byte_string, _unicode_string),
+ (_unicode_string_non_ascii, _unicode_string_non_ascii),
+ (_byte_string_non_ascii, _unicode_string_non_ascii),
+ (_byte_string_windows_encoded, _mangled_hello),
+ )
def test_ensure_unicode_string_handles_nonascii_exception_message(self):
message = '\u2013'
@@ -30,13 +34,6 @@ class TestEnsureUnicodeString(TestCase):
message = unicode_type(encoded_message)
self.assertEqual(string, message)
- @genty_dataset(
- (_unicode_string, _unicode_string),
- (_byte_string, _unicode_string),
- (_unicode_string_non_ascii, _unicode_string_non_ascii),
- (_byte_string_non_ascii, _unicode_string_non_ascii),
- (_byte_string_windows_encoded, _mangled_hello),
- )
def test_ensure_unicode_string_handles_various_strings(
self,
string,

14
remove_mock.patch
View File

@ -1,14 +0,0 @@
--- a/test/test_pytest/test_flaky_pytest_plugin.py
+++ b/test/test_pytest/test_flaky_pytest_plugin.py
@@ -2,7 +2,10 @@
from __future__ import unicode_literals
from io import StringIO
-from mock import Mock, patch
+try:
+ from unittest.mock import Mock, patch
+except ImportError:
+ from mock import Mock, patch
# pylint:disable=import-error
import pytest
from _pytest.runner import CallInfo

913
remove_nose.patch
View File

@ -1,913 +0,0 @@
--- a/test/test_nose/__init__.py
+++ /dev/null
@@ -1,3 +0,0 @@
-# coding: utf-8
-
-from __future__ import unicode_literals, absolute_import
--- a/test/test_nose/test_flaky_nose_plugin.py
+++ /dev/null
@@ -1,446 +0,0 @@
-# coding: utf-8
-
-from __future__ import unicode_literals
-
-from unittest import TestCase
-
-from genty import genty, genty_dataset
-import mock
-from mock import MagicMock, Mock, patch
-
-from flaky import defaults, flaky_nose_plugin
-from flaky.flaky_decorator import flaky
-from flaky.names import FlakyNames
-
-
-@genty
-class TestFlakyNosePlugin(TestCase):
- def setUp(self):
- super(TestFlakyNosePlugin, self).setUp()
-
- self._mock_test_result = MagicMock()
- self._mock_stream = None
- self._flaky_plugin = flaky_nose_plugin.FlakyPlugin()
- self._mock_nose_result = Mock(flaky_nose_plugin.TextTestResult)
- self._flaky_plugin.prepareTestResult(self._mock_nose_result)
- self._mock_test = MagicMock(name='flaky_plugin_test')
- self._mock_test_case = MagicMock(
- name='flaky_plugin_test_case',
- spec=TestCase
- )
- self._mock_test_case.address = MagicMock()
- self._mock_test_case.test = self._mock_test
- self._mock_test_module_name = 'test_module'
- self._mock_test_class_name = 'TestClass'
- self._mock_test_method_name = 'test_method'
- self._mock_test_names = '{}:{}.{}'.format(
- self._mock_test_module_name,
- self._mock_test_class_name,
- self._mock_test_method_name
- )
- self._mock_exception = Exception('Error in {}'.format(
- self._mock_test_method_name)
- )
- self._mock_stack_trace = ''
- self._mock_exception_type = Exception
- self._mock_error = (
- self._mock_exception_type,
- self._mock_exception,
- None,
- )
- self._mock_test_method = MagicMock(
- name=self._mock_test_method_name,
- spec=['__call__'] + list(FlakyNames().items()),
- )
- setattr(
- self._mock_test,
- self._mock_test_method_name,
- self._mock_test_method,
- )
-
- def _assert_flaky_plugin_configured(self):
- options = Mock()
- options.multiprocess_workers = 0
- conf = Mock()
- self._flaky_plugin.enabled = True
- with patch.object(flaky_nose_plugin, 'TextTestResult') as flaky_result:
- flaky_result.return_value = self._mock_test_result
- from io import StringIO
- self._mock_stream = MagicMock(spec=StringIO)
- with patch.object(self._flaky_plugin, '_get_stream') as get_stream:
- get_stream.return_value = self._mock_stream
- self._flaky_plugin.configure(options, conf)
-
- def test_flaky_plugin_report(self):
- flaky_report = 'Flaky tests passed; others failed. ' \
- 'No more tests; that ship has sailed.'
- self._test_flaky_plugin_report(flaky_report)
-
- def test_flaky_plugin_handles_success_for_test_method(self):
- self._test_flaky_plugin_handles_success()
-
- def test_flaky_plugin_handles_success_for_test_instance(self):
- self._test_flaky_plugin_handles_success(is_test_method=False)
-
- def test_flaky_plugin_handles_success_for_needs_rerun(self):
- self._test_flaky_plugin_handles_success(min_passes=2)
-
- def test_flaky_plugin_ignores_success_for_non_flaky_test(self):
- self._expect_test_not_flaky()
- self._flaky_plugin.addSuccess(self._mock_test_case)
- self._assert_test_ignored()
-
- def test_flaky_plugin_ignores_error_for_non_flaky_test(self):
- self._expect_test_not_flaky()
- self._flaky_plugin.handleError(self._mock_test_case, None)
- self._assert_test_ignored()
-
- def test_flaky_plugin_ignores_failure_for_non_flaky_test(self):
- self._expect_test_not_flaky()
- self._flaky_plugin.handleFailure(self._mock_test_case, None)
- self._assert_test_ignored()
-
- def test_flaky_plugin_ignores_error_for_nose_failure(self):
- self._mock_test_case.address.return_value = (
- None,
- self._mock_test_module_name,
- None,
- )
- self._flaky_plugin.handleError(self._mock_test_case, None)
- self._assert_test_ignored()
-
- def test_flaky_plugin_handles_error_for_test_method(self):
- self._test_flaky_plugin_handles_failure_or_error()
-
- def test_flaky_plugin_handles_error_for_test_instance(self):
- self._test_flaky_plugin_handles_failure_or_error(is_test_method=False)
-
- def test_flaky_plugin_handles_failure_for_test_method(self):
- self._test_flaky_plugin_handles_failure_or_error(is_failure=True)
-
- def test_flaky_plugin_handles_failure_for_test_instance(self):
- self._test_flaky_plugin_handles_failure_or_error(
- is_failure=True,
- is_test_method=False
- )
-
- def test_flaky_plugin_handles_failure_for_no_more_retries(self):
- self._test_flaky_plugin_handles_failure_or_error(
- is_failure=True,
- max_runs=1
- )
-
- def test_flaky_plugin_handles_additional_errors(self):
- self._test_flaky_plugin_handles_failure_or_error(
- current_errors=[self._mock_error]
- )
-
- def test_flaky_plugin_handles_bare_test(self):
- self._mock_test_names = self._mock_test_method_name
- self._mock_test.test = Mock()
- self._expect_call_test_address()
- attrib = defaults.default_flaky_attributes(2, 1)
- for name, value in attrib.items():
- setattr(
- self._mock_test.test,
- name,
- value,
- )
- delattr(self._mock_test, self._mock_test_method_name)
- self._flaky_plugin.prepareTestCase(self._mock_test_case)
- self.assertTrue(self._flaky_plugin.handleError(
- self._mock_test_case,
- self._mock_error,
- ))
- self.assertFalse(self._flaky_plugin.handleError(
- self._mock_test_case,
- self._mock_error,
- ))
-
- def _expect_call_test_address(self):
- self._mock_test_case.address.return_value = (
- None,
- None,
- self._mock_test_names
- )
-
- def _expect_test_flaky(self, is_test_method, max_runs, min_passes):
- self._expect_call_test_address()
- 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)
-
- 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 _assert_test_ignored(self):
- self._mock_test_case.address.assert_called_with()
- self.assertEqual(
- self._mock_test_case.mock_calls,
- [mock.call.address()],
- )
- self.assertEqual(self._mock_test.mock_calls, [])
- self.assertEqual(self._mock_nose_result.mock_calls, [])
-
- def _get_flaky_attributes(self):
- actual_flaky_attributes = {
- attr: getattr(
- self._mock_test_case,
- attr,
- None,
- ) for attr in FlakyNames()
- }
- for key, value in actual_flaky_attributes.items():
- if isinstance(value, list):
- actual_flaky_attributes[key] = tuple(value)
- return actual_flaky_attributes
-
- def _set_flaky_attribute(self, attr, value):
- setattr(self._mock_test, attr, value)
-
- def _assert_flaky_attributes_contains(
- self,
- expected_flaky_attributes,
- ):
- actual_flaky_attributes = self._get_flaky_attributes()
- self.assertDictContainsSubset(
- expected_flaky_attributes,
- actual_flaky_attributes,
- 'Unexpected flaky attributes. Expected {} got {}'.format(
- expected_flaky_attributes,
- actual_flaky_attributes
- )
- )
-
- def _test_flaky_plugin_handles_failure_or_error(
- self,
- current_errors=None,
- current_passes=0,
- current_runs=0,
- is_failure=False,
- 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._mock_error,
- )
- else:
- actual_plugin_handles_failure = self._flaky_plugin.handleError(
- self._mock_test_case,
- self._mock_error,
- )
-
- self.assertEqual(
- expected_plugin_handles_failure or None,
- actual_plugin_handles_failure,
- 'Expected plugin{} to handle the test run, but it did{}.'.format(
- ' to' if expected_plugin_handles_failure else '',
- '' if actual_plugin_handles_failure else ' not'
- ),
- )
- self._assert_flaky_attributes_contains(
- {
- FlakyNames.CURRENT_RUNS: current_runs + 1,
- FlakyNames.CURRENT_ERRORS: tuple(current_errors),
- },
- )
- expected_test_case_calls = [mock.call.address(), mock.call.address()]
- expected_result_calls = []
- if expected_plugin_handles_failure:
- expected_test_case_calls.append(('__hash__',))
- expected_stream_calls = [mock.call.writelines([
- self._mock_test_method_name,
- ' failed ({} runs remaining out of {}).'.format(
- max_runs - current_runs - 1, max_runs
- ),
- 'Exception: Error in test_method',
- '\n',
- ])]
- else:
- if did_plugin_retry_test:
- if is_failure:
- expected_result_calls.append(
- mock.call.addFailure(
- self._mock_test_case,
- self._mock_error,
- ),
- )
- else:
- expected_result_calls.append(mock.call.addError(
- self._mock_test_case,
- self._mock_error,
- ))
- expected_stream_calls = [mock.call.writelines([
- self._mock_test_method_name,
- ' failed; it passed {} out of the required {} times.'.format(
- current_passes,
- min_passes
- ),
- 'Exception: Error in test_method',
- '\n'
- ])]
- self.assertEqual(
- self._mock_nose_result.mock_calls,
- expected_result_calls,
- )
- self.assertEqual(
- self._mock_test_case.mock_calls,
- expected_test_case_calls,
- 'Unexpected TestCase calls: {} vs {}'.format(
- self._mock_test_case.mock_calls,
- expected_test_case_calls
- )
- )
- self.assertEqual(self._mock_stream.mock_calls, expected_stream_calls)
-
- 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,
- )
-
- 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)),
- )
- else:
- expected_stream_calls.append(mock.call.write('Success!\n'))
- self.assertEqual(
- self._mock_test_case.mock_calls,
- expected_test_case_calls,
- 'Unexpected TestCase calls = {} vs {}'.format(
- self._mock_test_case.mock_calls,
- expected_test_case_calls,
- ),
- )
- self.assertEqual(self._mock_stream.mock_calls, expected_stream_calls)
-
- def _test_flaky_plugin_report(self, expected_stream_value):
- self._assert_flaky_plugin_configured()
- mock_stream = Mock()
- self._mock_stream.getvalue.return_value = expected_stream_value
-
- self._flaky_plugin.report(mock_stream)
-
- self.assertEqual(
- mock_stream.mock_calls,
- [
- mock.call.write('===Flaky Test Report===\n\n'),
- mock.call.write(expected_stream_value),
- mock.call.write('\n===End Flaky Test Report===\n'),
- ],
- )
-
- @genty_dataset(
- multiprocess_plugin_absent=(None, 'StringIO'),
- processes_argument_absent=(0, 'StringIO'),
- processes_equals_one=(1, 'MultiprocessingStringIO'),
- processes_equals_two=(2, 'MultiprocessingStringIO'),
- )
- def test_flaky_plugin_get_stream(self, mp_workers, expected_class_name):
- options = Mock()
- conf = Mock()
- self._flaky_plugin.enabled = True
- options.multiprocess_workers = mp_workers
- if mp_workers is None:
- del options.multiprocess_workers
- self._flaky_plugin.configure(options, conf)
- # pylint:disable=protected-access
- self.assertEqual(
- self._flaky_plugin._stream.__class__.__name__,
- expected_class_name,
- )
--- a/test/test_nose/test_nose_example.py
+++ /dev/null
@@ -1,98 +0,0 @@
-# coding: utf-8
-
-from __future__ import unicode_literals
-
-from unittest import TestCase, skip
-
-from genty import genty, genty_dataset
-from nose.tools import raises
-
-from flaky import flaky
-
-
-# This is an end-to-end example of the flaky package in action. Consider it
-# a live tutorial, showing the various features in action.
-
-
-class ExampleTests(TestCase):
- _threshold = -1
-
- def test_non_flaky_thing(self):
- """Flaky will not interact with this test"""
-
- @raises(AssertionError)
- def test_non_flaky_failing_thing(self):
- """Flaky will also not interact with this test"""
- self.assertEqual(0, 1)
-
- @flaky(3, 2)
- def test_flaky_thing_that_fails_then_succeeds(self):
- """
- Flaky will run this test 3 times. It will fail once and then succeed twice.
- """
- self._threshold += 1
- if self._threshold < 1:
- raise Exception("Threshold is not high enough: {} vs {}.".format(
- self._threshold, 1),
- )
-
- @flaky(3, 2)
- def test_flaky_thing_that_succeeds_then_fails_then_succeeds(self):
- """
- Flaky will run this test 3 times. It will succeed once, fail once, and then succeed one more time.
- """
- self._threshold += 1
- if self._threshold == 1:
- self.assertEqual(0, 1)
-
- @flaky(2, 2)
- def test_flaky_thing_that_always_passes(self):
- """Flaky will run this test twice. Both will succeed."""
-
- @skip("This really fails! Remove this decorator to see the test failure.")
- @flaky()
- def test_flaky_thing_that_always_fails(self):
- """Flaky will run this test twice. Both will fail."""
- self.assertEqual(0, 1)
-
-
-@flaky
-class ExampleFlakyTests(TestCase):
- _threshold = -1
-
- def test_flaky_thing_that_fails_then_succeeds(self):
- """
- Flaky will run this test twice. It will fail once and then succeed.
- """
- self._threshold += 1
- if self._threshold < 1:
- raise Exception("Threshold is not high enough: {} vs {}.".format(
- self._threshold, 1),
- )
-
-
-def test_function():
- """
- Nose will import this function and wrap it in a :class:`FunctionTestCase`.
- It's included in the example to make sure flaky handles it correctly.
- """
-
-
-@flaky
-def test_flaky_function(param=[]):
- # pylint:disable=dangerous-default-value
- param_length = len(param)
- param.append(None)
- assert param_length == 1
-
-
-@genty
-class ExampleFlakyTestsWithUnicodeTestNames(ExampleFlakyTests):
- @genty_dataset('ascii name', 'ńőń ȁŝćȉȉ ŝƭȕƒƒ')
- def test_non_flaky_thing(self, message):
- self._threshold += 1
- if self._threshold < 1:
- raise Exception(
- "Threshold is not high enough: {} vs {} for '{}'.".format(
- self._threshold, 1, message),
- )
--- a/test/test_nose/test_nose_options_example.py
+++ /dev/null
@@ -1,54 +0,0 @@
-# coding: utf-8
-
-from __future__ import unicode_literals
-
-from unittest import TestCase
-
-from flaky import flaky
-
-# This is a series of tests that do not use the flaky decorator; the flaky
-# behavior is intended to be enabled with the --force-flaky option on the
-# command line.
-
-
-class ExampleTests(TestCase):
- _threshold = -2
-
- def test_something_flaky(self):
- """
- Flaky will run this test twice.
- It will fail once and then succeed once.
- This ensures that we mark tests as flaky even if they don't have a
- decorator when we use the command-line options.
- """
- self._threshold += 1
- if self._threshold < 0:
- raise Exception("Threshold is not high enough.")
-
- @flaky(3, 1)
- def test_flaky_thing_that_fails_then_succeeds(self):
- """
- Flaky will run this test 3 times.
- It will fail twice and then succeed once.
- This ensures that the flaky decorator overrides any command-line
- options we specify.
- """
- self._threshold += 1
- if self._threshold < 1:
- raise Exception("Threshold is not high enough.")
-
-
-@flaky(3, 1)
-class ExampleFlakyTests(TestCase):
- _threshold = -1
-
- def test_flaky_thing_that_fails_then_succeeds(self):
- """
- Flaky will run this test 3 times.
- It will fail twice and then succeed once.
- This ensures that the flaky decorator on a test suite overrides any
- command-line options we specify.
- """
- self._threshold += 1
- if self._threshold < 1:
- raise Exception("Threshold is not high enough.")
--- a/setup.py
+++ b/setup.py
@@ -68,9 +68,6 @@ def main():
cmdclass={'test': Tox},
zip_safe=False,
entry_points={
- 'nose.plugins.0.10': [
- 'flaky = flaky.flaky_nose_plugin:FlakyPlugin'
- ],
'pytest11': [
'flaky = flaky.flaky_pytest_plugin'
]
--- a/flaky/flaky_nose_plugin.py
+++ /dev/null
@@ -1,285 +0,0 @@
-# coding: utf-8
-
-from __future__ import unicode_literals
-
-import logging
-from optparse import OptionGroup
-import os
-
-from nose.failure import Failure
-from nose.plugins import Plugin
-from nose.result import TextTestResult
-
-from flaky._flaky_plugin import _FlakyPlugin
-
-
-class FlakyPlugin(_FlakyPlugin, Plugin):
- """
- Plugin for nosetests that allows retrying flaky tests.
- """
- name = 'flaky'
-
- def __init__(self):
- super(FlakyPlugin, self).__init__()
- self._logger = logging.getLogger('nose.plugins.flaky')
- self._flaky_result = None
- self._nose_result = None
- self._flaky_report = True
- self._force_flaky = False
- self._max_runs = None
- self._min_passes = None
- self._test_status = {}
- self._tests_that_reran = set()
- self._tests_that_have_been_reported = set()
-
- def options(self, parser, env=os.environ):
- """
- Base class override.
- Add options to the nose argument parser.
- """
- # pylint:disable=dangerous-default-value
- super(FlakyPlugin, self).options(parser, env=env)
- self.add_report_option(parser.add_option)
- group = OptionGroup(
- parser, "Force flaky", "Force all tests to be flaky.")
- self.add_force_flaky_options(group.add_option)
- parser.add_option_group(group)
-
- def _get_stream(self, multiprocess=False):
- """
- Get the stream used to store the flaky report.
- If this nose run is going to use the multiprocess plugin, then use
- a multiprocess-list backed StringIO proxy; otherwise, use the default
- stream.
-
- :param multiprocess:
- Whether or not this test run is configured for multiprocessing.
- :type multiprocess:
- `bool`
- :return:
- The stream to use for storing the flaky report.
- :rtype:
- :class:`StringIO` or :class:`MultiprocessingStringIO`
- """
- if multiprocess:
- from flaky.multiprocess_string_io import MultiprocessingStringIO
- return MultiprocessingStringIO()
- return self._stream
-
- def configure(self, options, conf):
- """Base class override."""
- super(FlakyPlugin, self).configure(options, conf)
- if not self.enabled:
- return
- is_multiprocess = int(getattr(options, 'multiprocess_workers', 0)) > 0
- self._stream = self._get_stream(is_multiprocess)
- self._flaky_result = TextTestResult(self._stream, [], 0)
- self._flaky_report = options.flaky_report
- self._flaky_success_report = options.flaky_success_report
- self._force_flaky = options.force_flaky
- self._max_runs = options.max_runs
- self._min_passes = options.min_passes
-
- def startTest(self, test):
- """
- Base class override. Called before a test is run.
-
- Add the test to the test status tracker, so it can potentially
- be rerun during afterTest.
-
- :param test:
- The test that is going to be run.
- :type test:
- :class:`nose.case.Test`
- """
- # pylint:disable=invalid-name
- self._test_status[test] = None
-
- def afterTest(self, test):
- """
- Base class override. Called after a test is run.
-
- If the test was marked for rerun, rerun the test.
-
- :param test:
- The test that has been run.
- :type test:
- :class:`nose.case.Test`
- """
- # pylint:disable=invalid-name
- if self._test_status[test]:
- self._tests_that_reran.add(id(test))
- test.run(self._flaky_result)
- self._test_status.pop(test, None)
-
- def _mark_test_for_rerun(self, test):
- """
- Base class override. Rerun a flaky test.
-
- In this case, don't actually rerun the test, but mark it for
- rerun during afterTest.
-
- :param test:
- The test that is going to be rerun.
- :type test:
- :class:`nose.case.Test`
- """
- self._test_status[test] = True
-
- def handleError(self, test, err):
- """
- Baseclass override. Called when a test raises an exception.
-
- If the test isn't going to be rerun again, then report the error
- to the nose test result.
-
- :param test:
- The test that has raised an error
- :type test:
- :class:`nose.case.Test`
- :param err:
- Information about the test failure (from sys.exc_info())
- :type err:
- `tuple` of `class`, :class:`Exception`, `traceback`
- :return:
- True, if the test will be rerun; False, if nose should handle it.
- :rtype:
- `bool`
- """
- # pylint:disable=invalid-name
- want_error = self._handle_test_error_or_failure(test, err)
- if not want_error and id(test) in self._tests_that_reran:
- self._nose_result.addError(test, err)
- return want_error or None
-
- def handleFailure(self, test, err):
- """
- Baseclass override. Called when a test fails.
-
- If the test isn't going to be rerun again, then report the failure
- to the nose test result.
-
- :param test:
- The test that has raised an error
- :type test:
- :class:`nose.case.Test`
- :param err:
- Information about the test failure (from sys.exc_info())
- :type err:
- `tuple` of `class`, :class:`Exception`, `traceback`
- :return:
- True, if the test will be rerun; False, if nose should handle it.
- :rtype:
- `bool`
- """
- # pylint:disable=invalid-name
- want_failure = self._handle_test_error_or_failure(test, err)
- if not want_failure and id(test) in self._tests_that_reran:
- self._nose_result.addFailure(test, err)
- return want_failure or None
-
- def addSuccess(self, test):
- """
- Baseclass override. Called when a test succeeds.
-
- Count remaining retries and compare with number of required successes
- that have not yet been achieved; retry if necessary.
-
- Returning True from this method keeps the test runner from reporting
- the test as a success; this way we can retry and only report as a
- success if we have achieved the required number of successes.
-
- :param test:
- The test that has succeeded
- :type test:
- :class:`nose.case.Test`
- :return:
- True, if the test will be rerun; False, if nose should handle it.
- :rtype:
- `bool`
- """
- # pylint:disable=invalid-name
- will_handle = self._handle_test_success(test)
- test_id = id(test)
- # If this isn't a rerun, the builtin reporter is going to report it as a success
- if will_handle and test_id not in self._tests_that_reran:
- self._tests_that_have_been_reported.add(test_id)
- # If this test hasn't already been reported as successful, then do it now
- if not will_handle and test_id in self._tests_that_reran and test_id not in self._tests_that_have_been_reported:
- self._nose_result.addSuccess(test)
- return will_handle or None
-
- def report(self, stream):
- """
- Baseclass override. Write details about flaky tests to the test report.
-
- :param stream:
- The test stream to which the report can be written.
- :type stream:
- `file`
- """
- if self._flaky_report:
- self._add_flaky_report(stream)
-
- def prepareTestResult(self, result):
- """
- Baseclass override. Called right before the first test is run.
-
- Stores the test result so that errors and failures can be reported
- to the nose test result.
-
- :param result:
- The nose test result that needs to be informed of test failures.
- :type result:
- :class:`nose.result.TextTestResult`
- """
- # pylint:disable=invalid-name
- self._nose_result = result
-
- def prepareTestCase(self, test):
- """
- Baseclass override. Called right before a test case is run.
-
- If the test class is marked flaky and the test callable is not, copy
- the flaky attributes from the test class to the test callable.
-
- :param test:
- The test that is being prepared to run
- :type test:
- :class:`nose.case.Test`
- """
- # pylint:disable=invalid-name
- if not isinstance(test.test, Failure):
- test_class = test.test
- self._copy_flaky_attributes(test, test_class)
- if self._force_flaky and not self._has_flaky_attributes(test):
- self._make_test_flaky(
- test, self._max_runs, self._min_passes)
-
- @staticmethod
- def _get_test_callable_name(test):
- """
- Base class override.
- """
- _, _, class_and_callable_name = test.address()
- first_dot_index = class_and_callable_name.find('.')
- test_callable_name = class_and_callable_name[first_dot_index + 1:]
- return test_callable_name
-
- @classmethod
- def _get_test_callable(cls, test):
- """
- Base class override.
-
- :param test:
- The test that has raised an error or succeeded
- :type test:
- :class:`nose.case.Test`
- """
- callable_name = cls._get_test_callable_name(test)
- test_callable = getattr(
- test.test,
- callable_name,
- getattr(test.test, 'test', test.test),
- )
- return test_callable