Improve flaky test handling using pytest-rerunfailures

Author: rodrigo.nogueira

aiohttp/pytest_plugin.py

@@ -64,6 +64,26 @@ def __call__(
     ) -> Awaitable[RawTestServer]: ...
 
 
+def get_flaky_threshold(
+    request: pytest.FixtureRequest,
+    base: float,
+    increment: float,
+) -> float:
+    """Calculate dynamic threshold for flaky tests based on rerun count.
+
+    When using `@pytest.mark.flaky(reruns=N)`:
+    - execution_count is 1-based (1 for first run, 2 for first rerun, etc.)
+    - With reruns=3, the test runs up to 4 times (1 initial + 3 reruns)
+    - Returns base threshold on first run, incrementing by `increment` per rerun
+
+    Example with reruns=3, base=20ms, increment=30ms:
+      Run 1: 20ms, Rerun 1: 50ms, Rerun 2: 80ms, Rerun 3: 110ms
+    """
+    execution_count: int = getattr(request.node, "execution_count", 0)
+    rerun_count = max(0, execution_count - 1)
+    return base + (rerun_count * increment)
+
+
 def pytest_addoption(parser):  # type: ignore[no-untyped-def]
     parser.addoption(
         "--aiohttp-fast",

requirements/test-common.in

@@ -8,6 +8,7 @@ proxy.py >= 2.4.4rc5
 pytest
 pytest-cov
 pytest-mock
+pytest-rerunfailures
 pytest-xdist
 pytest_codspeed
 python-on-whales

tests/test_client_middleware_digest_auth.py

@@ -26,6 +26,7 @@
 from aiohttp.payload import BytesIOPayload
 from aiohttp.pytest_plugin import AiohttpServer
 from aiohttp.web import Application, Request, Response
+from aiohttp.pytest_plugin import get_flaky_threshold
 
 
 @pytest.fixture
@@ -1331,25 +1332,27 @@ async def handler(request: Request) -> Response:
     assert auth_algorithms[0] == "MD5-sess"  # Not "MD5-SESS"
 
 
-def test_regex_performance() -> None:
-    value = "0" * 54773 + "\\0=a"
+@pytest.mark.flaky(reruns=3)
+def test_regex_performance(request: pytest.FixtureRequest) -> None:
+    """Test that the regex pattern doesn't suffer from ReDoS issues.
 
-    best_time = float("inf")
-    best_matches: list[tuple[str, str]] = []
+    Threshold starts at 20ms and increases on each rerun for CI variability.
+    """
+    REGEX_TIME_THRESHOLD_DEFAULT = 0.02  # 20ms
+    REGEX_TIME_INCREMENT_PER_RERUN = 0.03  # 30ms
+    # CI/platform variability (e.g., macOS runners ~40-50ms observed)
+    threshold_ms = get_flaky_threshold(
+        request, REGEX_TIME_THRESHOLD_DEFAULT, REGEX_TIME_INCREMENT_PER_RERUN
+    )
 
-    for _ in range(5):
-        start = time.perf_counter()
-        matches = _HEADER_PAIRS_PATTERN.findall(value)
-        elapsed = time.perf_counter() - start
+    value = "0" * 54773 + "\\0=a"
 
-        if elapsed < best_time:
-            best_time = elapsed
-            best_matches = matches
+    start = time.perf_counter()
+    matches = _HEADER_PAIRS_PATTERN.findall(value)
+    elapsed = time.perf_counter() - start
 
-    # Relaxed for CI/platform variability (e.g., macOS runners ~40-50ms observed)
     assert (
-        best_time < 0.1
-    ), f"Regex took {best_time * 1000:.1f}ms, expected <100ms - potential ReDoS issue"
+        elapsed < threshold_ms
+    ), f"Regex took {elapsed * 1000:.1f}ms, expected <{threshold_ms * 1000:.0f}ms - potential ReDoS issue"
 
-    # This example probably shouldn't produce a match either.
-    assert not best_matches
+    assert not matches

tests/test_imports.py

@@ -2,8 +2,8 @@
 import platform
 import sys
 from pathlib import Path
-
 import pytest
+from aiohttp.pytest_plugin import get_flaky_threshold
 
 
 def test___all__(pytester: pytest.Pytester) -> None:
@@ -28,48 +28,50 @@ def test_web___all__(pytester: pytest.Pytester) -> None:
     result.assert_outcomes(passed=0, errors=0)
 
 
-_IMPORT_TIME_THRESHOLD_PY312 = 350
+_IMPORT_TIME_THRESHOLD_PY312 = 300
 _IMPORT_TIME_THRESHOLD_DEFAULT = 200
-
-
-def _get_import_time_threshold() -> float:
-    if sys.version_info >= (3, 12):
-        return _IMPORT_TIME_THRESHOLD_PY312
-    return _IMPORT_TIME_THRESHOLD_DEFAULT
+_IMPORT_TIME_INCREMENT_PER_RERUN = 50
 
 
 @pytest.mark.internal
 @pytest.mark.dev_mode
+@pytest.mark.flaky(reruns=3)
 @pytest.mark.skipif(
     not sys.platform.startswith("linux") or platform.python_implementation() == "PyPy",
     reason="Timing is more reliable on Linux",
 )
-def test_import_time(pytester: pytest.Pytester) -> None:
+def test_import_time(request: pytest.FixtureRequest, pytester: pytest.Pytester) -> None:
     """Check that importing aiohttp doesn't take too long.
 
     Obviously, the time may vary on different machines and may need to be adjusted
     from time to time, but this should provide an early warning if something is
     added that significantly increases import time.
+
+    Threshold increases by _IMPORT_TIME_INCREMENT_PER_RERUN ms on each rerun
+    to account for CI variability.
     """
+    base_threshold = (
+        _IMPORT_TIME_THRESHOLD_PY312
+        if sys.version_info >= (3, 12)
+        else _IMPORT_TIME_THRESHOLD_DEFAULT
+    )
+    expected_time = get_flaky_threshold(
+        request, base_threshold, _IMPORT_TIME_INCREMENT_PER_RERUN
+    )
+
     root = Path(__file__).parent.parent
     old_path = os.environ.get("PYTHONPATH")
     os.environ["PYTHONPATH"] = os.pathsep.join([str(root)] + sys.path)
 
-    best_time_ms = 1000
     cmd = "import timeit; print(int(timeit.timeit('import aiohttp', number=1) * 1000))"
     try:
-        for _ in range(3):
-            r = pytester.run(sys.executable, "-We", "-c", cmd)
-
-            assert not r.stderr.str(), r.stderr.str()
-            runtime_ms = int(r.stdout.str())
-            if runtime_ms < best_time_ms:
-                best_time_ms = runtime_ms
+        r = pytester.run(sys.executable, "-We", "-c", cmd)
+        assert not r.stderr.str(), r.stderr.str()
+        runtime_ms = int(r.stdout.str())
     finally:
         if old_path is None:
             os.environ.pop("PYTHONPATH")
         else:
             os.environ["PYTHONPATH"] = old_path
 
-    expected_time = _get_import_time_threshold()
-    assert best_time_ms < expected_time
+    assert runtime_ms < expected_time