Django concurrency: can't create data for testing in async function?

[aliscie]

Hello I have this quastion anout chango channels testing do u have answer for it pleases?

• Goal Test async function with test database
• Problem I have async functions for Django channels and when I create data in setUp the data are not reflected in the project!!!
• My tests: note: these tests inside a APITestCase class

class WebsocketTests(APITestCase):
        def setUp(self):
        user1 = User.objects.create(username='Alex',email='[email protected]')
        user2 = User.objects.create(username='Sam',email='[email protected]')
        user3 = User.objects.create(username='Clover',email='[email protected]')

    async def test_connect(self):
        communicator = WebsocketCommunicator(application, 'alerts/')
        connected, subprotocol = await communicator.connect()
        assert connected
        await communicator.disconnect()

@database_sync_to_async
def get_user(querys):
    print(User.objects.count()) #<======= this returns 0 while it should return 3

    try:
        token = parse_qs(querys.decode("utf8"))['token'][0]
        token_data = UntypedToken(token)
        user_id = token_data["user_id"]
    except:
        return 'token is invalid'

    try:
        return User.objects.get(id=user_id)
    except User.DoesNotExist:
        return 'AnonymousUser'
class QueryAuthMiddleware:
    def __init__(self, app):
        self.app = app

    async def __call__(self, scope, receive, send):
        scope['user'] = await get_user(scope["query_string"])
        return await self.app(scope, receive, send)


application = ProtocolTypeRouter({
    "http": django_asgi_app,
    "websocket": QueryAuthMiddleware(
        URLRouter(
            websocket_urlpatterns
        )
    ),
})

#settings.py
DATABASES = {
        'default': {
            'ENGINE': 'django.db.backends.sqlite3',
            'NAME': os.path.join(BASE_DIR, 'db.sqlite3'),
            'OPTIONS': {
                'timeout': 20,  # in seconds
                # see also
                # https://docs.python.org/3.7/library/sqlite3.html#sqlite3.connect
            },
            'TEST': {
                'NAME': os.path.join(BASE_DIR, "db_test.sqlite3"),
            },

        }
    }

• I tride
  1. replace APITestCase with django TestCase
  2. switch to postgres
  3. use from asgiref.sync import sync_to_async
     https://stackoverflow.com/posts/68190400/edit

[hishnash]

@aliscie converted this to a discussion rather than issue.

From my experience testing any async code is best doing in pytest take a look at the test directory in this project for examples

[hishnash]

how are you running the tests?

[aliscie]

I run them in pycharm! does that effect? or pytest ~mydir/test_websocket.py

[hishnash]

can you share the test_websocket.py file?

[aliscie]

import asyncio

import pytest
from asgiref.sync import sync_to_async
from channels.db import database_sync_to_async
from channels.testing import WebsocketCommunicator
from icecream import ic

from Functions.TestClass import TestClass
from Functions.calendar_setup import calendar_setup
from calendars.models import Event
from vytrac_26930.asgi import application
from rest_framework.test import APIClient, APITestCase


class WebsocketTests(APITestCase):
    def setUp(self):
        user, create = User.objects.get_or_create(username='Clover', email='[email protected]', password='password')
        user.is_email_verified = True
        user.is_role_verified = True
        user.is_staff = True
        user.is_superuser = True
        user.save()

        user2, create2 = User.objects.get_or_create(date_joined='2021-05-28T13:30:50.884397Z', username='Alex',
                                                    email='[email protected]', password='password')
        user2.is_email_verified = True
        user2.is_staff = True
        user2.save()
        assert user2.is_superuser == False
        user.save()

        user3, create3 = User.objects.get_or_create(username='Sam', email='[email protected]', password='password',
                                                    is_email_verified=True)
        user3.is_email_verified = True
        user3.save()

        assert user3.is_staff == False
        assert user3.is_superuser == False

        for i in [create, create2, create3]:
            ic('=========== laready exist =========== ') if not i else None

        client = APIClient()
        lg_res = client.post('/users/login/', {'username': 'Clover', 'password': 'password'})
        lg_res2 = client.post('/users/login/', {'username': 'Alex', 'password': 'password'})
        lg_res3 = client.post('/users/login/', {'username': 'Sam', 'password': 'password'})
        token = lg_res.data["access"]
        token2 = lg_res2.data["access"]
        token3 = lg_res3.data["access"]
        res = client.get('/statistics/')
        assert res.status_code == status.HTTP_400_BAD_REQUEST

        client.credentials(HTTP_AUTHORIZATION=f'Bearer {token}')
        client = client

        res = client.get('/statistics/')
        assert res.status_code == status.HTTP_200_OK
        x = [user, user2, user3, token, token2, token3, client]
        self.user = user
        self.user2 = user2
        self.user3 = user3
        self.token = token
        self.token2 = token2
        self.token3 = token3
        self.client = client

    @pytest.mark.django_db(transaction=True)
    @pytest.mark.asyncio
    @pytest.mark.timeout(60)
    async def test_get_singals(self):
        communicator = WebsocketCommunicator(application, f'alerts/?token={self.token}')
        connected, subprotocol = await communicator.connect()
        assert connected
        # res = await communicator.receive_from()
        # ic(res)
        # res = await communicator.receive_from()
        # event = await sync_to_async(Event.objects.create)(created_by=self.user, title='my new event')
        # try:
        #     res = await communicator.receive_from()
        # except asyncio.TimeoutError:
        #     ic({"results": f"timeout error on"})
        # assert f'"object_id": {event.id}' in res
        # ic(res)
        await communicator.disconnect()

[hishnash]

still looks like a django test rather than a native pytest test.

take a look at the tests in here: https://github.com/hishnash/djangochannelsrestframework/tree/master/tests

Pytest will not be running the setup/teardown code correctly. You should instead be using fixtures