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.
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
Authentication for testing Communicators #903
Comments
There's nothing easy right now, unfortunately. I'll transform this ticket into a feature request for such. |
Currently I'm using something like this: from django.test import TestCase
from channels.testing import WebsocketCommunicator
from project.routing import application
class WebsocketTestCase(TestCase):
@async_test
async def test_auth(self):
user = User.objects.create_user(**user_kwargs)
self.client.login(username=user.username, password=password)
headers = [(b'origin', b'...'), (b'cookie', self.client.cookies.output(header='', sep='; ').encode())]
communicator = WebsocketCommunicator(application, '/endpoint/', headers)
connected, _ = await communicator.connect()
self.assertTrue(connected)
self.assertEquals(communicator.instance.scope['user'], user)
await communicator.disconnect() Basically you take the main ASGI app instance and connect to a particular endpoint with the right headers. Then you can send/receive data. Endpoint should be wrapped with |
@andrewgodwin thanks for the prompt answer and considering that |
Here is a version for Note: Transactions don't seem to work in this async setting. As a result all modifications to the database will remain at the end of a test. from importlib import import_module
from channels.db import database_sync_to_async
from django.conf import settings
from django.http import HttpRequest, SimpleCookie
def _login(user, backend=None):
from django.contrib.auth import login
engine = import_module(settings.SESSION_ENGINE)
# Create a fake request to store login details.
request = HttpRequest()
request.session = engine.SessionStore()
login(request, user, backend)
# Save the session values.
request.session.save()
# Create a cookie to represent the session.
session_cookie = settings.SESSION_COOKIE_NAME
cookies = SimpleCookie()
cookies[session_cookie] = request.session.session_key
cookie_data = {
'max-age': None,
'path': '/',
'domain': settings.SESSION_COOKIE_DOMAIN,
'secure': settings.SESSION_COOKIE_SECURE or None,
'expires': None,
}
cookies[session_cookie].update(cookie_data)
return cookies
@database_sync_to_async
def login(**credentials):
from django.contrib.auth import authenticate
user = authenticate(**credentials)
if user:
return _login(user)
else:
return SimpleCookie()
@database_sync_to_async
def force_login(user, backend=None):
def get_backend():
from django.contrib.auth import load_backend
for backend_path in settings.AUTHENTICATION_BACKENDS:
backend = load_backend(backend_path)
if hasattr(backend, 'get_user'):
return backend_path
if backend is None:
backend = get_backend()
user.backend = backend
return _login(user, backend)
# I'm not sure if this will work
@database_sync_to_async
def logout(cookies):
"""Log out the user by removing the cookies and session object."""
from django.contrib.auth import logout
engine = import_module(settings.SESSION_ENGINE)
session_cookie = cookies.get(settings.SESSION_COOKIE_NAME)
request = HttpRequest()
session_key = # get the session key from the cookie
request.session = engine.SessionStore(session_key)
logout(request)
return SimpleCookie() Usage: Thanks to @laevilgenius for providing the import pytest
from channels.auth import AuthMiddlewareStack
from channels.testing import WebsocketCommunicator
pytestmark = pytest.mark.asyncio
async def test_authentication(user):
cookies = await force_login(user)
headers = [(b'cookie', cookies.output(header='', sep='; ').encode())]
communicator = WebsocketCommunicator(AuthMiddlewareStack(my_consumer), 'my/path/', headers)
# ...
await logout(cookies) |
If you mean that test cases are not automatically wrapped in a transaction, then yes, that is correct (that functionality comes from Django's |
Yes, this is what I meant. I didn't mention it above but I use Here I currently get my user object from: @database_sync_to_async
def create_user():
return get_user_model().objects.create_user('test-user', 'pw')
@pytest.fixture
async def user(db):
return await create_user() |
@andrewgodwin How should this be done?
|
It can't mutate the scope, so it would have to be an argument to the constructor; probably just |
This is working in my project so far: class AuthWebsocketCommunicator(WebsocketCommunicator):
async def __new__(cls, *args, **kwargs):
instance = super().__new__(cls)
await instance.__init__(*args, **kwargs)
return instance
async def __init__(self, application, path, headers=None, subprotocols=None, user=None):
if user is not None:
await self.force_login(user)
cookie_header = (b'cookie', self._session_cookie)
if headers:
index = None
for i, header in enumerate(headers):
if header[0] == cookie_header[0]:
cookie_index = i
break
if index is None:
headers.append(cookie_header)
else:
headers[index] = (
cookie_header[0],
b'; '.join((cookie_header[1], headers[index][1]))
)
else:
headers = [cookie_header]
super().__init__(application, path, headers, subprotocols) ( communicator = await AuthWebsocketCommunicator(AuthMiddlewareStack(my_consumer), 'my/path/', user=user) Is this the way to go for the communicators? I'd subclass However, SQLite can’t support this high level of concurrency, e.g. when you log out in the test: Setting Solutions are to use another session engine, e.g. What about the other methods like |
Well, first off, don't override the constructor to be async like that is my suggestion - that's just going to confuse a lot of people (as by and large, even async classes have sync Secondly, the implementation is a bit strange - rather than running through the cookie layer, I would probably just add the user to the scope directly, as |
Hi all, Thanks to your comment yesterday and @dgilge I got this to work for me.
when creating an AuthWebsocketCommunicator, I simply pass in a user that I logged in using the django test client. seems to be working so far! |
@mapes911 Thanks for sharing this. However, I thought to do the login within the communicator class because the communicator should be kind of a replacement for the @andrewgodwin class AuthCommunicator(ApplicationCommunicator):
# ...
async def __init__(self, application, path, headers=None, subprotocols=None, user=None):
if user is not None:
session = await self.force_login(user)
# Scope should be set already from WebsocketCommunicator or HttpCommunicator
self.scope.update({"user": user, "session": session})
super().__init__(application, self.scope)
async def force_login(self, user, backend=None):
from django.contrib.auth import login
engine = import_module(settings.SESSION_ENGINE)
request = HttpRequest()
request.session = engine.SessionStore()
await database_sync_to_async(login)(request, user, backend)
await database_sync_to_async(request.session.save)()
return request.session Or should the login and logout stuff be seperate functions (no methods)? In this case we could store the session as |
no, i don think, you shouldn't need to |
You'll notice that nothing async happens in the constructor, it happens in |
@mapes911 You could also set it outside constructor. I'm using this workaround to set authenticated user: communicator = WebsocketCommunicator(Consumer, "/")
communicator.scope['user'] = user Feels a bit fragile, but it works. Maybe the auth provider could be simple wrapper/middleware around communicator instead of replacing communicator with different class? |
The methods in these comments don't currently seem to work - setting the communicator's scope after construction does nothing because it is copied into the application's scope in the constructor. Passing cookies also does nothing. I can subclass the communicator to do this, at which point It is difficult to separate what problems are due to this though and what are due to other issues; I cannot use pytest. |
Also running into this problem right now. @fish-face, I managed to get things working by setting the scope on the application instance inside the communicator rather than on the communicator itself. communicator = WebsocketCommunicator(consumer, "game/{0}/team/".format(game_id))
communicator.instance.scope["user"] = host
connected, subprotocol = await communicator.connect()
assert connected |
Part of troubleshooting and fixing EDD-1318, where WebSocket connections to edd-test.jbei.org are not properly authenticating, and are immediately closed. The tests are replacing the previous, hacky way to "login" by specially crafting a scope object sent to the Consumer class. Instead, use the Django Test Client to do a force_login() call with the test user, and then initialize the WebsocketCommunicator object with the HTTP headers that a real client would send along -- particularly the cookies. This is similar to the approach described in this GitHub issue comment in the channels project: django/channels#903 (comment) Also taking the opportunity to refactor a handful of scattered calls to the sync_to_async decorator. Each call now has its own async alias function, so the Consumer code reads more like typical async code.
I was wondering how it is possible to do authentication with the
Communicator
objects (e.g.WebsocketCommunicator
while testing (similar toclient.force_login()
for channels 1.x). Nothing is mentioned in the documentation at the moment. Is this available (maybe I missed it somehow) now or planned for the future?Great job on channels 2.0! :)
The text was updated successfully, but these errors were encountered: