Refactor get_flaky_threshold into rerun_adjusted_threshold fixture

- Converted get_flaky_threshold function to a pytest fixture using indirect parametrization
- Renamed to rerun_adjusted_threshold for clarity
- Updated RST docstring with usage examples and rerun count logic
- Added proper type annotations (tuple[float, float]) for mypy compliance
- Updated test_imports.py and test_client_middleware_digest_auth.py to use new fixture
- Improved code readability by splitting long assertion lines
- Restored macOS timing observation comment (40-50ms)

Author: rodrigo.nogueira

aiohttp/pytest_plugin.py

@@ -64,21 +64,22 @@ 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
+@pytest.fixture
+def rerun_adjusted_threshold(request: pytest.FixtureRequest) -> float:
+    """Calculate dynamic threshold based on rerun count (via indirect parametrization).
+
+    Returns ``base + (rerun_count * increment)``.
+    The ``rerun_count`` is determined from ``pytest-rerunfailures`` (0 for initial run,
+    1 for first rerun, etc.).
+
+    Usage::
+
+        @pytest.mark.flaky(reruns=3)
+        @pytest.mark.parametrize("rerun_adjusted_threshold", [(20, 30)], indirect=True)
+        def test_timing(rerun_adjusted_threshold: float) -> None: ...
     """
+    param: tuple[float, float] = request.param
+    base, increment = param
     execution_count: int = getattr(request.node, "execution_count", 0)
     rerun_count = max(0, execution_count - 1)
     return base + (rerun_count * increment)

tests/test_client_middleware_digest_auth.py

@@ -24,7 +24,7 @@
 )
 from aiohttp.client_reqrep import ClientResponse
 from aiohttp.payload import BytesIOPayload
-from aiohttp.pytest_plugin import AiohttpServer, get_flaky_threshold
+from aiohttp.pytest_plugin import AiohttpServer
 from aiohttp.web import Application, Request, Response
 
 
@@ -1331,27 +1331,28 @@ async def handler(request: Request) -> Response:
     assert auth_algorithms[0] == "MD5-sess"  # Not "MD5-SESS"
 
 
+_REGEX_TIME_THRESHOLD = (0.02, 0.03)  # (base=20ms, increment=30ms)
+
+
 @pytest.mark.flaky(reruns=3)
-def test_regex_performance(request: pytest.FixtureRequest) -> None:
+@pytest.mark.parametrize(
+    "rerun_adjusted_threshold", [_REGEX_TIME_THRESHOLD], indirect=True
+)
+def test_regex_performance(rerun_adjusted_threshold: float) -> None:
     """Test that the regex pattern doesn't suffer from ReDoS issues.
 
     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
-    )
-
     value = "0" * 54773 + "\\0=a"
 
     start = time.perf_counter()
     matches = _HEADER_PAIRS_PATTERN.findall(value)
     elapsed = time.perf_counter() - start
 
-    assert (
-        elapsed < threshold_ms
-    ), f"Regex took {elapsed * 1000:.1f}ms, expected <{threshold_ms * 1000:.0f}ms - potential ReDoS issue"
+    # Relaxed for CI/platform variability (e.g., macOS runners ~40-50ms observed)
+    assert elapsed < rerun_adjusted_threshold, (
+        f"Regex took {elapsed * 1000:.1f}ms, "
+        f"expected <{rerun_adjusted_threshold * 1000:.0f}ms - potential ReDoS issue"
+    )
 
     assert not matches

tests/test_imports.py

@@ -5,8 +5,6 @@
 
 import pytest
 
-from aiohttp.pytest_plugin import get_flaky_threshold
-
 
 def test___all__(pytester: pytest.Pytester) -> None:
     """See https://github.com/aio-libs/aiohttp/issues/6197"""
@@ -33,16 +31,26 @@ def test_web___all__(pytester: pytest.Pytester) -> None:
 _IMPORT_TIME_THRESHOLD_PY312 = 300
 _IMPORT_TIME_THRESHOLD_DEFAULT = 200
 _IMPORT_TIME_INCREMENT_PER_RERUN = 50
+_IMPORT_TIME_THRESHOLD = (
+    (_IMPORT_TIME_THRESHOLD_PY312, _IMPORT_TIME_INCREMENT_PER_RERUN)
+    if sys.version_info >= (3, 12)
+    else (_IMPORT_TIME_THRESHOLD_DEFAULT, _IMPORT_TIME_INCREMENT_PER_RERUN)
+)
 
 
 @pytest.mark.internal
 @pytest.mark.dev_mode
 @pytest.mark.flaky(reruns=3)
+@pytest.mark.parametrize(
+    "rerun_adjusted_threshold", [_IMPORT_TIME_THRESHOLD], indirect=True
+)
 @pytest.mark.skipif(
     not sys.platform.startswith("linux") or platform.python_implementation() == "PyPy",
     reason="Timing is more reliable on Linux",
 )
-def test_import_time(request: pytest.FixtureRequest, pytester: pytest.Pytester) -> None:
+def test_import_time(
+    rerun_adjusted_threshold: float, 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
@@ -52,15 +60,6 @@ def test_import_time(request: pytest.FixtureRequest, pytester: pytest.Pytester)
     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)
@@ -76,4 +75,4 @@ def test_import_time(request: pytest.FixtureRequest, pytester: pytest.Pytester)
         else:
             os.environ["PYTHONPATH"] = old_path
 
-    assert runtime_ms < expected_time
+    assert runtime_ms < rerun_adjusted_threshold