Testing Django Async Views and Channels: A Practical Guide

pip install django channels pytest pytest-django pytest-asyncio

# pytest.ini
[pytest]
DJANGO_SETTINGS_MODULE = myproject.settings.test
asyncio_mode = auto

# settings/test.py
INSTALLED_APPS = [
    ...
    'channels',
]

CHANNEL_LAYERS = {
    'default': {
        'BACKEND': 'channels.layers.InMemoryChannelLayer',
    }
}

ASGI_APPLICATION = 'myproject.asgi.application'

# myapp/views.py
import asyncio
from django.http import JsonResponse
from django.views import View
from django.contrib.auth.decorators import login_required
from django.utils.decorators import method_decorator
import httpx


async def health_check(request):
    return JsonResponse({'status': 'ok'})


async def fetch_external_data(request):
    async with httpx.AsyncClient() as client:
        response = await client.get('https://api.example.com/data')
    return JsonResponse(response.json())


@method_decorator(login_required, name='dispatch')
class AsyncArticleView(View):
    async def get(self, request, pk):
        from myapp.models import Article
        try:
            article = await Article.objects.aget(pk=pk)
        except Article.DoesNotExist:
            from django.http import Http404
            raise Http404

        return JsonResponse({
            'id': article.id,
            'title': article.title,
            'content': article.content,
        })

    async def post(self, request):
        import json
        data = json.loads(request.body)
        from myapp.models import Article
        article = await Article.objects.acreate(
            title=data['title'],
            content=data['content'],
            author=request.user,
        )
        return JsonResponse({'id': article.id}, status=201)

# tests/test_async_views.py
import pytest
from django.test import AsyncClient
from django.contrib.auth import get_user_model
from unittest.mock import AsyncMock, patch

User = get_user_model()


@pytest.fixture
def async_client():
    return AsyncClient()


@pytest.fixture
async def user(db):
    return await User.objects.acreate_user(
        username='alice', email='alice@example.com', password='pass'
    )


@pytest.mark.asyncio
@pytest.mark.django_db(transaction=True)
async def test_health_check(async_client):
    response = await async_client.get('/health/')
    assert response.status_code == 200
    assert response.json() == {'status': 'ok'}


@pytest.mark.asyncio
@pytest.mark.django_db(transaction=True)
async def test_article_detail_requires_auth(async_client):
    response = await async_client.get('/api/articles/1/')
    assert response.status_code == 302  # Redirect to login


@pytest.mark.asyncio
@pytest.mark.django_db(transaction=True)
async def test_article_detail_returns_data(async_client, user):
    from myapp.models import Article
    article = await Article.objects.acreate(
        title='Test Article', content='Body', author=user
    )

    await async_client.aforce_login(user)
    response = await async_client.get(f'/api/articles/{article.pk}/')

    assert response.status_code == 200
    data = response.json()
    assert data['title'] == 'Test Article'
    assert data['id'] == article.pk


@pytest.mark.asyncio
@pytest.mark.django_db(transaction=True)
async def test_fetch_external_data_calls_api(async_client):
    mock_response = AsyncMock()
    mock_response.json.return_value = {'key': 'value'}

    with patch('httpx.AsyncClient.get', return_value=mock_response):
        response = await async_client.get('/api/external/')

    assert response.status_code == 200
    assert response.json() == {'key': 'value'}

@pytest.mark.asyncio
@pytest.mark.django_db(transaction=True)
async def test_async_orm_operations(user):
    from myapp.models import Article

    # Create
    article = await Article.objects.acreate(
        title='Async Article', content='...', author=user
    )
    assert article.pk is not None

    # Get
    fetched = await Article.objects.aget(pk=article.pk)
    assert fetched.title == 'Async Article'

    # Filter with async iteration
    titles = []
    async for a in Article.objects.filter(author=user):
        titles.append(a.title)
    assert 'Async Article' in titles

    # Aggregate
    count = await Article.objects.filter(author=user).acount()
    assert count == 1

    # Update
    await Article.objects.filter(pk=article.pk).aupdate(title='Updated')
    await article.arefresh_from_db()
    assert article.title == 'Updated'

    # Delete
    await article.adelete()
    assert not await Article.objects.filter(pk=article.pk).aexists()

# myapp/consumers.py
import json
from channels.generic.websocket import AsyncWebsocketConsumer
from channels.db import database_sync_to_async


