Reputation: 1
I'm building a simple app where users can join tables to play chess, connecting via WebSocket. Everything works as intended in the development environment: users can log in, join rooms, sit at tables, and move pieces without any WebSocket connection issues.
However, when I try to write a unit test using TransactionTestCase, the connection attempt fails. Specifically, the connected value in connected, subprotocol = await communicator.connect() returns False, even though there are no problems in the development environment.
Here's a simplified version of my setup:
WebSocket Consumer (consumers.py):
class TableConsumer(AsyncWebsocketConsumer):
async def connect(self):
self.table_id = self.scope["url_route"]["kwargs"]["table_id"]
self.table_group_id = f"table_{self.table_id}"
# Join the room group
await self.channel_layer.group_add(self.table_group_id, self.channel_name)
await self.accept()
# Send the current game state to the new connection
await self.send_current_state()
Unit Test (tests.py):
from channels.testing import WebsocketCommunicator
from django.test import TransactionTestCase, SimpleTestCase, TestCase
from django.conf import settings
from chess_django.asgi import application
from .models import Game, Board
from asgiref.sync import sync_to_async
class ChannelLayerSettingsTest(TestCase):
def test_channel_layers_for_testing(self):
self.assertEqual(
settings.CHANNEL_LAYERS['default']['BACKEND'],
"channels.layers.InMemoryChannelLayer"
)
class WebSocketTests(TransactionTestCase):
async def test_websocket_connect(self):
new_game = await sync_to_async(Game.objects.create)()
communicator = WebsocketCommunicator(application, f"/ws/table/{new_game.id}/")
connected, subprotocol = await communicator.connect()
self.assertTrue(connected)
await communicator.disconnect()
ASGI Configuration (asgi.py):
import os
from channels.auth import AuthMiddlewareStack
from channels.routing import ProtocolTypeRouter, URLRouter
from channels.security.websocket import AllowedHostsOriginValidator
from django.core.asgi import get_asgi_application
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'chess_django.settings')
django_asgi_app = get_asgi_application()
import table.routing
application = ProtocolTypeRouter(
{
"http": get_asgi_application(),
"websocket": AllowedHostsOriginValidator(
AuthMiddlewareStack(URLRouter(table.routing.websocket_urlpatterns))
),
}
)
Routing (routing.py):
from django.urls import re_path
from . import consumers
websocket_urlpatterns = [
re_path(r"ws/table/(?P<table_id>\w+)/$", consumers.TableConsumer.as_asgi()),
]
Settings (settings.py):
INSTALLED_APPS = [
'daphne',
'channels',
'table',
'menu',
'django.contrib.admin',
'django.contrib.auth',
'django.contrib.contenttypes',
'django.contrib.sessions',
'django.contrib.messages',
'django.contrib.staticfiles',
]
#...
ASGI_APPLICATION = "chess_django.asgi.application"
if "test" in sys.argv:
CHANNEL_LAYERS = {
"default": {
"BACKEND": "channels.layers.InMemoryChannelLayer",
},
}
else:
CHANNEL_LAYERS = {
"default": {
"BACKEND": "channels_redis.core.RedisChannelLayer",
"CONFIG": {
"hosts": [("127.0.0.1", 6379)],
},
},
}
Issue: When running the test, the connection fails, with connected being False. This issue does not occur in the development environment where everything functions correctly. I tried without mocking CHANNEL_LAYERS (using default channels_redis), and the test still failed.
Here is a link https://github.com/Przekazmierczak/chess_not_finished to my GitHub repository with the full project setup.
What could be causing this issue in the test environment, and how can I fix it?
Any help would be greatly appreciated!
Upvotes: 0
Views: 71