1
1
import re
2
+ from importlib import import_module
2
3
4
+ import django
3
5
import pytest
6
+ from django .conf import settings
4
7
5
8
from channels .consumer import AsyncConsumer
6
9
from channels .db import database_sync_to_async
@@ -93,15 +96,12 @@ async def test_session_samesite_invalid(samesite_invalid):
93
96
94
97
@pytest .mark .django_db (transaction = True )
95
98
@pytest .mark .asyncio
96
- async def test_muliple_sessions ():
99
+ async def test_multiple_sessions ():
97
100
"""
98
101
Create two application instances and test then out of order to verify that
99
102
separate scopes are used.
100
103
"""
101
104
102
- async def inner (scope , receive , send ):
103
- send (scope ["path" ])
104
-
105
105
class SimpleHttpApp (AsyncConsumer ):
106
106
async def http_request (self , event ):
107
107
await database_sync_to_async (self .scope ["session" ].save )()
@@ -123,3 +123,84 @@ async def http_request(self, event):
123
123
124
124
first_response = await first_communicator .get_response ()
125
125
assert first_response ["body" ] == b"/first/"
126
+
127
+
128
+ @pytest .mark .django_db (transaction = True )
129
+ @pytest .mark .asyncio
130
+ async def test_session_saves ():
131
+ """
132
+ Saves information to a session and validates that it actually saves to the backend
133
+ """
134
+
135
+ class SimpleHttpApp (AsyncConsumer ):
136
+ @database_sync_to_async
137
+ def set_fav_color (self ):
138
+ self .scope ["session" ]["fav_color" ] = "blue"
139
+
140
+ async def http_request (self , event ):
141
+ if django .VERSION >= (5 , 1 ):
142
+ await self .scope ["session" ].aset ("fav_color" , "blue" )
143
+ else :
144
+ await self .set_fav_color ()
145
+ await self .send (
146
+ {"type" : "http.response.start" , "status" : 200 , "headers" : []}
147
+ )
148
+ await self .send (
149
+ {
150
+ "type" : "http.response.body" ,
151
+ "body" : self .scope ["session" ].session_key .encode (),
152
+ }
153
+ )
154
+
155
+ app = SessionMiddlewareStack (SimpleHttpApp .as_asgi ())
156
+
157
+ communicator = HttpCommunicator (app , "GET" , "/first/" )
158
+
159
+ response = await communicator .get_response ()
160
+ session_key = response ["body" ].decode ()
161
+
162
+ SessionStore = import_module (settings .SESSION_ENGINE ).SessionStore
163
+ session = SessionStore (session_key = session_key )
164
+ if django .VERSION >= (5 , 1 ):
165
+ session_fav_color = await session .aget ("fav_color" )
166
+ else :
167
+ session_fav_color = await database_sync_to_async (session .get )("fav_color" )
168
+
169
+ assert session_fav_color == "blue"
170
+
171
+
172
+ @pytest .mark .django_db (transaction = True )
173
+ @pytest .mark .asyncio
174
+ async def test_session_save_update_error ():
175
+ """
176
+ Intentionally deletes the session to ensure that SuspiciousOperation is raised
177
+ """
178
+
179
+ async def inner (scope , receive , send ):
180
+ send (scope ["path" ])
181
+
182
+ class SimpleHttpApp (AsyncConsumer ):
183
+ @database_sync_to_async
184
+ def set_fav_color (self ):
185
+ self .scope ["session" ]["fav_color" ] = "blue"
186
+
187
+ async def http_request (self , event ):
188
+ # Create a session as normal:
189
+ await database_sync_to_async (self .scope ["session" ].save )()
190
+
191
+ # Then simulate it's deletion from somewhere else:
192
+ # (e.g. logging out from another request)
193
+ SessionStore = import_module (settings .SESSION_ENGINE ).SessionStore
194
+ session = SessionStore (session_key = self .scope ["session" ].session_key )
195
+ await database_sync_to_async (session .flush )()
196
+
197
+ await self .send (
198
+ {"type" : "http.response.start" , "status" : 200 , "headers" : []}
199
+ )
200
+
201
+ app = SessionMiddlewareStack (SimpleHttpApp .as_asgi ())
202
+
203
+ communicator = HttpCommunicator (app , "GET" , "/first/" )
204
+
205
+ with pytest .raises (django .core .exceptions .SuspiciousOperation ):
206
+ await communicator .get_response ()
0 commit comments