class ChatConsumer(AsyncWebsocketConsumer):
    async def connect(self):
        self.room_name = self.scope['url_route']['kwargs']['room_name']
        self.room_group_name = f'chat_{self.room_name}'

        # Join room group
        await self.channel_layer.group_add(
            self.room_group_name,
            self.channel_name,
        )
        await self.accept()

    async def disconnect(self, close_code):
        await self.channel_layer.group_discard(
            self.room_group_name,
            self.channel_name,
        )

    async def receive(self, text_data):
        data = json.loads(text_data)
        message = data['message']
        username = self.scope['user'].username

        # Broadcast to room group
        await self.channel_layer.group_send(
            self.room_group_name,
            {
                'type': 'chat_message',
                'message': message,
                'username': username,
            },
        )

    async def chat_message(self, event):
        await self.send(text_data=json.dumps({
            'message': event['message'],
            'username': event['username'],
        }))

# tests/test_consumers.py
import pytest
import json
from channels.testing import WebsocketCommunicator
from channels.layers import get_channel_layer
from django.contrib.auth import get_user_model
from myproject.asgi import application

User = get_user_model()


@pytest.fixture
async def auth_user(db):
    return await User.objects.acreate_user(
        username='alice', password='pass'
    )


@pytest.mark.asyncio
@pytest.mark.django_db(transaction=True)
async def test_websocket_connect(auth_user):
    communicator = WebsocketCommunicator(
        application, '/ws/chat/testroom/'
    )
    communicator.scope['user'] = auth_user

    connected, subprotocol = await communicator.connect()
    assert connected

    await communicator.disconnect()


@pytest.mark.asyncio
@pytest.mark.django_db(transaction=True)
async def test_websocket_send_receive(auth_user):
    communicator = WebsocketCommunicator(
        application, '/ws/chat/testroom/'
    )
    communicator.scope['user'] = auth_user

    await communicator.connect()

    # Send a message
    await communicator.send_json_to({'message': 'Hello, world!'})

    # Receive the broadcast
    response = await communicator.receive_json_from()
    assert response['message'] == 'Hello, world!'
    assert response['username'] == 'alice'

    await communicator.disconnect()


@pytest.mark.asyncio
@pytest.mark.django_db(transaction=True)
async def test_websocket_broadcasts_to_room(auth_user, db):
    # Connect two clients to the same room
    user2 = await User.objects.acreate_user(username='bob', password='pass')

    comm1 = WebsocketCommunicator(application, '/ws/chat/shared/')
    comm2 = WebsocketCommunicator(application, '/ws/chat/shared/')
    comm1.scope['user'] = auth_user
    comm2.scope['user'] = user2

    await comm1.connect()
    await comm2.connect()

    # Alice sends a message
    await comm1.send_json_to({'message': 'Hi Bob!'})

    # Both clients receive it
    msg1 = await comm1.receive_json_from()
    msg2 = await comm2.receive_json_from()

    assert msg1['message'] == 'Hi Bob!'
    assert msg2['message'] == 'Hi Bob!'

    await comm1.disconnect()
    await comm2.disconnect()


@pytest.mark.asyncio
@pytest.mark.django_db(transaction=True)
async def test_unauthenticated_websocket_rejected():
    from django.contrib.auth.models import AnonymousUser

    communicator = WebsocketCommunicator(application, '/ws/chat/private/')
    communicator.scope['user'] = AnonymousUser()

    connected, code = await communicator.connect()
    assert not connected

# myapp/views.py
import asyncio
from django.http import StreamingHttpResponse


async def event_stream(request):
    async def generate():
        for i in range(3):
            yield f'data: Event {i}\n\n'
            await asyncio.sleep(0.01)

    return StreamingHttpResponse(generate(), content_type='text/event-stream')

@pytest.mark.asyncio
@pytest.mark.django_db(transaction=True)
async def test_sse_stream(async_client):
    response = await async_client.get('/api/events/')
    assert response.status_code == 200
    assert response['Content-Type'] == 'text/event-stream'

    content = b''.join([chunk async for chunk in response.streaming_content])
    assert b'data: Event 0' in content
    assert b'data: Event 2' in content

# Testing Celery tasks in async context
@pytest.mark.asyncio
@pytest.mark.django_db(transaction=True)
async def test_triggers_background_task(async_client, user, mocker):
    mock_task = mocker.patch('myapp.tasks.process_article.delay')

    await async_client.aforce_login(user)
    response = await async_client.post(
        '/api/articles/',
        json.dumps({'title': 'New Post', 'content': 'Body'}),
        content_type='application/json',
    )

    assert response.status_code == 201
    mock_task.assert_called_once()