Add rerun_if() to request a rerun from inside a test

README.rst

@@ -250,6 +250,27 @@ Exception classes are also accepted and match any subclass:
        raise AssertionError()
 
 
+To request a re-run from *inside* a test -- for example when generated data
+happens to hit a known-bad case, similar to ``hypothesis.assume()`` -- use
+``rerun_if``:
+
+.. code-block:: python
+
+  from pytest_rerunfailures import RerunRequest, rerun_if
+
+  @pytest.mark.flaky(reruns=2, only_rerun=[RerunRequest])
+  def test_something_with_different_people():
+      person1, person2 = PersonFactory.build_batch(2)
+      rerun_if(person1.name == person2.name)
+      ...
+
+``rerun_if`` aborts the current attempt by raising ``RerunRequest`` if the
+condition is true, and the test is re-run while it still has re-runs left
+(``@pytest.mark.flaky(reruns=...)`` or ``--reruns``). Without a re-run
+budget the request behaves like a regular test failure. Filtering with
+``only_rerun=[RerunRequest]`` ensures only the requested condition triggers
+a re-run, not other failures.
+
 You can use ``@pytest.mark.flaky(condition)`` similarly as ``@pytest.mark.skipif(condition)``, see `pytest-mark-skipif <https://docs.pytest.org/en/6.2.x/reference.html#pytest-mark-skipif>`_
 
 .. code-block:: python

changes/58.feature.rst

@@ -0,0 +1,4 @@
+Add ``rerun_if()`` helper and ``RerunRequest`` exception to request a re-run
+from inside a test (similar to ``hypothesis.assume()``), e.g. combined with
+``@pytest.mark.flaky(reruns=N, only_rerun=[RerunRequest])`` to retry only on
+the requested condition.

src/pytest_rerunfailures.py

@@ -79,6 +79,25 @@ def works_with_current_xdist():
     "exponential backoff (delay * factor ** (attempt - 1)). defaults to 1.0, "
     "i.e. a constant delay."
 )
+
+
+class RerunRequest(Exception):
+    """Raised by :func:`rerun_if` to discard an attempt and request a rerun."""
+
+
+def rerun_if(condition, reason=None):
+    """Request a rerun from inside a test when ``condition`` is true.
+
+    The current attempt is aborted and reported as a failure that can be
+    retried if the test still has reruns left (via ``@pytest.mark.flaky`` or
+    ``--reruns``). Without a rerun budget it behaves like a regular test
+    failure. Combine with ``only_rerun=[RerunRequest]`` to rerun *only* on
+    this condition and not on other failures.
+    """
+    if condition:
+        raise RerunRequest(reason or "rerun requested from test")
+
+
 ONLY_RERUN_DESC = (
     "If passed, only rerun errors matching the regex provided. "
     "Pass this flag multiple times (or list one regex per line in the ini "

tests/test_pytest_rerunfailures.py

@@ -548,6 +548,55 @@ def test_pass():
     assert_outcomes(result, passed=1, rerun=2)
 
 
+def test_rerun_if_requests_rerun_from_inside_test(testdir):
+    testdir.makepyfile(
+        """
+        import pytest
+        from pytest_rerunfailures import rerun_if
+
+        import py
+        path = py.path.local(__file__).dirpath().ensure('test.res')
+
+        @pytest.mark.flaky(reruns=1)
+        def test_pass():
+            count = int(path.read() or 0)
+            path.write(count + 1)
+            rerun_if(count == 0, 'first attempt has bad data')
+        """
+    )
+    result = testdir.runpytest()
+    assert_outcomes(result, passed=1, rerun=1)
+
+
+def test_rerun_if_only_rerun_restricts_to_requested_reruns(testdir):
+    testdir.makepyfile(
+        """
+        import pytest
+        from pytest_rerunfailures import RerunRequest, rerun_if
+
+        @pytest.mark.flaky(reruns=2, only_rerun=[RerunRequest])
+        def test_fail():
+            rerun_if(False)
+            assert False
+        """
+    )
+    result = testdir.runpytest()
+    assert_outcomes(result, passed=0, failed=1, rerun=0)
+
+
+def test_rerun_if_without_rerun_budget_fails_normally(testdir):
+    testdir.makepyfile(
+        """
+        from pytest_rerunfailures import rerun_if
+
+        def test_fail():
+            rerun_if(True)
+        """
+    )
+    result = testdir.runpytest()
+    assert_outcomes(result, passed=0, failed=1, rerun=0)
+
+
 def test_no_extra_test_summary_for_reruns_by_default(testdir):
     testdir.makepyfile(
         f"""