pytest-rerunfailures

import pytest_rerunfailures

pytest --reruns 5

pytest --reruns 5 --reruns-delay 1

pytest --reruns 3 --only-rerun AssertionError --only-rerun ValueError

pytest --reruns 3 --rerun-except ConnectionError

import pytest

@pytest.mark.flaky(reruns=5)
def test_sometimes_fails():
    import random
    assert random.choice([True, False])

@pytest.mark.flaky(reruns=3, reruns_delay=2)
def test_with_delay():
    # Will be retried up to 3 times with 2-second delays
    pass

@pytest.mark.flaky(reruns=2, condition="sys.platform.startswith('win32')")
def test_windows_only_reruns():
    # Only reruns on Windows
    pass

def pytest_addoption(parser):
    """
    Adds command-line options for test reruns.
    
    Options added:
    --reruns: Number of times to re-run failed tests (default: 0)
    --reruns-delay: Delay in seconds between reruns
    --only-rerun: Regex patterns for errors to rerun (repeatable)
    --rerun-except: Regex patterns for errors NOT to rerun (repeatable)
    --fail-on-flaky: Fail with exit code 7 if flaky test passes on rerun
    """

def get_reruns_count(item):
    """
    Get the number of reruns for a test item.
    Priority: marker > command line > ini file
    
    Args:
        item: pytest test item
        
    Returns:
        int: Number of reruns for this test
    """

def get_reruns_delay(item):
    """
    Get the delay between reruns for a test item.
    
    Args:
        item: pytest test item
        
    Returns:
        float: Delay in seconds between reruns
    """

def get_reruns_condition(item):
    """
    Evaluate whether reruns should happen for a test item.
    
    Args:
        item: pytest test item
        
    Returns:
        bool: Whether reruns are enabled for this test
    """

def evaluate_condition(item, mark, condition: object) -> bool:
    """
    Evaluate condition expressions for flaky marker.
    
    Args:
        item: pytest test item
        mark: pytest marker object
        condition: str or bool condition to evaluate
        
    Returns:
        bool: Result of condition evaluation
    """

def pytest_configure(config):
    """
    Configure the plugin and register flaky marker.
    
    Args:
        config: pytest configuration object
    """

def pytest_runtest_protocol(item, nextitem):
    """
    Main test execution protocol with rerun logic.
    
    Args:
        item: pytest test item to execute
        nextitem: next test item in queue
        
    Returns:
        bool: True if protocol handled (reruns enabled), None for default behavior
    """

def pytest_runtest_teardown(item, nextitem):
    """
    Handle test teardown with rerun logic.
    
    Args:
        item: pytest test item
        nextitem: next test item in queue
    """

@pytest.hookimpl(hookwrapper=True)
def pytest_runtest_makereport(item, call):
    """
    Create test reports with rerun tracking.
    
    Args:
        item: pytest test item
        call: pytest call info
        
    Returns:
        TestReport: Enhanced with rerun information
    """

def pytest_report_teststatus(report):
    """
    Report test status including rerun status.
    
    Args:
        report: pytest test report
        
    Returns:
        tuple: (outcome, letter, verbose_word) for rerun status
    """

def pytest_terminal_summary(terminalreporter):
    """
    Display rerun summary in terminal output.
    
    Args:
        terminalreporter: pytest terminal reporter
    """

@pytest.hookimpl(trylast=True)
def pytest_sessionfinish(session, exitstatus):
    """
    Handle session finish with flaky test detection.
    
    Args:
        session: pytest session
        exitstatus: current exit status
    """

def check_options(config):
    """
    Validate command-line options compatibility.
    
    Args:
        config: pytest configuration object
        
    Raises:
        pytest.UsageError: If incompatible options are used
    """

def works_with_current_xdist():
    """
    Check compatibility with pytest-xdist version.
    
    Returns:
        bool or None: True if compatible, False if incompatible, None if not installed
    """

def is_master(config):
    """
    Determine if running on master node in distributed tests.
    
    Args:
        config: pytest configuration object
        
    Returns:
        bool: True if master node, False if worker
    """

