2020from django .core .asgi import get_asgi_application
2121
2222
23+ @pytest .fixture (scope = "session" )
24+ def asgi_app ():
25+ """Shared ASGI application for all tests."""
26+ return get_asgi_application ()
27+
28+
2329@pytest .mark .asyncio
24- async def test_async_user_access ():
30+ async def test_async_user_access (asgi_app ):
2531 """
2632 Test that middleware can access request.user in async context.
2733
@@ -32,19 +38,18 @@ async def test_async_user_access():
3238 trigger the lazy loading bug. This test verifies the middleware works
3339 in the common case.
3440 """
35- app = get_asgi_application ()
36- async with AsyncClient (transport = ASGITransport (app = app ), base_url = "http://testserver" ) as ac :
41+ async with AsyncClient (transport = ASGITransport (app = asgi_app ), base_url = "http://testserver" ) as ac :
3742 response = await ac .get ("/test/async-user" )
3843
3944 assert response .status_code == 200
4045 data = response .json ()
4146 assert data ["status" ] == "success"
4247 assert "django_version" in data
43- print (f"✓ Async user access test passed: { data ['message' ]} " )
4448
4549
50+ @pytest .mark .django_db (transaction = True )
4651@pytest .mark .asyncio
47- async def test_async_authenticated_user_access ():
52+ async def test_async_authenticated_user_access (asgi_app ):
4853 """
4954 Test that middleware can access an authenticated user in async context.
5055
@@ -91,14 +96,12 @@ def create_session():
9196 session_cookie = await create_session ()
9297
9398 if not session_cookie :
94- print ("⚠ Warning: Could not create authenticated session, skipping auth test" )
95- return
99+ pytest .skip ("Could not create authenticated session" )
96100
97101 # Make request with session cookie - this should trigger the bug in v6.7.11
98102 # Disable exception capture to see the SynchronousOnlyOperation clearly
99103 with override_settings (POSTHOG_MW_CAPTURE_EXCEPTIONS = False ):
100- app = get_asgi_application ()
101- async with AsyncClient (transport = ASGITransport (app = app ), base_url = "http://testserver" ) as ac :
104+ async with AsyncClient (transport = ASGITransport (app = asgi_app ), base_url = "http://testserver" ) as ac :
102105 response = await ac .get (
103106 "/test/async-user" ,
104107 cookies = {"sessionid" : session_cookie .value }
@@ -108,28 +111,25 @@ def create_session():
108111 data = response .json ()
109112 assert data ["status" ] == "success"
110113 assert data ["user_authenticated" ] == True
111- print (f"✓ Async authenticated user access test passed: { data ['message' ]} " )
112114
113115
114116@pytest .mark .asyncio
115- async def test_sync_user_access ():
117+ async def test_sync_user_access (asgi_app ):
116118 """
117119 Test that middleware works with sync views.
118120
119121 This should always work regardless of middleware version.
120122 """
121- app = get_asgi_application ()
122- async with AsyncClient (transport = ASGITransport (app = app ), base_url = "http://testserver" ) as ac :
123+ async with AsyncClient (transport = ASGITransport (app = asgi_app ), base_url = "http://testserver" ) as ac :
123124 response = await ac .get ("/test/sync-user" )
124125
125126 assert response .status_code == 200
126127 data = response .json ()
127128 assert data ["status" ] == "success"
128- print (f"✓ Sync user access test passed: { data ['message' ]} " )
129129
130130
131131@pytest .mark .asyncio
132- async def test_async_exception_capture ():
132+ async def test_async_exception_capture (asgi_app ):
133133 """
134134 Test that middleware handles exceptions from async views.
135135
@@ -138,59 +138,23 @@ async def test_async_exception_capture():
138138 causes a 500 response. See test_exception_capture.py for tests that verify
139139 actual exception capture to PostHog.
140140 """
141- app = get_asgi_application ()
142- async with AsyncClient (transport = ASGITransport (app = app ), base_url = "http://testserver" ) as ac :
141+ async with AsyncClient (transport = ASGITransport (app = asgi_app ), base_url = "http://testserver" ) as ac :
143142 response = await ac .get ("/test/async-exception" )
144143
145144 # Django returns 500 for unhandled exceptions
146145 assert response .status_code == 500
147- print ("✓ Async exception raises 500 (captured via process_exception)" )
148146
149147
150148@pytest .mark .asyncio
151- async def test_sync_exception_capture ():
149+ async def test_sync_exception_capture (asgi_app ):
152150 """
153151 Test that middleware handles exceptions from sync views.
154152
155153 The middleware's process_exception() method captures view exceptions to PostHog.
156154 This test verifies the exception causes a 500 response.
157155 """
158- app = get_asgi_application ()
159- async with AsyncClient (transport = ASGITransport (app = app ), base_url = "http://testserver" ) as ac :
156+ async with AsyncClient (transport = ASGITransport (app = asgi_app ), base_url = "http://testserver" ) as ac :
160157 response = await ac .get ("/test/sync-exception" )
161158
162159 # Django returns 500 for unhandled exceptions
163160 assert response .status_code == 500
164- print ("✓ Sync exception raises 500 (captured via process_exception)" )
165-
166-
167- if __name__ == "__main__" :
168- """Run tests directly with asyncio for quick testing."""
169- import asyncio
170-
171- async def run_all_tests ():
172- print ("\n Running PostHog Django middleware tests...\n " )
173-
174- try :
175- await test_async_user_access ()
176- except Exception as e :
177- print (f"✗ Async user access test failed: { e } " )
178-
179- try :
180- await test_sync_user_access ()
181- except Exception as e :
182- print (f"✗ Sync user access test failed: { e } " )
183-
184- try :
185- await test_async_exception_capture ()
186- except Exception as e :
187- print (f"✗ Async exception capture test failed: { e } " )
188-
189- try :
190- await test_sync_exception_capture ()
191- except Exception as e :
192- print (f"✗ Sync exception capture test failed: { e } " )
193-
194- print ("\n All tests completed!\n " )
195-
196- asyncio .run (run_all_tests ())
0 commit comments