1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
|
from importlib import import_module
from unittest import mock
import pytest
from asgiref.sync import sync_to_async
from django.conf import settings
from django.contrib.auth import (
BACKEND_SESSION_KEY,
HASH_SESSION_KEY,
SESSION_KEY,
get_user_model,
user_logged_in,
user_logged_out,
)
from django.contrib.auth.models import AnonymousUser
from channels.auth import get_user, login, logout
from channels.db import database_sync_to_async
class CatchSignal:
"""
Capture (and detect) a django signal event.
This should be used as a Contextmanager.
:Example:
with CatchSignal(user_logged_in) as handler:
# do the django action here that will create the signal
assert handler.called
:Async Example:
async with CatchSignal(user_logged_in) as handler:
await ... # the django action the creates the signal
assert handler.called
"""
def __init__(self, signal):
self.handler = mock.Mock()
self.signal = signal
async def __aenter__(self):
await sync_to_async(self.signal.connect)(self.handler)
return self.handler
async def __aexit__(self, exc_type, exc, tb):
await sync_to_async(self.signal.disconnect)(self.handler)
def __enter__(self):
self.signal.connect(self.handler)
return self.handler
def __exit__(self, exc_type, exc_val, exc_tb):
self.signal.disconnect(self.handler)
@pytest.fixture
def user_bob():
return get_user_model().objects.create(username="bob", email="bob@example.com")
@pytest.fixture
def user_bill():
return get_user_model().objects.create(username="bill", email="bill@example.com")
@pytest.fixture
def session():
SessionStore = import_module(settings.SESSION_ENGINE).SessionStore
session = SessionStore()
session.create()
return session
async def assert_is_logged_in(scope, user):
"""
Assert that the provided user is logged in to the session contained within
the scope.
"""
assert "user" in scope
assert scope["user"] == user
session = scope["session"]
# logged in!
assert SESSION_KEY in session
assert BACKEND_SESSION_KEY in session
assert HASH_SESSION_KEY in session
assert isinstance(
await get_user(scope), await database_sync_to_async(get_user_model)()
)
assert await get_user(scope) == user
@pytest.mark.django_db(transaction=True)
@pytest.mark.asyncio
async def test_login_no_session_in_scope():
"""
Test to ensure that a `ValueError` is raised if when tying to login a user
to a scope that has no session.
"""
msg = (
"Cannot find session in scope. You should wrap your consumer in "
"SessionMiddleware."
)
with pytest.raises(ValueError, match=msg):
await login(scope={}, user=None)
@pytest.mark.django_db(transaction=True)
@pytest.mark.asyncio
async def test_login_no_user_in_scope(session):
"""
Test the login method to ensure it raises a `ValueError` if no user is
passed and is no user in the scope.
"""
scope = {"session": session}
with pytest.raises(
ValueError,
match="User must be passed as an argument or must be present in the scope.",
):
await login(scope, user=None)
@pytest.mark.django_db(transaction=True)
@pytest.mark.asyncio
async def test_login_user_as_argument(session, user_bob):
"""
Test that one can login to a scope that has a session by passing the scope
and user as arguments to the login function.
"""
scope = {"session": session}
assert isinstance(await get_user(scope), AnonymousUser)
# not logged in
assert SESSION_KEY not in session
async with CatchSignal(user_logged_in) as handler:
assert not handler.called
await login(scope, user=user_bob)
assert handler.called
await assert_is_logged_in(scope, user_bob)
@pytest.mark.django_db(transaction=True)
@pytest.mark.asyncio
async def test_login_user_on_scope(session, user_bob):
"""
Test that in the absence of a user being passed to the `login` function the
function will use the user set on the scope.
"""
scope = {"session": session, "user": user_bob}
# check that we are not logged in on the session
assert isinstance(await get_user(scope), AnonymousUser)
async with CatchSignal(user_logged_in) as handler:
assert not handler.called
await login(scope, user=None)
assert handler.called
await assert_is_logged_in(scope, user_bob)
@pytest.mark.django_db(transaction=True)
@pytest.mark.asyncio
async def test_login_change_user(session, user_bob, user_bill):
"""
Test logging in a second user into a scope were another user is already logged in.
"""
scope = {"session": session}
# check that we are not logged in on the session
assert isinstance(await get_user(scope), AnonymousUser)
async with CatchSignal(user_logged_in) as handler:
assert not handler.called
await login(scope, user=user_bob)
assert handler.called
await assert_is_logged_in(scope, user_bob)
session_key = session[SESSION_KEY]
assert session_key
async with CatchSignal(user_logged_in) as handler:
assert not handler.called
await login(scope, user=user_bill)
assert handler.called
await assert_is_logged_in(scope, user_bill)
assert session_key != session[SESSION_KEY]
@pytest.mark.django_db(transaction=True)
@pytest.mark.asyncio
async def test_logout(session, user_bob):
"""
Test that one can logout a user from a logged in session.
"""
scope = {"session": session}
# check that we are not logged in on the session
assert isinstance(await get_user(scope), AnonymousUser)
async with CatchSignal(user_logged_in) as handler:
assert not handler.called
await login(scope, user=user_bob)
assert handler.called
await assert_is_logged_in(scope, user_bob)
assert SESSION_KEY in session
session_key = session[SESSION_KEY]
assert session_key
async with CatchSignal(user_logged_out) as handler:
assert not handler.called
await logout(scope)
assert handler.called
assert isinstance(await get_user(scope), AnonymousUser)
assert isinstance(scope["user"], AnonymousUser)
assert SESSION_KEY not in session
@pytest.mark.django_db(transaction=True)
@pytest.mark.asyncio
async def test_logout_not_logged_in(session):
"""
Test that the `logout` function does nothing in the case were there is no
user logged in.
"""
scope = {"session": session}
# check that we are not logged in on the session
assert isinstance(await get_user(scope), AnonymousUser)
async with CatchSignal(user_logged_out) as handler:
assert not handler.called
await logout(scope)
assert not handler.called
assert "user" not in scope
assert isinstance(await get_user(scope), AnonymousUser)
|
