Allow to pass a callable condition to the flaky marker

CHANGES.rst

@@ -4,7 +4,10 @@ Changelog
 15.2 (unreleased)
 -----------------
 
-- Nothing changed yet.
+- Allow ``@pytest.mark.flaky(condition)`` to accept a callable or a string
+  to be evaluated. The evaluated string has access to the exception instance
+  via the ``error`` object.
+  (`#230 <https://github.com/pytest-dev/pytest-rerunfailures/issues/230>`_)
 
 
 15.1 (2025-05-08)

docs/mark.rst

@@ -42,8 +42,12 @@ This will retry the test 5 times with a 2-second pause between attempts.
 ``condition``
 ^^^^^^^^^^^^^
 
-Re-run the test only if a specified condition is met.
-The condition can be any expression that evaluates to ``True`` or ``False``.
+Re-run the test only if a specified condition is met. The condition can be a
+boolean, a string to be evaluated, or a callable.
+
+**Boolean condition:**
+
+The simplest condition is a boolean value.
 
 .. code-block:: python
 
@@ -56,6 +60,45 @@ The condition can be any expression that evaluates to ``True`` or ``False``.
 
 In this example, the test will only be re-run if the operating system is Windows.
 
+**String condition:**
+
+The condition can be a string that will be evaluated. The evaluation context
+contains the following objects: ``os``, ``sys``, ``platform``, ``config`` (the
+pytest config object), and ``error`` (the exception instance that caused the
+test failure).
+
+.. code-block:: python
+
+   class MyError(Exception):
+       def __init__(self, code):
+           self.code = code
+
+   @pytest.mark.flaky(reruns=2, condition="error.code == 123")
+   def test_fail_with_my_error():
+       raise MyError(123)
+
+**Callable condition:**
+
+The condition can be a callable (e.g., a function or a lambda) that will be
+passed the exception instance that caused the test failure. The test will be
+rerun only if the callable returns ``True``.
+
+.. code-block:: python
+
+   def should_rerun(err):
+       return isinstance(err, ValueError)
+
+   @pytest.mark.flaky(reruns=2, condition=should_rerun)
+   def test_fail_with_value_error():
+       raise ValueError("some error")
+
+   @pytest.mark.flaky(reruns=2, condition=lambda e: isinstance(e, NameError))
+   def test_fail_with_name_error():
+       raise NameError("some other error")
+
+If the callable itself raises an exception, it will be caught, a warning
+will be issued, and the test will not be rerun.
+
 
 ``only_rerun``
 ^^^^^^^^^^^^^^

src/pytest_rerunfailures.py

@@ -162,19 +162,30 @@ def get_reruns_delay(item):
     return delay
 
 
-def get_reruns_condition(item):
+def get_reruns_condition(item, report):
     rerun_marker = _get_marker(item)
 
-    condition = True
     if rerun_marker is not None and "condition" in rerun_marker.kwargs:
-        condition = evaluate_condition(
-            item, rerun_marker, rerun_marker.kwargs["condition"]
+        return evaluate_condition(
+            item, rerun_marker, rerun_marker.kwargs["condition"], report
         )
 
-    return condition
+    return True
+
 
+def evaluate_condition(item, mark, condition: object, report) -> bool:
+    if callable(condition):
+        try:
+            exc = getattr(report, "excinfo", None)
+            return bool(condition(exc.value if exc else None))
+        except Exception as exc:
+            msglines = [
+                f"Error evaluating {mark.name!r} condition as a callable",
+                *traceback.format_exception_only(type(exc), exc),
+            ]
+            warnings.warn("\n".join(msglines))
+            return False
 
-def evaluate_condition(item, mark, condition: object) -> bool:
     # copy from python3.8 _pytest.skipping.py
 
     result = False
@@ -185,6 +196,7 @@ def evaluate_condition(item, mark, condition: object) -> bool:
             "sys": sys,
             "platform": platform,
             "config": item.config,
+            "error": getattr(report.excinfo, "value", None),
         }
         if hasattr(item, "obj"):
             globals_.update(item.obj.__globals__)  # type: ignore[attr-defined]
@@ -306,14 +318,10 @@ def _should_hard_fail_on_error(item, report, excinfo):
 def _should_not_rerun(item, report, reruns):
     xfail = hasattr(report, "wasxfail")
     is_terminal_error = item._terminal_errors[report.when]
-    condition = get_reruns_condition(item)
-    return (
-        item.execution_count > reruns
-        or not report.failed
-        or xfail
-        or is_terminal_error
-        or not condition
-    )
+    if item.execution_count > reruns or not report.failed or xfail or is_terminal_error:
+        return True
+
+    return not get_reruns_condition(item, report)
 
 
 def is_master(config):
@@ -518,6 +526,7 @@ def pytest_runtest_teardown(item, nextitem):
 def pytest_runtest_makereport(item, call):
     outcome = yield
     result = outcome.get_result()
+    result.excinfo = call.excinfo
     if result.when == "setup":
         # clean failed statuses at the beginning of each test/rerun
         setattr(item, "_test_failed_statuses", {})