def show_rerun(terminalreporter, lines):
    """
    Display rerun test information.
    
    Args:
        terminalreporter: pytest terminal reporter
        lines: list to append rerun information to
    """

class XDistHooks:
    """Hooks for pytest-xdist integration."""
    
    def pytest_configure_node(self, node):
        """
        Configure xdist worker nodes.
        
        Args:
            node: xdist worker node
        """
    
    def pytest_handlecrashitem(self, crashitem, report, sched):
        """
        Handle crashed tests in distributed execution.
        
        Args:
            crashitem: crashed test item identifier
            report: crash report
            sched: xdist scheduler
        """

class StatusDB:
    """Base in-memory database for tracking test reruns and failures."""
    
    def add_test_failure(self, crashitem):
        """
        Record a test failure.
        
        Args:
            crashitem: test item identifier
        """
    
    def get_test_failures(self, crashitem):
        """
        Get failure count for a test.
        
        Args:
            crashitem: test item identifier
            
        Returns:
            int: Number of failures recorded
        """
    
    def set_test_reruns(self, crashitem, reruns):
        """
        Set rerun count for a test.
        
        Args:
            crashitem: test item identifier
            reruns: number of reruns allowed
        """
    
    def get_test_reruns(self, crashitem):
        """
        Get rerun count for a test.
        
        Args:
            crashitem: test item identifier
            
        Returns:
            int: Number of reruns allowed
        """

class ServerStatusDB(StatusDB):
    """Server-side socket database for distributed testing."""
    
    @property
    def sock_port(self):
        """
        Get the socket port for client connections.
        
        Returns:
            int: Port number
        """
    
    def run_server(self):
        """Run the database server to handle client connections."""

class ClientStatusDB(StatusDB):
    """Client-side socket database for distributed testing."""
    
    def __init__(self, sock_port):
        """
        Initialize client database.
        
        Args:
            sock_port: server socket port to connect to
        """

@pytest.mark.flaky(
    reruns=1,                    # int: Number of reruns (default: 1)
    reruns_delay=0,              # float: Delay between reruns in seconds
    condition=True,              # str|bool: Condition for when reruns should happen
    only_rerun=None,             # str|list[str]: Regex patterns for errors to rerun
    rerun_except=None            # str|list[str]: Regex patterns for errors NOT to rerun
)
def test_function():
    """Test that may be flaky and need reruns."""

[tool.pytest.ini_options]
reruns = "3"
reruns_delay = "1"

# Only rerun on specific exceptions
@pytest.mark.flaky(only_rerun=["AssertionError", "ConnectionError"])
def test_network_operation():
    pass

# Skip reruns for specific exceptions
@pytest.mark.flaky(rerun_except="KeyboardInterrupt")
def test_user_interaction():
    pass

import sys

# Only rerun on Windows
@pytest.mark.flaky(reruns=3, condition="sys.platform.startswith('win32')")
def test_windows_specific():
    pass

# Only rerun in CI environment
@pytest.mark.flaky(reruns=2, condition="'CI' in os.environ")
def test_ci_sensitive():
    pass

# Enable crash recovery with parallel execution
pytest -n auto --reruns 3

@pytest.hookimpl(tryfirst=True)
def pytest_runtest_makereport(item, call):
    if hasattr(item, 'execution_count'):
        print(f"Test {item.nodeid} executed {item.execution_count} times")

# Test item attributes added by plugin
class TestItem:
    execution_count: int
    _test_failed_statuses: dict[str, bool]
    _terminal_errors: dict[str, bool]

# Report attributes added by plugin  
class TestReport:
    rerun: int  # 0 for first run, 1+ for reruns
    outcome: str  # may be "rerun" for intermediate reports

# Configuration constants
HAS_PYTEST_HANDLECRASHITEM: bool  # True if xdist crash handling available
RERUNS_DESC: str  # Description for --reruns option
RERUNS_DELAY_DESC: str  # Description for --reruns-delay option
suspended_finalizers: dict  # Global storage for suspended test finalizers