Test database not keeping data

[yannik131]

I am unable to access database entries created in the setUp method inside a django TestCase from within a WebSocketConsumer. I have set up a repository at https://github.com/yannik131/channels-test. I have tried both with and without pytest.

Steps to reproduce:

git clone https://github.com/yannik131/channels-test
cd channels-test
python3 -m venv env/
source env/bin/activate
export DJANGO_SETTINGS_MODULE=channels_test.settings
pip3 install -r requirements.txt
python3 manage.py test websocket

Expected behaviour: The User query in the websocket.consumers.Consumer should return the user created in websocket.tests.Tests.

Actual behaviour: The query is empty.

Tested with channels 3.0.4 and django 3.2.10 (see requirements.txt). Python versions 3.7.5 and 3.8.3.

[bayodesegun]

I lost more than a day to this issue! I finally got it working by using a hack: creating my test data using a method decorated with database_sync_to_async, calling this method from each test. Here is an extract from my test code as an example:

"""Chat app websocket tests."""
from django.test import TestCase
from rest_framework_simplejwt.tokens import RefreshToken
from channels.testing import WebsocketCommunicator
from channels.db import database_sync_to_async
from my_proj.asgi import application

from my_apps.overlay.tests import utils
from my_apps.chat.models import ChatSessionMessage


class WebsocketConsumerTestCase(TestCase):

    @database_sync_to_async
    def create_test_data(self):
        self.user = utils.create_user()
        self.session = utils.create_chat_session().uri
        self.token = str(RefreshToken.for_user(self.user).access_token)

    @database_sync_to_async
    def check_chat_session_message(self, number):
        self.assertEquals(number, ChatSessionMessage.objects.count())

    async def test_connect_to_chat__unauthenticated_users_not_accepted(self):
        await self.create_test_data()
        
        # invalid token
        communicator = WebsocketCommunicator(application, "/ws/%s/?token=abczxy" % self.session)
        connected, subprotocol = await communicator.connect()
        self.assertFalse(connected)

        # No token
        communicator = WebsocketCommunicator(application, "/ws/%s/" % self.session)
        connected, subprotocol = await communicator.connect()
        self.assertFalse(connected)

    async def test_connect_to_chat__authenticated_user_accepted(self):
        await self.create_test_data()
        
        # proper authentication
        communicator = WebsocketCommunicator(
            application, "/ws/%s/?token=%s" % (self.session, self.token))
        connected, subprotocol = await communicator.connect()
        self.assertTrue(connected)

        # Close
        await communicator.disconnect()

    async def test_chat_message_is_saved_when_specified(self):
        await self.create_test_data()
        
        communicator = WebsocketCommunicator(
            application, "/ws/%s/?token=%s" % (self.session, self.token))
        connected, subprotocol = await communicator.connect()
        self.assertTrue(connected)

        # Send message, don't store
        json_message = {
            'username': self.user.username,
            'message': 'This is a test message'
        }
        await communicator.send_json_to(json_message)
        response = await communicator.receive_json_from()
        await self.check_chat_session_message(0)

        # Send message, store
        json_message['store'] = True
        await communicator.send_json_to(json_message)
        response = await communicator.receive_json_from()
        await self.check_chat_session_message(1)

        # Close
        await communicator.disconnect()