diff --git a/README.rst b/README.rst index 2ff8820..78bed08 100644 --- a/README.rst +++ b/README.rst @@ -334,6 +334,31 @@ Each retried attempt's traceback is appended to the ``rerun test summary info`` section. The section is emitted automatically when the flag is set, so ``-rR`` is not required. +Per-failure rerun policy (hook) +------------------------------- + +Plugins can give a *specific failure class* its own rerun count and semantics via the +``pytest_rerunfailures_rerun_policy`` hook, without changing how other failures are +rerun. The hook is called once per test, on its first failing report, and returns a +``RerunPolicy`` (or ``None`` for the default behavior): + +.. code-block:: python + + # conftest.py + from pytest_rerunfailures import RerunPolicy + + def pytest_rerunfailures_rerun_policy(item, report, call): + # e.g. give transient provider/infra errors one rerun, pass on recovery — + # while other failures keep the global --reruns / --all-reruns-need-to-pass + # behavior. + exc = call.excinfo.value if call.excinfo else None + if isinstance(exc, ConnectionError): + return RerunPolicy(reruns=1, all_reruns_need_to_pass=False) + return None + +``RerunPolicy`` fields left ``None`` keep the plugin's default for that item. The policy +is fixed at the item's first failure and applies to its whole rerun sequence. + Output ------ diff --git a/changes/311.feature.rst b/changes/311.feature.rst new file mode 100644 index 0000000..45635aa --- /dev/null +++ b/changes/311.feature.rst @@ -0,0 +1,8 @@ +Add the ``--all-reruns-need-to-pass`` option (also available as an ini setting +and as the ``all_reruns_need_to_pass`` argument of the ``flaky`` marker): after +an initial failure, every rerun must pass for the test to succeed. Higher-scoped +fixtures stay alive across the attempts of such a sequence. + +Add the ``pytest_rerunfailures_rerun_policy`` hook: plugins can return a +``RerunPolicy`` to give a specific failure class its own rerun count / +all-reruns-need-to-pass mode, without changing how other failures are rerun. diff --git a/pyproject.toml b/pyproject.toml index d7ca7a9..f7244f5 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -57,6 +57,10 @@ release = [ "zestreleaser-towncrier>=2", ] +[tool.setuptools] +package-dir = { "" = "src" } +py-modules = [ "pytest_rerunfailures", "pytest_rerunfailures_newhooks" ] + [tool.setuptools.dynamic] readme = { file = [ "HEADER.rst", "README.rst", "CHANGES.rst" ] } diff --git a/src/pytest_rerunfailures.py b/src/pytest_rerunfailures.py index db3974a..0b0cf38 100644 --- a/src/pytest_rerunfailures.py +++ b/src/pytest_rerunfailures.py @@ -12,6 +12,7 @@ import traceback import warnings from contextlib import suppress +from dataclasses import dataclass from itertools import chain from typing import Any @@ -91,6 +92,32 @@ def works_with_current_xdist(): ) +@dataclass(frozen=True) +class RerunPolicy: + """Per-failure rerun policy returned by ``pytest_rerunfailures_rerun_policy``. + + Any field left ``None`` keeps the plugin's default for that item. + + Attributes: + reruns: Rerun count for this failure (overrides ``--reruns`` / the marker). + all_reruns_need_to_pass: Override all-reruns-need-to-pass for this failure. + """ + + reruns: int | None = None + all_reruns_need_to_pass: bool | None = None + + +def pytest_addhooks(pluginmanager): + """Register this plugin's own hookspecs (see ``pytest_rerunfailures_newhooks``). + + The specs live in a dedicated module so ``add_hookspecs`` sees only them, not this + module's own hook implementations. + """ + import pytest_rerunfailures_newhooks + + pluginmanager.add_hookspecs(pytest_rerunfailures_newhooks) + + # command line options def pytest_addoption(parser): group = parser.getgroup( @@ -169,6 +196,16 @@ def pytest_addoption(parser): dest="fail_on_flaky", help="Fail the test run with exit code 7 if a flaky test passes on a rerun.", ) + group._addoption( + "--all-reruns-need-to-pass", + action="store_true", + dest="all_reruns_need_to_pass", + default=False, + help=( + "If enabled, after an initial failure, all reruns must pass " + "for the test to succeed." + ), + ) group._addoption( "--rerun-show-tracebacks", action="store_true", @@ -191,6 +228,11 @@ def pytest_addoption(parser): arg_type = "string" parser.addini("reruns", RERUNS_DESC, type=arg_type) parser.addini("reruns_delay", RERUNS_DELAY_DESC, type=arg_type) + parser.addini( + "all_reruns_need_to_pass", + "If enabled, all reruns must pass after initial failure", + type=arg_type, + ) parser.addini( "reruns_delay_backoff_factor", RERUNS_DELAY_BACKOFF_FACTOR_DESC, @@ -344,6 +386,33 @@ def get_reruns_delay(item): return delay +def get_all_reruns_need_to_pass(item): + """Get whether all reruns need to pass from marker, config, or ini.""" + rerun_marker = _get_marker(item) + + # Check marker kwargs first + if rerun_marker is not None and "all_reruns_need_to_pass" in rerun_marker.kwargs: + return rerun_marker.kwargs["all_reruns_need_to_pass"] + + # Check command-line option + all_need_pass = item.session.config.getvalue("all_reruns_need_to_pass") + if all_need_pass is not None: + return all_need_pass + + # Check ini value + try: + all_need_pass = item.session.config.getini("all_reruns_need_to_pass") + if all_need_pass: + # Parse string values like "True", "true", "1", etc. + if isinstance(all_need_pass, str): + return all_need_pass.lower() in ("true", "1", "yes", "on") + return bool(all_need_pass) + except (TypeError, ValueError): + pass + + return False + + def get_reruns_delay_backoff_factor(item): rerun_marker = _get_marker(item) @@ -747,6 +816,71 @@ def _should_not_rerun(item, report, reruns, condition): return not condition +def _get_rerun_settings(item): + """Return the (reruns, all_reruns_need_to_pass) in force for item. + + The configured values, overridden by the item's per-failure rerun policy once it + has one. + """ + reruns = get_reruns_count(item) + all_reruns_need_to_pass = get_all_reruns_need_to_pass(item) + policy = getattr(item, "_rerunfailures_policy", None) + if policy is not None: + if policy.reruns is not None: + reruns = policy.reruns + if policy.all_reruns_need_to_pass is not None: + all_reruns_need_to_pass = policy.all_reruns_need_to_pass + return reruns, all_reruns_need_to_pass + + +def _in_all_reruns_sequence(item): + """Return whether item is in an all-reruns-need-to-pass sequence. + + A failing call on the first attempt starts one when all-reruns-need-to-pass is in + force for item. + """ + _, all_reruns_need_to_pass = _get_rerun_settings(item) + return all_reruns_need_to_pass and getattr(item, "_initial_call_failed", False) + + +def _will_rerun(item): + """Return whether another attempt of item may follow the one in progress. + + pytest_runtest_teardown asks this to keep higher-scoped fixtures alive for that + attempt, and the all-reruns-need-to-pass branch of the protocol loop to find the + last attempt. The teardown runs before the loop decides, so the answer comes from + the attempt's setup and call phases. An all-reruns-need-to-pass sequence reruns + after a passing attempt too; otherwise only a failure reruns, as in + _should_not_rerun, which the loop uses for those items. + """ + reruns, _ = _get_rerun_settings(item) + _test_failed_statuses = getattr(item, "_test_failed_statuses", {}) + if _in_all_reruns_sequence(item): + # every attempt that reaches its call phase is followed by a rerun until the + # reruns are used up, whatever its outcome + return item.execution_count <= reruns and "call" in _test_failed_statuses + # Exceeding re-run limits, being free of failue statuses, encountering + # allowable exceptions indicate that the test is not to be re-run. The + # final condition decision is made after teardown, when every failed phase + # is known. A failure can also be carried by failed subtests alone, which + # leaves the call phase itself passing. + return ( + item.execution_count <= reruns + and (any(_test_failed_statuses.values()) or _get_num_failed_subtests(item) > 0) + and not any(item._test_xfailed.values()) + and not any(item._terminal_errors.values()) + ) + + +def _clean_up_for_rerun(item, report): + """Clean item's cached results from any level of setups before a rerun.""" + _remove_cached_results_from_failed_fixtures(item) + _remove_failed_setup_state_from_session(item) + _discard_test_class_instance(item) + _remove_failed_subtests_from_report(item, report) + _remove_failed_subtest_reports_from_stats(item.config, item.session, item.nodeid) + + def is_master(config): return not (hasattr(config, "workerinput") or hasattr(config, "slaveinput")) @@ -755,10 +889,12 @@ def pytest_configure(config): # add flaky marker config.addinivalue_line( "markers", - "flaky(reruns=1, reruns_delay=0, reruns_delay_backoff_factor=1.0): mark " - "test to re-run up to 'reruns' times. Add a delay of 'reruns_delay' " - "seconds between re-runs, multiplied by 'reruns_delay_backoff_factor' " - "after each attempt for an exponential backoff.", + "flaky(reruns=1, reruns_delay=0, reruns_delay_backoff_factor=1.0, " + "all_reruns_need_to_pass=False): mark test to re-run up to 'reruns' " + "times. Add a delay of 'reruns_delay' seconds between re-runs, " + "multiplied by 'reruns_delay_backoff_factor' after each attempt for an " + "exponential backoff. If 'all_reruns_need_to_pass' is True, all reruns " + "must pass.", ) check_options(config) @@ -1189,7 +1325,6 @@ def pytest_runtest_teardown(item, nextitem): # -> teardown needs to be skipped as well return - _test_failed_statuses = getattr(item, "_test_failed_statuses", {}) max_suite_reruns = item.session.config.option.max_suite_reruns if ( max_suite_reruns is not None @@ -1198,18 +1333,9 @@ def pytest_runtest_teardown(item, nextitem): _restore_suspended_finalizers(item) return - # Only remove non-function level actions from the stack if the test is to be re-run - # Exceeding re-run limits, being free of failue statuses, encountering - # allowable exceptions indicate that the test may need to be re-run. The - # final condition decision is made after teardown, when every failed phase - # is known. A failure can also be carried by failed subtests alone, which - # leaves the call phase itself passing. - if ( - item.execution_count <= reruns - and (any(_test_failed_statuses.values()) or _get_num_failed_subtests(item) > 0) - and not any(item._test_xfailed.values()) - and not any(item._terminal_errors.values()) - ): + # Only remove non-function level actions from the stack if the test may be + # re-run (see _will_rerun for what that takes). + if _will_rerun(item): # clean cached results from any level of setups _remove_cached_results_from_failed_fixtures(item) @@ -1262,7 +1388,28 @@ def pytest_runtest_makereport(item, call): result.when, False ) or hasattr(result, "wasxfail") - if result.when == "teardown" and getattr(item, "_finalizers_suspended", False): + # A failing call on the first attempt starts an all-reruns-need-to-pass sequence. + # Recorded here because the teardown that follows needs it before the protocol + # loop sees this report. + if ( + result.when == "call" + and result.failed + and getattr(item, "execution_count", None) == 1 + ): + item._initial_call_failed = True + + # On the first failure, ask plugins for a per-failure rerun policy so the protocol + # can give this failure class its own rerun count / semantics. + if result.failed and not hasattr(item, "_rerunfailures_policy"): + item._rerunfailures_policy = item.ihook.pytest_rerunfailures_rerun_policy( + item=item, report=result, call=call + ) + + if ( + result.when == "teardown" + and getattr(item, "_finalizers_suspended", False) + and not _in_all_reruns_sequence(item) + ): condition = get_reruns_condition(item, _get_reruns_condition_failures(item)) if item._terminal_errors["teardown"] or not condition: result = _teardown_suspended_finalizers(item, call, result) @@ -1288,71 +1435,121 @@ def pytest_runtest_protocol(item, nextitem): delay = get_reruns_delay(item) delay_backoff_factor = get_reruns_delay_backoff_factor(item) + all_reruns_need_to_pass = get_all_reruns_need_to_pass(item) parallel = not is_master(item.config) db = item.session.config.failures_db item.execution_count = db.get_test_failures(item.nodeid) + item._initial_call_failed = False db.set_test_reruns(item.nodeid, reruns) # A rerun count limits extra attempts, not the initial test execution. if item.execution_count > reruns and item.execution_count > 0: return True + # Track rerun results when all reruns need to pass + rerun_results = [] # Track result of each rerun (True=passed, False=failed) + policy_applied = False + need_to_run = True while need_to_run: item.execution_count += 1 item.ihook.pytest_runtest_logstart(nodeid=item.nodeid, location=item.location) reports = runtestprotocol(item, nextitem=nextitem, log=False) - condition = get_reruns_condition(item, _get_reruns_condition_failures(item)) + # Apply a per-failure rerun policy (pytest_rerunfailures_rerun_policy) once the + # first attempt has run and makereport has classified its failure. Lets a plugin + # override the rerun count / all-reruns mode for THIS item, without changing how + # other failures are rerun. + if not policy_applied: + policy = getattr(item, "_rerunfailures_policy", None) + if policy is not None: + policy_applied = True + reruns, all_reruns_need_to_pass = _get_rerun_settings(item) + db.set_test_reruns(item.nodeid, reruns) + + max_suite_reruns = item.session.config.option.max_suite_reruns rerun_triggered = False - for report in reports: # 3 reports: setup, call, teardown - report.rerun = item.execution_count - 1 - if rerun_triggered: - if report.failed: - report.outcome = "rerun" - item.ihook.pytest_runtest_logreport(report=report) - elif ( - condition - and not _reruns_condition_matches_phase(item, report.when) - and ( - report.failed - or (report.when == "call" and _get_num_failed_subtests(item) > 0) - ) + if all_reruns_need_to_pass and item._initial_call_failed: + # The initial run (execution_count==1) failed and every rerun after it + # must run. A rerun that never reaches its call phase ends it too. + will_rerun = _will_rerun(item) + if ( + will_rerun + and max_suite_reruns is not None + and not db.try_increment_suite_reruns(max_suite_reruns) ): - # Another failed phase matched the condition and will carry - # this intermediate attempt's rerun report. Do not publish a - # nonmatching failure as a final result first. - continue - elif _should_not_rerun(item, report, reruns, condition): - # no rerun needed or one already triggered, log normally - item.ihook.pytest_runtest_logreport(report=report) - else: - # failure detected and reruns not exhausted, since i < reruns - max_suite_reruns = item.session.config.option.max_suite_reruns - if max_suite_reruns is not None: - if not db.try_increment_suite_reruns(max_suite_reruns): - # Suite-wide limit exhausted -- log as final failure. - _restore_suspended_finalizers(item) - item.ihook.pytest_runtest_logreport(report=report) + # Suite-wide limit exhausted -- this attempt is the last one. + will_rerun = False + _restore_suspended_finalizers(item) + + for report in reports: # 3 reports: setup, call, teardown + report.rerun = item.execution_count - 1 + if report.when == "call": + if item.execution_count > 1: + # True if the rerun passed, False if it failed + rerun_results.append(not report.failed) + if will_rerun: + report.outcome = "rerun" + time.sleep( + delay * delay_backoff_factor ** (item.execution_count - 1) + ) + if not parallel or works_with_current_xdist(): + # will rerun test, log intermediate result + item.ihook.pytest_runtest_logreport(report=report) + _clean_up_for_rerun(item, report) + rerun_triggered = True continue - - report.outcome = "rerun" - time.sleep(delay * delay_backoff_factor ** (item.execution_count - 1)) - - if not parallel or works_with_current_xdist(): - # will rerun test, log intermediate result + if not all(rerun_results): + # At least one rerun failed, mark final outcome as failed + report.outcome = "failed" + elif rerun_triggered and report.failed: + report.outcome = "rerun" + item.ihook.pytest_runtest_logreport(report=report) + else: + condition = get_reruns_condition(item, _get_reruns_condition_failures(item)) + for report in reports: # 3 reports: setup, call, teardown + report.rerun = item.execution_count - 1 + if rerun_triggered: + if report.failed: + report.outcome = "rerun" item.ihook.pytest_runtest_logreport(report=report) + elif ( + condition + and not _reruns_condition_matches_phase(item, report.when) + and ( + report.failed + or ( + report.when == "call" and _get_num_failed_subtests(item) > 0 + ) + ) + ): + # Another failed phase matched the condition and will carry + # this intermediate attempt's rerun report. Do not publish a + # nonmatching failure as a final result first. + continue + elif _should_not_rerun(item, report, reruns, condition): + # no rerun needed or one already triggered, log normally + item.ihook.pytest_runtest_logreport(report=report) + else: + # failure detected and reruns not exhausted, since i < reruns + if max_suite_reruns is not None: + if not db.try_increment_suite_reruns(max_suite_reruns): + # Suite-wide limit exhausted -- log as final failure. + _restore_suspended_finalizers(item) + item.ihook.pytest_runtest_logreport(report=report) + continue - # cleanin item's cashed results from any level of setups - _remove_cached_results_from_failed_fixtures(item) - _remove_failed_setup_state_from_session(item) - _discard_test_class_instance(item) - _remove_failed_subtests_from_report(item, report) - _remove_failed_subtest_reports_from_stats( - item.config, item.session, item.nodeid - ) + report.outcome = "rerun" + time.sleep( + delay * delay_backoff_factor ** (item.execution_count - 1) + ) + + if not parallel or works_with_current_xdist(): + # will rerun test, log intermediate result + item.ihook.pytest_runtest_logreport(report=report) - rerun_triggered = True + _clean_up_for_rerun(item, report) + rerun_triggered = True # Do not retain ExceptionInfo tracebacks and their frame locals for the # lifetime of the collected item/session. diff --git a/src/pytest_rerunfailures_newhooks.py b/src/pytest_rerunfailures_newhooks.py new file mode 100644 index 0000000..d0232cf --- /dev/null +++ b/src/pytest_rerunfailures_newhooks.py @@ -0,0 +1,31 @@ +"""Hook specifications added by pytest-rerunfailures. + +Kept in a dedicated module so ``PluginManager.add_hookspecs`` registers only these +specs and not the plugin's own ``pytest_*`` hook implementations (pytest treats any +``pytest_``-prefixed function in a namespace as a hookspec). +""" + +import pytest + + +@pytest.hookspec(firstresult=True) +def pytest_rerunfailures_rerun_policy(item, report, call): + """Return a ``RerunPolicy`` for a test's first failure, or ``None`` for default. + + Consulted once per test, on its first failing report (``.when`` may be + ``"setup"``, ``"call"``, or ``"teardown"``). The returned policy is locked to the + item's WHOLE rerun sequence -- it is not recomputed on later attempts, so a + differently-classed failure on a rerun inherits this policy (the rerun count / + mode is fixed by the first failure). Lets a plugin give a specific failure class + (e.g. provider-infra errors) its own rerun count / semantics, without affecting + how other failures are rerun. The first non-None result wins. + + Args: + item: The test item that failed. + report: The first failing ``TestReport`` (``.when`` is ``"setup"``, ``"call"``, + or ``"teardown"``). + call: The ``CallInfo`` (``call.excinfo`` carries the exception). + + Returns: + A ``pytest_rerunfailures.RerunPolicy``, or ``None`` for default behavior. + """ diff --git a/tests/test_pytest_rerunfailures.py b/tests/test_pytest_rerunfailures.py index 2094448..87ff1aa 100644 --- a/tests/test_pytest_rerunfailures.py +++ b/tests/test_pytest_rerunfailures.py @@ -62,6 +62,24 @@ def temporary_crash(count=1): os._exit(1)""" +def events_fixture(scope="session"): + """Source of an autouse fixture logging its setup/teardown to events.txt.""" + return f""" + import py + import pytest + + @pytest.fixture(scope="{scope}", autouse=True) + def {scope}_events(): + path = py.path.local(__file__).dirpath().ensure("events.txt") + path.write(path.read() + "setup\\n") + yield + path.write(path.read() + "teardown\\n")""" + + +def read_events(testdir): + return testdir.tmpdir.join("events.txt").read().split() + + def check_outcome_field(outcomes, field_name, expected_value): field_value = outcomes.get(field_name, 0) assert field_value == expected_value, ( @@ -3341,6 +3359,472 @@ def test_1(session_fixture, function_fixture): result.stdout.fnmatch_lines("session teardown") +def test_all_reruns_need_to_pass_disabled_by_default(testdir): + """Test that default behavior is unchanged (flag disabled by default).""" + testdir.makepyfile( + """ + import py + def test_default_behavior(): + path = py.path.local(__file__).dirpath().ensure('test.res') + count = int(path.read() or 0) + path.write(count + 1) + if count == 0: + raise Exception('Fail on first attempt') + # Pass on second attempt + """ + ) + result = testdir.runpytest("--reruns", "3") + # Should pass because one successful rerun is enough (default behavior) + assert_outcomes(result, passed=1, rerun=1) + + +def test_all_reruns_need_to_pass_all_pass(testdir): + """Test that when all reruns pass, test passes.""" + testdir.makepyfile( + """ + import py + def test_all_pass(): + path = py.path.local(__file__).dirpath().ensure('test.res') + count = int(path.read() or 0) + path.write(count + 1) + if count == 0: + raise Exception('Fail on first attempt') + # Pass on all subsequent attempts + """ + ) + result = testdir.runpytest("--reruns", "3", "--all-reruns-need-to-pass") + # Should pass because all 3 reruns pass + assert_outcomes(result, passed=1, rerun=3) + + +def test_all_reruns_need_to_pass_some_fail(testdir): + """Test that when some reruns fail, test fails.""" + testdir.makepyfile( + """ + import py + def test_some_fail(): + path = py.path.local(__file__).dirpath().ensure('test.res') + count = int(path.read() or 0) + path.write(count + 1) + # Fail on attempt 0 (initial) and attempt 2 (second rerun) + if count == 0 or count == 2: + raise Exception(f'Fail on attempt {count}') + # Pass on attempts 1 and 3 + """ + ) + result = testdir.runpytest("--reruns", "3", "--all-reruns-need-to-pass") + # Should fail because rerun 2 fails + assert_outcomes(result, passed=0, failed=1, rerun=3) + + +def test_all_reruns_need_to_pass_all_fail(testdir): + """Test that when all reruns fail, test fails.""" + testdir.makepyfile( + """ + def test_all_fail(): + raise Exception('Always fail') + """ + ) + result = testdir.runpytest("--reruns", "3", "--all-reruns-need-to-pass") + # Should fail because all reruns fail + assert_outcomes(result, passed=0, failed=1, rerun=3) + + +def test_all_reruns_need_to_pass_marker_override(testdir): + """Test that marker can override command-line flag.""" + testdir.makepyfile( + """ + import pytest + import py + + @pytest.mark.flaky(reruns=3, all_reruns_need_to_pass=True) + def test_marker_override(): + path = py.path.local(__file__).dirpath().ensure('test.res') + count = int(path.read() or 0) + path.write(count + 1) + if count == 0: + raise Exception('Fail on first attempt') + # Pass on all subsequent attempts + """ + ) + # Not passing --all-reruns-need-to-pass on command line, but marker enables it + result = testdir.runpytest("--verbose") + assert_outcomes(result, passed=1, rerun=3) + + +def test_all_reruns_need_to_pass_marker_can_disable(testdir): + """Test that marker can disable flag even when set on command line.""" + testdir.makepyfile( + """ + import pytest + import py + + @pytest.mark.flaky(reruns=3, all_reruns_need_to_pass=False) + def test_marker_disables(): + path = py.path.local(__file__).dirpath().ensure('test.res') + count = int(path.read() or 0) + path.write(count + 1) + if count == 0: + raise Exception('Fail on first attempt') + # Pass on second attempt + """ + ) + # Passing --all-reruns-need-to-pass, but marker disables it + result = testdir.runpytest("--all-reruns-need-to-pass", "--verbose") + # Should pass with just one successful rerun (default behavior) + assert_outcomes(result, passed=1, rerun=1) + + +def test_all_reruns_need_to_pass_with_only_rerun(testdir): + """Test interaction with --only-rerun flag.""" + testdir.makepyfile( + """ + import py + def test_with_only_rerun(): + path = py.path.local(__file__).dirpath().ensure('test.res') + count = int(path.read() or 0) + path.write(count + 1) + if count == 0: + raise ValueError('Fail on first attempt') + # Pass on subsequent attempts + """ + ) + result = testdir.runpytest( + "--reruns", "3", "--all-reruns-need-to-pass", "--only-rerun", "ValueError" + ) + # Should pass because all reruns pass + assert_outcomes(result, passed=1, rerun=3) + + +def test_all_reruns_need_to_pass_initial_pass(testdir): + """Test that flag has no effect if test passes on first attempt.""" + testdir.makepyfile( + """ + def test_initial_pass(): + pass # Always passes + """ + ) + result = testdir.runpytest("--reruns", "3", "--all-reruns-need-to-pass") + # Should pass without any reruns + assert_outcomes(result, passed=1, rerun=0) + + +def test_all_reruns_need_to_pass_zero_reruns(testdir): + """Test that flag has no effect with reruns=0.""" + testdir.makepyfile( + """ + def test_zero_reruns(): + raise Exception('Fail') + """ + ) + result = testdir.runpytest("--reruns", "0", "--all-reruns-need-to-pass") + # Should fail without any reruns + assert_outcomes(result, passed=0, failed=1, rerun=0) + + +def test_all_reruns_need_to_pass_setup_failure(testdir): + """Test behavior when setup fails - only call phase is tracked.""" + testdir.makepyfile( + """ + import pytest + import py + + @pytest.fixture + def failing_fixture(): + path = py.path.local(__file__).dirpath().ensure('test.res') + count = int(path.read() or 0) + path.write(count + 1) + if count < 2: + raise Exception(f'Fixture fails on attempt {count}') + return "ok" + + def test_setup_failure(failing_fixture): + pass + """ + ) + result = testdir.runpytest("--reruns", "3", "--all-reruns-need-to-pass") + # Note: all_reruns_need_to_pass only tracks call phase failures, not setup/teardown + # Setup eventually passes, so test passes + assert_outcomes(result, passed=1, rerun=2) + + +def test_all_reruns_need_to_pass_command_line(testdir): + """Test that command line flag works as expected.""" + testdir.makepyfile( + """ + import py + def test_cli_flag(): + path = py.path.local(__file__).dirpath().ensure('test.res') + count = int(path.read() or 0) + path.write(count + 1) + if count == 0: + raise Exception('Fail on first attempt') + # Pass on all subsequent attempts + """ + ) + result = testdir.runpytest("--reruns", "3", "--all-reruns-need-to-pass") + # Should pass because all reruns pass + assert_outcomes(result, passed=1, rerun=3) + + +@pytest.mark.parametrize( + "xdist_args", + [ + [], + pytest.param( + ["-n", "1"], + marks=pytest.mark.skipif(not has_xdist, reason="requires pytest-xdist"), + ), + ], +) +def test_all_reruns_need_to_pass_keeps_session_fixture_across_passing_reruns( + testdir, xdist_args +): + """A passing rerun of the last test must not tear the session fixture down. + + The last test has no next item, so pytest tears every scope down after each + attempt unless the plugin keeps the fixtures alive for the coming rerun. + """ + testdir.makeconftest(events_fixture()) + testdir.makepyfile( + f""" + def test_flaky(): + {temporary_failure()}""" + ) + result = testdir.runpytest( + *xdist_args, "--reruns", "3", "--all-reruns-need-to-pass" + ) + assert_outcomes(result, passed=1, rerun=3) + assert read_events(testdir) == ["setup", "teardown"] + + +@pytest.mark.parametrize("lowered", [0, 1]) +def test_all_reruns_need_to_pass_policy_lowered_budget_tears_fixtures_down_once( + testdir, lowered +): + """The attempt the policy's budget makes final is the one that tears down. + + The policy applies after the first failure's call phase, before that attempt's + teardown: with a budget of 0 the first attempt is already the final one. + """ + testdir.makeconftest( + events_fixture() + + f""" + from pytest_rerunfailures import RerunPolicy + def pytest_rerunfailures_rerun_policy(item, report, call): + return RerunPolicy(reruns={lowered})""" + ) + testdir.makepyfile("def test_fail(): raise Exception('boom')") + result = testdir.runpytest("--reruns", "3", "--all-reruns-need-to-pass") + assert_outcomes(result, passed=0, failed=1, rerun=lowered) + assert read_events(testdir) == ["setup", "teardown"] + + +def test_all_reruns_need_to_pass_zero_reruns_tears_fixtures_down_once(testdir): + testdir.makeconftest(events_fixture()) + testdir.makepyfile("def test_fail(): raise Exception('boom')") + result = testdir.runpytest("--reruns", "0", "--all-reruns-need-to-pass") + assert_outcomes(result, passed=0, failed=1) + assert read_events(testdir) == ["setup", "teardown"] + + +def test_all_reruns_need_to_pass_setup_failure_on_rerun_tears_fixtures_down_once( + testdir, +): + """A rerun that fails in setup ends the sequence, so the fixtures go down then.""" + testdir.makeconftest(events_fixture()) + testdir.makepyfile( + """ + import py + import pytest + + @pytest.fixture + def fails_on_second_setup(): + path = py.path.local(__file__).dirpath().ensure('setup.res') + count = int(path.read() or 0) + path.write(count + 1) + if count == 1: + raise Exception('Setup fails on the first rerun') + + def test_fail(fails_on_second_setup): + raise Exception('Fail on the first attempt') + """ + ) + result = testdir.runpytest("--reruns", "3", "--all-reruns-need-to-pass") + assert_outcomes(result, passed=0, error=1, rerun=1) + assert read_events(testdir) == ["setup", "teardown"] + + +def test_all_reruns_need_to_pass_session_fixture_with_following_test(testdir): + """With a test after the flaky one the session fixture is never at risk.""" + testdir.makeconftest(events_fixture()) + testdir.makepyfile( + f""" + def test_flaky(): + {temporary_failure()} + + def test_after(): + pass""" + ) + result = testdir.runpytest("--reruns", "3", "--all-reruns-need-to-pass") + assert_outcomes(result, passed=2, rerun=3) + assert read_events(testdir) == ["setup", "teardown"] + + +def test_all_reruns_need_to_pass_module_fixture_across_module_boundary(testdir): + """A module fixture survives the reruns of the last test of its module.""" + testdir.makeconftest(events_fixture(scope="module")) + testdir.makepyfile( + test_a=f""" + def test_flaky(): + {temporary_failure()}""", + test_b="def test_after(): pass", + ) + result = testdir.runpytest("--reruns", "3", "--all-reruns-need-to-pass") + assert_outcomes(result, passed=2, rerun=3) + # One setup/teardown pair per module. + assert read_events(testdir) == ["setup", "teardown"] * 2 + + +def test_all_reruns_need_to_pass_disabled_by_policy_tears_fixtures_down_once( + testdir, +): + """A policy that turns the mode off makes the first passing rerun the last.""" + testdir.makeconftest( + events_fixture() + + """ + from pytest_rerunfailures import RerunPolicy + def pytest_rerunfailures_rerun_policy(item, report, call): + return RerunPolicy(reruns=3, all_reruns_need_to_pass=False)""" + ) + testdir.makepyfile( + f""" + def test_flaky(): + {temporary_failure()}""" + ) + result = testdir.runpytest("--reruns", "3", "--all-reruns-need-to-pass") + assert_outcomes(result, passed=1, rerun=1) + assert read_events(testdir) == ["setup", "teardown"] + + +def test_all_reruns_need_to_pass_only_started_by_a_failing_first_call(testdir): + """A failing call after the first attempt does not start the sequence. + + The first call passes and its teardown fails, which reruns the test the plain + way; the second call then fails, but the passing third attempt is the last one. + """ + testdir.makeconftest(events_fixture()) + testdir.makepyfile( + """ + import py + import pytest + + @pytest.fixture + def fails_on_first_teardown(): + yield + path = py.path.local(__file__).dirpath().ensure('teardown.res') + count = int(path.read() or 0) + path.write(count + 1) + if count == 0: + raise Exception('Teardown fails on the first attempt') + + def test_flaky(fails_on_first_teardown): + path = py.path.local(__file__).dirpath().ensure('test.res') + count = int(path.read() or 0) + path.write(count + 1) + if count == 1: + raise Exception('Fail on the second attempt') + """ + ) + result = testdir.runpytest("--reruns", "3", "--all-reruns-need-to-pass") + assert_outcomes(result, passed=2, rerun=2) + # The first attempt's teardown cannot tell that its own teardown is about to fail, + # so that attempt tears the session down once; the passing third attempt ends it. + assert read_events(testdir) == ["setup", "teardown", "setup", "teardown"] + + +def test_all_reruns_need_to_pass_item_run_twice_starts_a_fresh_sequence(testdir): + """A failing first attempt of one run does not arm the next run of the same item.""" + testdir.makeconftest( + """ + def pytest_collection_modifyitems(items): + items[:] = [items[0], items[1], items[0]]""" + ) + testdir.makepyfile( + """ + import py + + def test_a(): + path = py.path.local(__file__).dirpath().ensure('test.res') + count = int(path.read() or 0) + path.write(count + 1) + # First run: fail, pass, pass. Second run: the fourth call passes, and + # a fifth would fail if that run were a rerun sequence. + if count in (0, 4): + raise Exception(f'Fail on call {count}') + + def test_b(): + pass + """ + ) + result = testdir.runpytest("--reruns", "2", "--all-reruns-need-to-pass") + assert_outcomes(result, passed=3, rerun=2) + + +def test_rerun_policy_lowered_budget_tears_fixtures_down_once(testdir): + """Without all-reruns-need-to-pass the policy's budget still ends the lifetime.""" + testdir.makeconftest( + events_fixture() + + """ + from pytest_rerunfailures import RerunPolicy + def pytest_rerunfailures_rerun_policy(item, report, call): + return RerunPolicy(reruns=1)""" + ) + testdir.makepyfile("def test_fail(): raise Exception('boom')") + result = testdir.runpytest("--reruns", "3") + assert_outcomes(result, passed=0, failed=1, rerun=1) + assert read_events(testdir) == ["setup", "teardown"] + + +@pytest.mark.parametrize( + "policy_args", ["reruns=1", "reruns=1, all_reruns_need_to_pass=False"] +) +def test_rerun_policy_raised_from_zero_tears_fixtures_down_once(testdir, policy_args): + """A policy can raise a zero rerun count, and the fixtures stay for that rerun.""" + testdir.makeconftest( + events_fixture() + + f""" + from pytest_rerunfailures import RerunPolicy + def pytest_rerunfailures_rerun_policy(item, report, call): + return RerunPolicy({policy_args})""" + ) + testdir.makepyfile( + f""" + def test_flaky(): + {temporary_failure()}""" + ) + result = testdir.runpytest("--reruns", "0", "--all-reruns-need-to-pass") + assert_outcomes(result, passed=1, rerun=1) + assert read_events(testdir) == ["setup", "teardown"] + + +def test_rerun_flaky_condition_false_tears_fixtures_down(testdir): + """A false condition ends the sequence at the first failure: fixtures go down.""" + testdir.makeconftest(events_fixture() + events_fixture(scope="module")) + testdir.makepyfile( + """ + import pytest + + @pytest.mark.flaky(reruns=2, condition=False) + def test_fail(): + raise Exception('boom') + """ + ) + result = testdir.runpytest() + assert_outcomes(result, passed=0, failed=1) + assert read_events(testdir) == ["setup", "setup", "teardown", "teardown"] + + @pytest.mark.parametrize("mark_params", ["", "reruns=1"]) def test_force_reruns(testdir, mark_params): testdir.makepyfile( @@ -3459,6 +3943,253 @@ def test_pass(): assert result.ret != 0 +# --- pytest_rerunfailures_rerun_policy: per-failure rerun policy hook --- + + +def test_rerun_policy_none_keeps_global_behavior(testdir): + """A policy hook that returns None does not change the global rerun behavior.""" + testdir.makeconftest( + """ + def pytest_rerunfailures_rerun_policy(item, report, call): + return None + """ + ) + testdir.makepyfile("def test_fail(): raise Exception('boom')") + result = testdir.runpytest("--reruns", "2") + assert_outcomes(result, passed=0, failed=1, rerun=2) + + +def test_rerun_policy_overrides_rerun_count(testdir): + """A policy can cap the rerun count for its failure below the global --reruns.""" + testdir.makeconftest( + """ + from pytest_rerunfailures import RerunPolicy + def pytest_rerunfailures_rerun_policy(item, report, call): + return RerunPolicy(reruns=1) + """ + ) + testdir.makepyfile("def test_fail(): raise Exception('boom')") + # Global --reruns is 5, but the policy caps this failure at a single rerun. + result = testdir.runpytest("--reruns", "5") + assert_outcomes(result, passed=0, failed=1, rerun=1) + + +def test_rerun_policy_recovers_on_single_rerun(testdir): + """A failure that recovers on its one rerun passes under all-reruns mode.""" + testdir.makeconftest( + """ + from pytest_rerunfailures import RerunPolicy + def pytest_rerunfailures_rerun_policy(item, report, call): + return RerunPolicy(reruns=1, all_reruns_need_to_pass=False) + """ + ) + testdir.makepyfile( + """ + import py + def test_transient(): + path = py.path.local(__file__).dirpath().ensure('test.res') + count = int(path.read() or 0) + path.write(count + 1) + if count == 0: + raise Exception('transient failure') + """ + ) + # Global mode is all-reruns-need-to-pass with 3 reruns; the policy overrides it + # to "1 rerun, pass on recovery" for this failure. + result = testdir.runpytest("--reruns", "3", "--all-reruns-need-to-pass") + assert_outcomes(result, passed=1, rerun=1) + + +def test_rerun_policy_overrides_all_reruns_need_to_pass(testdir): + """A policy can disable all-reruns-need-to-pass for its failure (sustained fail).""" + testdir.makeconftest( + """ + from pytest_rerunfailures import RerunPolicy + def pytest_rerunfailures_rerun_policy(item, report, call): + return RerunPolicy(reruns=1, all_reruns_need_to_pass=False) + """ + ) + testdir.makepyfile("def test_fail(): raise Exception('always down')") + result = testdir.runpytest("--reruns", "3", "--all-reruns-need-to-pass") + # One rerun (not three), and it fails -> a single failure. + assert_outcomes(result, passed=0, failed=1, rerun=1) + + +def test_rerun_policy_only_applies_to_matching_failures(testdir): + """The policy can target one failure class; others keep default behavior.""" + testdir.makeconftest( + """ + from pytest_rerunfailures import RerunPolicy + def pytest_rerunfailures_rerun_policy(item, report, call): + if call.excinfo is not None and call.excinfo.typename == "InfraError": + return RerunPolicy(reruns=1, all_reruns_need_to_pass=False) + return None + """ + ) + testdir.makepyfile( + """ + class InfraError(Exception): + pass + + def test_infra(): + raise InfraError('provider down') + + def test_behavioral(): + raise AssertionError('real bug') + """ + ) + result = testdir.runpytest("--reruns", "3", "--all-reruns-need-to-pass") + # test_infra: policy -> 1 rerun; test_behavioral: default -> 3. Both fail. + assert_outcomes(result, passed=0, failed=2, rerun=4) + + +@pytest.mark.skipif(not has_xdist, reason="requires pytest-xdist") +def test_rerun_policy_under_xdist(testdir): + """The policy override + recovery work under xdist (policy per-item).""" + testdir.makeconftest( + """ + from pytest_rerunfailures import RerunPolicy + def pytest_rerunfailures_rerun_policy(item, report, call): + return RerunPolicy(reruns=1, all_reruns_need_to_pass=False) + """ + ) + testdir.makepyfile( + """ + import py + def test_transient(): + path = py.path.local(__file__).dirpath().ensure('test.res') + count = int(path.read() or 0) + path.write(count + 1) + if count == 0: + raise Exception('transient failure') + """ + ) + result = testdir.runpytest("-n", "1", "--reruns", "3", "--all-reruns-need-to-pass") + assert_outcomes(result, passed=1, rerun=1) + + +def test_rerun_policy_no_hook_impl_keeps_global_behavior(testdir): + """With NO policy hook implemented at all, global rerun behavior is unchanged. + + Guards the additive contract: suites that never implement the hook must behave + exactly as before (this is the common case). + """ + testdir.makepyfile("def test_fail(): raise Exception('boom')") + result = testdir.runpytest("--reruns", "2") + assert_outcomes(result, passed=0, failed=1, rerun=2) + + +def test_rerun_policy_can_increase_rerun_count(testdir): + """A policy can RAISE the rerun count for its failure above the global --reruns.""" + testdir.makeconftest( + """ + from pytest_rerunfailures import RerunPolicy + def pytest_rerunfailures_rerun_policy(item, report, call): + return RerunPolicy(reruns=3) + """ + ) + testdir.makepyfile("def test_fail(): raise Exception('boom')") + # Global --reruns is 1, but the policy raises this failure's count to 3. + result = testdir.runpytest("--reruns", "1") + assert_outcomes(result, passed=0, failed=1, rerun=3) + + +def test_rerun_policy_firstresult_first_non_none_wins(testdir): + """firstresult=True: the first non-None policy wins over a later implementation.""" + testdir.makepyfile( + plugin_first=""" + import pytest + from pytest_rerunfailures import RerunPolicy + @pytest.hookimpl(tryfirst=True) + def pytest_rerunfailures_rerun_policy(item, report, call): + return RerunPolicy(reruns=1) + """ + ) + testdir.makeconftest( + """ + from pytest_rerunfailures import RerunPolicy + def pytest_rerunfailures_rerun_policy(item, report, call): + return RerunPolicy(reruns=5) + """ + ) + testdir.makepyfile("def test_fail(): raise Exception('boom')") + testdir.syspathinsert() # make plugin_first importable for `-p` + # plugin_first is consulted first (tryfirst) and returns non-None, so its reruns=1 + # wins over the conftest's reruns=5 (which would give rerun=5 if it won). + result = testdir.runpytest("-p", "plugin_first", "--reruns", "3") + assert_outcomes(result, passed=0, failed=1, rerun=1) + + +def test_rerun_policy_disables_all_reruns_stops_on_first_pass(testdir): + """Overriding all_reruns_need_to_pass=False takes effect with reruns>=2. + + Distinguishes the two modes (which are indistinguishable at reruns=1): a + fail/pass/fail sequence stops on the first passing rerun under the override + (passed), whereas global all-reruns-need-to-pass would run every rerun and fail + because a later rerun failed. + """ + testdir.makeconftest( + """ + from pytest_rerunfailures import RerunPolicy + def pytest_rerunfailures_rerun_policy(item, report, call): + return RerunPolicy(reruns=2, all_reruns_need_to_pass=False) + """ + ) + testdir.makepyfile( + """ + import py + def test_flaky(): + path = py.path.local(__file__).dirpath().ensure('test.res') + count = int(path.read() or 0) + path.write(count + 1) + if count != 1: # fail (0), pass (1), fail (2) + raise Exception('flaky failure') + """ + ) + result = testdir.runpytest("--reruns", "2", "--all-reruns-need-to-pass") + # Override -> normal mode: stops on the first passing rerun. Without the override + # (all-reruns-need-to-pass) it would be passed=0, failed=1, rerun=2. + assert_outcomes(result, passed=1, rerun=1) + + +def test_rerun_policy_locked_at_first_failure(testdir): + """The policy is fixed at the first failure; a differently-classed rerun failure + inherits it and is NOT recomputed.""" + testdir.makeconftest( + """ + from pytest_rerunfailures import RerunPolicy + def pytest_rerunfailures_rerun_policy(item, report, call): + # Only the first failure (InfraError) ever reaches this hook; + # give it 1 rerun. + if call.excinfo is not None and call.excinfo.typename == "InfraError": + return RerunPolicy(reruns=1, all_reruns_need_to_pass=False) + # would apply if (wrongly) recomputed on the rerun + return RerunPolicy(reruns=3) + """ + ) + testdir.makepyfile( + """ + import py + class InfraError(Exception): + pass + + def test_mixed(): + path = py.path.local(__file__).dirpath().ensure('test.res') + count = int(path.read() or 0) + path.write(count + 1) + if count == 0: + # first failure -> policy reruns=1 + raise InfraError('provider blip') + # different class on the rerun + raise AssertionError('real bug on the rerun') + """ + ) + result = testdir.runpytest("--reruns", "3", "--all-reruns-need-to-pass") + # Locked to the infra policy (1 rerun), NOT recomputed to reruns=3 for the + # behavioral failure that surfaces on the rerun. + assert_outcomes(result, passed=0, failed=1, rerun=1) + + def test_max_suite_reruns_caps_total_reruns(testdir): """Suite limit stops reruns once the total across all tests is reached.""" testdir.makepyfile(