|
| 1 | +"""Hypothesis property tests for Bulkhead. |
| 2 | +
|
| 3 | +Properties verified: |
| 4 | +1. Observed in-flight count never exceeds max_concurrent under any interleaving. |
| 5 | +2. With acquire_timeout=0 and a full bulkhead, the call raises BulkheadFullError. |
| 6 | +3. Successful acquisitions are released — back-to-back calls eventually drain |
| 7 | + without leaking slots. |
| 8 | +""" |
| 9 | + |
| 10 | +import asyncio |
| 11 | +from http import HTTPStatus |
| 12 | + |
| 13 | +import httpx2 |
| 14 | +import pytest |
| 15 | +from hypothesis import given, settings |
| 16 | +from hypothesis import strategies as st |
| 17 | + |
| 18 | +from httpware import AsyncClient |
| 19 | +from httpware.errors import BulkheadFullError |
| 20 | +from httpware.middleware.resilience.bulkhead import Bulkhead |
| 21 | + |
| 22 | + |
| 23 | +class _InFlightHandler: |
| 24 | + """Tracks max simultaneous in-flight count across all calls.""" |
| 25 | + |
| 26 | + def __init__(self, delay: float) -> None: |
| 27 | + self.delay = delay |
| 28 | + self.in_flight = 0 |
| 29 | + self.max_in_flight = 0 |
| 30 | + self.calls = 0 |
| 31 | + |
| 32 | + async def __call__(self, request: httpx2.Request) -> httpx2.Response: |
| 33 | + self.calls += 1 |
| 34 | + self.in_flight += 1 |
| 35 | + self.max_in_flight = max(self.max_in_flight, self.in_flight) |
| 36 | + try: |
| 37 | + await asyncio.sleep(self.delay) |
| 38 | + return httpx2.Response(HTTPStatus.OK, request=request) |
| 39 | + finally: |
| 40 | + self.in_flight -= 1 |
| 41 | + |
| 42 | + |
| 43 | +@given( |
| 44 | + max_concurrent=st.integers(min_value=1, max_value=8), |
| 45 | + n_requests=st.integers(min_value=1, max_value=32), |
| 46 | + delay=st.floats(min_value=0.001, max_value=0.005), |
| 47 | +) |
| 48 | +@settings(max_examples=30, deadline=None) |
| 49 | +async def test_in_flight_never_exceeds_max_concurrent( |
| 50 | + max_concurrent: int, |
| 51 | + n_requests: int, |
| 52 | + delay: float, |
| 53 | +) -> None: |
| 54 | + handler = _InFlightHandler(delay=delay) |
| 55 | + transport = httpx2.MockTransport(handler) |
| 56 | + client = AsyncClient( |
| 57 | + httpx2_client=httpx2.AsyncClient(transport=transport), |
| 58 | + middleware=[Bulkhead(max_concurrent=max_concurrent, acquire_timeout=None)], |
| 59 | + ) |
| 60 | + await asyncio.gather(*(client.get(f"https://example.test/{i}") for i in range(n_requests))) |
| 61 | + assert handler.calls == n_requests |
| 62 | + assert handler.max_in_flight <= max_concurrent |
| 63 | + |
| 64 | + |
| 65 | +@given( |
| 66 | + max_concurrent=st.integers(min_value=1, max_value=4), |
| 67 | + extra_requests=st.integers(min_value=1, max_value=8), |
| 68 | +) |
| 69 | +@settings(max_examples=20, deadline=None) |
| 70 | +async def test_fail_fast_rejects_when_at_capacity( |
| 71 | + max_concurrent: int, |
| 72 | + extra_requests: int, |
| 73 | +) -> None: |
| 74 | + handler = _InFlightHandler(delay=0.05) # hold slots long enough for fail-fast to fire |
| 75 | + transport = httpx2.MockTransport(handler) |
| 76 | + client = AsyncClient( |
| 77 | + httpx2_client=httpx2.AsyncClient(transport=transport), |
| 78 | + middleware=[Bulkhead(max_concurrent=max_concurrent, acquire_timeout=0)], |
| 79 | + ) |
| 80 | + |
| 81 | + # Fill the bulkhead with max_concurrent long-running tasks. |
| 82 | + holders = [asyncio.create_task(client.get(f"https://example.test/hold-{i}")) for i in range(max_concurrent)] |
| 83 | + await asyncio.sleep(0.005) # let the holders acquire their slots |
| 84 | + |
| 85 | + # Any extra requests should fail fast with BulkheadFullError. |
| 86 | + for i in range(extra_requests): |
| 87 | + with pytest.raises(BulkheadFullError): |
| 88 | + await client.get(f"https://example.test/extra-{i}") |
| 89 | + |
| 90 | + # Cleanup the holders. |
| 91 | + await asyncio.gather(*holders) |
| 92 | + |
| 93 | + |
| 94 | +@given( |
| 95 | + max_concurrent=st.integers(min_value=1, max_value=4), |
| 96 | + n_requests=st.integers(min_value=4, max_value=16), |
| 97 | +) |
| 98 | +@settings(max_examples=20, deadline=None) |
| 99 | +async def test_no_slot_leak_after_drain(max_concurrent: int, n_requests: int) -> None: |
| 100 | + """After all calls complete, the bulkhead has its full capacity available.""" |
| 101 | + handler = _InFlightHandler(delay=0.001) |
| 102 | + bulkhead = Bulkhead(max_concurrent=max_concurrent, acquire_timeout=None) |
| 103 | + transport = httpx2.MockTransport(handler) |
| 104 | + client = AsyncClient( |
| 105 | + httpx2_client=httpx2.AsyncClient(transport=transport), |
| 106 | + middleware=[bulkhead], |
| 107 | + ) |
| 108 | + |
| 109 | + await asyncio.gather(*(client.get(f"https://example.test/{i}") for i in range(n_requests))) |
| 110 | + |
| 111 | + # Bulkhead should be drained — _value equals max_concurrent again. |
| 112 | + # asyncio.Semaphore._value is implementation detail but reliable across CPython 3.11+. |
| 113 | + assert bulkhead._sem._value == max_concurrent # noqa: SLF001 |
0 commit comments