Skip to content

Commit 749d506

Browse files
committed
move workaround to conftest for other test files
1 parent f1748a0 commit 749d506

File tree

2 files changed

+39
-38
lines changed

2 files changed

+39
-38
lines changed

tests/conftest.py

Lines changed: 39 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,45 @@
1+
import anyio
12
import pytest
3+
import sse_starlette
4+
from packaging import version
25

36

47
@pytest.fixture
58
def anyio_backend():
69
return "asyncio"
10+
11+
12+
SSE_STARLETTE_VERSION = version.parse(sse_starlette.__version__)
13+
NEEDS_RESET = SSE_STARLETTE_VERSION < version.parse("3.0.0")
14+
15+
16+
@pytest.fixture(autouse=True)
17+
def reset_sse_app_status():
18+
"""Reset sse-starlette's global AppStatus singleton before each test.
19+
20+
AppStatus.should_exit_event is a global asyncio.Event that gets bound to
21+
an event loop. This ensures each test gets a fresh Event and prevents
22+
RuntimeError("bound to a different event loop") during parallel test
23+
execution with pytest-xdist.
24+
25+
NOTE: This fixture is only necessary for sse-starlette < 3.0.0.
26+
Version 3.0+ eliminated the global state issue entirely by using
27+
context-local events instead of module-level singletons, providing
28+
automatic test isolation without manual cleanup.
29+
30+
See <https://github.com/sysid/sse-starlette/pull/141> for more details.
31+
"""
32+
if not NEEDS_RESET:
33+
yield
34+
return
35+
36+
# lazy import to avoid import errors
37+
from sse_starlette.sse import AppStatus
38+
39+
# Setup: Reset before test
40+
AppStatus.should_exit_event = anyio.Event() # type: ignore[attr-defined]
41+
42+
yield
43+
44+
# Teardown: Reset after test to prevent contamination
45+
AppStatus.should_exit_event = anyio.Event() # type: ignore[attr-defined]

tests/shared/test_sse.py

Lines changed: 0 additions & 38 deletions
Original file line numberDiff line numberDiff line change
@@ -5,10 +5,8 @@
55
import anyio
66
import httpx
77
import pytest
8-
import sse_starlette
98
from anyio.abc import TaskGroup
109
from inline_snapshot import snapshot
11-
from packaging import version
1210
from pydantic import AnyUrl
1311
from starlette.applications import Starlette
1412
from starlette.requests import Request
@@ -34,42 +32,6 @@
3432
Tool,
3533
)
3634

37-
SSE_STARLETTE_VERSION = version.parse(sse_starlette.__version__)
38-
NEEDS_RESET = SSE_STARLETTE_VERSION < version.parse("3.0.0")
39-
40-
41-
@pytest.fixture(autouse=True)
42-
def reset_sse_app_status():
43-
"""Reset sse-starlette's global AppStatus singleton before each test.
44-
45-
AppStatus.should_exit_event is a global asyncio.Event that gets bound to
46-
an event loop. This ensures each test gets a fresh Event and prevents
47-
RuntimeError("bound to a different event loop") during parallel test
48-
execution with pytest-xdist.
49-
50-
NOTE: This fixture is only necessary for sse-starlette < 3.0.0.
51-
Version 3.0+ eliminated the global state issue entirely by using
52-
context-local events instead of module-level singletons, providing
53-
automatic test isolation without manual cleanup.
54-
55-
See <https://github.com/sysid/sse-starlette/pull/141> for more details.
56-
"""
57-
if not NEEDS_RESET:
58-
yield
59-
return
60-
61-
# lazy import to avoid import errors
62-
from sse_starlette.sse import AppStatus
63-
64-
# Setup: Reset before test
65-
AppStatus.should_exit_event = anyio.Event() # type: ignore[attr-defined]
66-
67-
yield
68-
69-
# Teardown: Reset after test to prevent contamination
70-
AppStatus.should_exit_event = anyio.Event() # type: ignore[attr-defined]
71-
72-
7335
SERVER_NAME = "test_server_for_SSE"
7436
TEST_SERVER_HOST = "testserver"
7537
TEST_SERVER_BASE_URL = f"http://{TEST_SERVER_HOST}"

0 commit comments

Comments
 (0)