Skip to content

Run the test suite on both AnyIO backends #3511

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
Kludex wants to merge 9 commits into
base: close-client-sse-iterators
Choose a base branch
from
+189 −223
Open

Run the test suite on both AnyIO backends #3511

Changes from 1 commit
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
a1b2b89
Restore Trio support and test both AnyIO backends
Kludex Sep 16, 2026
a922cdf
Merge remote-tracking branch 'origin/main' into restore-trio-support
Kludex Sep 16, 2026
4848495
Ignore Trio's Linux pidfd encoding warning
Kludex Sep 16, 2026
249279a
Run race-condition log checks on every exit path
Kludex Sep 16, 2026
9538140
Preserve successor streams when closing reused request IDs
Kludex Sep 16, 2026
64554aa
Accept class-based SSE iterators during cleanup
Kludex Sep 16, 2026
8d8614a
Avoid waiting for remote subscription cancellation writes
Kludex Sep 16, 2026
d060810
Keep replay network delivery outside the event-store lock
Kludex Sep 16, 2026
62d4732
Merge focused Trio prerequisites into backend enablement
Kludex Sep 18, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Prev Previous commit
Next Next commit
Run race-condition log checks on every exit path
  • Loading branch information
@Kludex
Kludex committed Sep 16, 2026
commit 249279a6434193c38ac4e9402da55ec95602f688
189 changes: 96 additions & 93 deletions tests/issues/test_1363_race_condition_streamable_http.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -99,57 +99,58 @@ async def test_race_condition_invalid_accept_headers(caplog: pytest.LogCaptureFi
- This should trigger the race condition where message_router encounters ClosedResourceError
"""
app = create_app()
with caplog.at_level(logging.ERROR), anyio.fail_after(5):
async with app.router.lifespan_context(app):
# Test with missing text/event-stream in Accept header
async with httpx2.AsyncClient(
transport=httpx2.ASGITransport(app=app), base_url="http://testserver", timeout=5.0
) as client:
response = await client.post(
"/",
json={"jsonrpc": "2.0", "method": "initialize", "id": 1, "params": {}},
headers={
"Accept": "application/json", # Missing text/event-stream
"Content-Type": "application/json",
},
)
# Should get 406 Not Acceptable due to missing text/event-stream
assert response.status_code == 406

# Test with missing application/json in Accept header
async with httpx2.AsyncClient(
transport=httpx2.ASGITransport(app=app), base_url="http://testserver", timeout=5.0
) as client:
response = await client.post(
"/",
json={"jsonrpc": "2.0", "method": "initialize", "id": 1, "params": {}},
headers={
"Accept": "text/event-stream", # Missing application/json
"Content-Type": "application/json",
},
)
# Should get 406 Not Acceptable due to missing application/json
assert response.status_code == 406

# Test with completely invalid Accept header
async with httpx2.AsyncClient(
transport=httpx2.ASGITransport(app=app), base_url="http://testserver", timeout=5.0
) as client:
response = await client.post(
"/",
json={"jsonrpc": "2.0", "method": "initialize", "id": 1, "params": {}},
headers={
"Accept": "text/plain", # Invalid Accept header
"Content-Type": "application/json",
},
)
# Should get 406 Not Acceptable
assert response.status_code == 406

# Let the message routers finish before lifespan shutdown cancels them.
await anyio.wait_all_tasks_blocked()

check_logs_for_race_condition_errors(caplog, "test_race_condition_invalid_accept_headers")
try:
with caplog.at_level(logging.ERROR), anyio.fail_after(5):
async with app.router.lifespan_context(app):
# Test with missing text/event-stream in Accept header
async with httpx2.AsyncClient(
transport=httpx2.ASGITransport(app=app), base_url="http://testserver", timeout=5.0
) as client:
response = await client.post(
"/",
json={"jsonrpc": "2.0", "method": "initialize", "id": 1, "params": {}},
headers={
"Accept": "application/json", # Missing text/event-stream
"Content-Type": "application/json",
},
)
# Should get 406 Not Acceptable due to missing text/event-stream
assert response.status_code == 406

# Test with missing application/json in Accept header
async with httpx2.AsyncClient(
transport=httpx2.ASGITransport(app=app), base_url="http://testserver", timeout=5.0
) as client:
response = await client.post(
"/",
json={"jsonrpc": "2.0", "method": "initialize", "id": 1, "params": {}},
headers={
"Accept": "text/event-stream", # Missing application/json
"Content-Type": "application/json",
},
)
# Should get 406 Not Acceptable due to missing application/json
assert response.status_code == 406

# Test with completely invalid Accept header
async with httpx2.AsyncClient(
transport=httpx2.ASGITransport(app=app), base_url="http://testserver", timeout=5.0
) as client:
response = await client.post(
"/",
json={"jsonrpc": "2.0", "method": "initialize", "id": 1, "params": {}},
headers={
"Accept": "text/plain", # Invalid Accept header
"Content-Type": "application/json",
},
)
# Should get 406 Not Acceptable
assert response.status_code == 406

# Let the message routers finish before lifespan shutdown cancels them.
await anyio.wait_all_tasks_blocked()
finally:
check_logs_for_race_condition_errors(caplog, "test_race_condition_invalid_accept_headers")


@pytest.mark.anyio
Expand All @@ -159,26 +160,27 @@ async def test_race_condition_invalid_content_type(caplog: pytest.LogCaptureFixt
This test reproduces the race condition scenario with Content-Type validation failure.
"""
app = create_app()
with caplog.at_level(logging.ERROR), anyio.fail_after(5):
async with app.router.lifespan_context(app):
# Test with invalid Content-Type
async with httpx2.AsyncClient(
transport=httpx2.ASGITransport(app=app), base_url="http://testserver", timeout=5.0
) as client:
response = await client.post(
"/",
json={"jsonrpc": "2.0", "method": "initialize", "id": 1, "params": {}},
headers={
"Accept": "application/json, text/event-stream",
"Content-Type": "text/plain", # Invalid Content-Type
},
)
assert response.status_code == 400

# Let the message router finish before lifespan shutdown cancels it.
await anyio.wait_all_tasks_blocked()

check_logs_for_race_condition_errors(caplog, "test_race_condition_invalid_content_type")
try:
with caplog.at_level(logging.ERROR), anyio.fail_after(5):
async with app.router.lifespan_context(app):
# Test with invalid Content-Type
async with httpx2.AsyncClient(
transport=httpx2.ASGITransport(app=app), base_url="http://testserver", timeout=5.0
) as client:
response = await client.post(
"/",
json={"jsonrpc": "2.0", "method": "initialize", "id": 1, "params": {}},
headers={
"Accept": "application/json, text/event-stream",
"Content-Type": "text/plain", # Invalid Content-Type
},
)
assert response.status_code == 400

# Let the message router finish before lifespan shutdown cancels it.
await anyio.wait_all_tasks_blocked()
finally:
check_logs_for_race_condition_errors(caplog, "test_race_condition_invalid_content_type")


@pytest.mark.anyio
Expand All @@ -188,25 +190,26 @@ async def test_race_condition_message_router_async_for(caplog: pytest.LogCapture
in async for loop while transport cleanup closes streams concurrently.
"""
app = create_app(json_response=True)
with caplog.at_level(logging.ERROR), anyio.fail_after(5):
async with app.router.lifespan_context(app):
# Use httpx2.ASGITransport to test the ASGI app directly
async with httpx2.AsyncClient(
transport=httpx2.ASGITransport(app=app), base_url="http://testserver", timeout=5.0
) as client:
# Send a valid initialize request
response = await client.post(
"/",
json={"jsonrpc": "2.0", "method": "initialize", "id": 1, "params": {}},
headers={
"Accept": "application/json, text/event-stream",
"Content-Type": "application/json",
},
)
# Should get a successful response
assert response.status_code in (200, 201)

# Let the message router finish before lifespan shutdown cancels it.
await anyio.wait_all_tasks_blocked()

check_logs_for_race_condition_errors(caplog, "test_race_condition_message_router_async_for")
try:
with caplog.at_level(logging.ERROR), anyio.fail_after(5):
async with app.router.lifespan_context(app):
# Use httpx2.ASGITransport to test the ASGI app directly
async with httpx2.AsyncClient(
transport=httpx2.ASGITransport(app=app), base_url="http://testserver", timeout=5.0
) as client:
# Send a valid initialize request
response = await client.post(
"/",
json={"jsonrpc": "2.0", "method": "initialize", "id": 1, "params": {}},
headers={
"Accept": "application/json, text/event-stream",
"Content-Type": "application/json",
},
)
# Should get a successful response
assert response.status_code in (200, 201)

# Let the message router finish before lifespan shutdown cancels it.
await anyio.wait_all_tasks_blocked()
finally:
check_logs_for_race_condition_errors(caplog, "test_race_condition_message_router_async_for")