tests/test_pytest_rerunfailures.py

@@ -1357,3 +1357,167 @@ def test_1(session_fixture, function_fixture):
     result = testdir.runpytest()
     assert_outcomes(result, passed=0, failed=1, rerun=1)
     result.stdout.fnmatch_lines("session teardown")
+
+
+def test_rerun_with_callable_condition(testdir):
+    testdir.makepyfile(
+        """
+        import pytest
+
+        def my_condition(error):
+            return isinstance(error, ValueError)
+
+        @pytest.mark.flaky(reruns=2, condition=my_condition)
+        def test_fail_two():
+            raise ValueError("some error")
+
+        @pytest.mark.flaky(reruns=2, condition=my_condition)
+        def test_fail_two_but_no_rerun():
+            raise NameError("some other error")
+
+        """
+    )
+    result = testdir.runpytest()
+    assert_outcomes(result, passed=0, failed=2, rerun=2)
+
+
+def test_rerun_with_lambda_condition(testdir):
+    testdir.makepyfile(
+        """
+        import pytest
+
+        @pytest.mark.flaky(reruns=2, condition=lambda e: isinstance(e, ValueError))
+        def test_fail_two():
+            raise ValueError("some error")
+
+        @pytest.mark.flaky(reruns=2, condition=lambda e: isinstance(e, ValueError))
+        def test_fail_two_but_no_rerun():
+            raise NameError("some other error")
+
+        """
+    )
+    result = testdir.runpytest()
+    assert_outcomes(result, passed=0, failed=2, rerun=2)
+
+
+def test_rerun_with_string_condition_and_error_object(testdir):
+    testdir.makepyfile(
+        """
+        import pytest
+
+        class MyError(Exception):
+            def __init__(self, code):
+                self.code = code
+
+        @pytest.mark.flaky(reruns=2, condition="error.code == 123")
+        def test_fail_two():
+            raise MyError(123)
+
+        @pytest.mark.flaky(reruns=2, condition="error.code == 123")
+        def test_fail_two_but_no_rerun():
+            raise MyError(456)
+
+        """
+    )
+    result = testdir.runpytest()
+    assert_outcomes(result, passed=0, failed=2, rerun=2)
+
+
+def test_reruns_with_callable_condition(testdir):
+    testdir.makepyfile(
+        """
+        import pytest
+
+        @pytest.mark.flaky(reruns=2, condition=lambda err: True)
+        def test_fail_two():
+            assert False"""
+    )
+
+    result = testdir.runpytest()
+    assert_outcomes(result, passed=0, failed=1, rerun=2)
+
+
+def test_reruns_with_callable_condition_returning_false(testdir):
+    testdir.makepyfile(
+        """
+        import pytest
+
+        @pytest.mark.flaky(reruns=2, condition=lambda err: False)
+        def test_fail_two():
+            assert False"""
+    )
+
+    result = testdir.runpytest()
+    assert_outcomes(result, passed=0, failed=1, rerun=0)
+
+
+def test_reruns_with_callable_condition_inspecting_exception(testdir):
+    testdir.makepyfile(
+        """
+        import pytest
+
+        class CustomError(Exception):
+            def __init__(self, code):
+                self.code = code
+
+        def should_rerun(err):
+            return err.code == 123
+
+        @pytest.mark.flaky(reruns=2, condition=should_rerun)
+        def test_fail_two():
+            raise CustomError(123)
+
+        @pytest.mark.flaky(reruns=2, condition=should_rerun)
+        def test_fail_three():
+            raise CustomError(456)
+        """
+    )
+
+    result = testdir.runpytest()
+    assert_outcomes(result, passed=0, failed=2, rerun=2)
+
+
+def test_reruns_with_string_condition_using_error(testdir):
+    testdir.makepyfile(
+        """
+        import pytest
+
+        class CustomError(Exception):
+            def __init__(self, code):
+                self.code = code
+
+        @pytest.mark.flaky(reruns=2, condition=\"error.code == 123\")
+        def test_fail_two():
+            raise CustomError(123)
+
+        @pytest.mark.flaky(reruns=2, condition=\"error.code == 456\")
+        def test_fail_three():
+            raise CustomError(123)
+
+        """
+    )
+
+    result = testdir.runpytest()
+    assert_outcomes(result, passed=0, failed=2, rerun=2)
+
+
+def test_reruns_with_callable_condition_raising_exception(testdir):
+    testdir.makepyfile(
+        """
+        import pytest
+
+        def condition(err):
+            raise ValueError(\"Whoops!\")
+
+        @pytest.mark.flaky(reruns=2, condition=condition)
+        def test_fail_two():
+            assert False
+        """
+    )
+
+    result = testdir.runpytest()
+    assert_outcomes(result, passed=0, failed=1, rerun=0)
+    result.stdout.fnmatch_lines([
+        "*UserWarning: Error evaluating 'flaky' condition as a callable*",
+        "*ValueError: Whoops!*",
+    ])