Authentication for testing Communicators

[akittas]

I was wondering how it is possible to do authentication with the Communicator objects (e.g. WebsocketCommunicator while testing (similar to client.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! :)

[andrewgodwin]

There's nothing easy right now, unfortunately. I'll transform this ticket into a feature request for such.

[andriilahuta]

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 AuthMiddlewareStack of course.

[akittas]

@andrewgodwin thanks for the prompt answer and considering that @laevilgenius many thanks for that! I will definitely try it

[dgilge]

Here is a version for py.test without TestCase that works for me so far. (It's adjusted from django.test.Client.)

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 headers part.

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)

[andrewgodwin]

Transactions don't seem to work in this async setting

If you mean that test cases are not automatically wrapped in a transaction, then yes, that is correct (that functionality comes from Django's TestCase). There's probably another feature to add separately here which is to provide an async-compatible transaction wrapper for tests, but that's quite tricky.

[dgilge]

Yes, this is what I meant. pytest-django wrappes tests cases into transactions if you use the db fixture or the django_db mark. But it has no effect (in the current version). (pytest-django uses parts of Django's TestCase to accomplish it.)

I didn't mention it above but I use pytest-django database fixtures and you will have to if you use database-backed sessions (and don't have another way to get the database running) as far as I understand.

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()

[dgilge]

@andrewgodwin How should this be done?

• Let the __init__ method have another kwarg login_user where you can pass a user object and the login will be done like in my post above? (I think we don't implement that.)

• Convert my functions login logout and force_login to WebsocketCommunitator methods you can use like

  communicator = WebsocketCommunicator(my_consumer, '/')
  await communicator.force_login(my_user)
  await communicator.connect()
  # ...
  await communicator.logout()
  # ...

  The methods do the stuff shown in my post above (I'd make sure that they send the signals user_logged_in, user_logged_out and user_login_failed) and that's it. (The scopes' user objects will not be modifed, particularly is_authenticated – because this reflects reality, doesn't it?) I'd store the session as _session class attribute to be able to perform a logout.

[andrewgodwin]

It can't mutate the scope, so it would have to be an argument to the constructor; probably just user=.

[dgilge]

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)

(force_login is similar to my post above.) As a result you have to await the communicater:

communicator = await AuthWebsocketCommunicator(AuthMiddlewareStack(my_consumer), 'my/path/', user=user)

Is this the way to go for the communicators? I'd subclass ApplicationCommunicator and let HttpCommunicator and WebsocketCommunicator inherit from it.

However, SQLite can’t support this high level of concurrency, e.g. when you log out in the test: django.db.utils.OperationalError: database table is locked: django_session

Setting 'timeout': 60 didn't help, at least in my case.

Solutions are to use another session engine, e.g. SESSION_ENGINE = 'django.contrib.sessions.backends.cache' or another database engine.

What about the other methods like logout? They won't mutate the scope but update the session and send signals.

[andrewgodwin]

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 __init__).

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 AuthMiddleware will see it's already there and let it through. You'd probably need to add a throwaway session in there as well.

[mapes911]

Hi all, @andrewgodwin thanks for writing Channels. amazing stuff :)

Thanks to your comment yesterday and @dgilge I got this to work for me.

from channels.testing import WebsocketCommunicator

class AuthWebsocketCommunicator(WebsocketCommunicator):
    def __init__(self, application, path, headers=None, subprotocols=None, user=None):
        super(AuthWebsocketCommunicator, self).__init__(application, path, headers, subprotocols)
        if user is not None:
            self.scope['user'] = user

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!

[dgilge]

@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 django.test.Client.

@andrewgodwin I didn't think of just putting the user in the scope, sorry. The other point: I also don't like the async constructer. But I'm afraid I don't know how to do it without because you might be using the ORM for logging in (depending on the backend) and you should use await database_sync_to_async in these cases, shouldn't you?

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 user.session attribute.

[hishnash]

no, i don think, you shouldn't need to async on the Init since this is being called directly from the py.test and at this level, py.test is not going to have more than one core-routine running. (sure if you use AuthCommunicator somewhere else that is not testing this would be an issue)

[andrewgodwin]

You'll notice that nothing async happens in the constructor, it happens in get_response (including sending the request to the backend), including sending the request to the app in the first place. That's where you can do ORM stuff if you need.

[tricoder42]

@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?

[fish-face]

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 force_login seems to do nothing.

It is difficult to separate what problems are due to this though and what are due to other issues; I cannot use pytest.

[jtimmons]

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

[bramirez-ads]

The last solution written by @jtimmons is no longer viable, given that the instance field was removed not too shortly after that comment was made.

I discovered a less invasive solution:

import functools
from asgiref.sync import sync_to_async
from channels.testing import WebsocketCommunicator

class AuthWebsocketCommunicator(WebsocketCommunicator):
    def __init__(self, application, path, user, *args, **kwargs):
        super().__init__(self._asgi_with_user(application, user), path,
                         *args, **kwargs)

    @classmethod
    def _asgi_with_user(cls, asgi_app, user):
        """
        Update the scope of an ASGI app such that a particular user
        is already assumed to have been authenticated.
        """
        async def app(scope, receive, send):
            scope['user'] = user
            return await asgi_app(scope, receive, send)
        functools.update_wrapper(app, asgi_app)
        return app

Here's a sample test, which also shows how to pass AllowedHostsOriginValidator if you've included it in your router:

from django.contrib.auth.models import User
from django.test import TestCase

from myapp.asgi import application

class EndpointTest(TestCase):
    async def test_endpoint_can_connect(self):
        user = await sync_to_async(User.objects.create_user)(
            username='foo',
            email='foo@example.com',
        )
        headers = [(b'origin', b'http://localhost:80')] # Bypass AllowedHostsOriginValidator
        communicator = AuthWebsocketCommunicator(application, '/path/to/endpoint', user,
                                                 headers=headers)
        connected, _subprotocol = await communicator.connect()
        self.assertTrue(connected)
        await communicator.disconnect()