Merge pull request #170 from UNLV-CS472-672/unit-tests-for-chat

PR 6: Add WebSocket and JWT Middleware Unit Tests for Chat Functionality

Author: ashley-arellano

.github/workflows/django_ci_cd.yml

@@ -53,7 +53,9 @@ jobs:
 
       # Generate coverage XML report for Codecov
       - name: Run tests and generate coverage XML report
-        run: pytest --cov --cov-branch --cov-report=xml
+        run: |
+          coverage erase
+          pytest --cov --cov-branch --cov-report=xml
 
       # Upload to Codecov (external visualization tool)
       - name: Upload coverage reports to Codecov

backend/apps/chat/consumers.py

@@ -35,22 +35,30 @@ async def receive(self, text_data):
         if 'message' in text_data_json:
             message = text_data_json['message']
 
-            # Create a new message object
-            username, id, timestamp = await self.save_message(self.user_id, self.room_name, message)
-
-            # Serialize timestamp
-            formatted = timestamp.isoformat()
-
-            await self.channel_layer.group_send( # sends message to all users in a channel group
-                self.room_group_name,
-                {
-                    'type': 'chat_message', # Calls chat_message()
-                    'message': message,
-                    'username': username,
-                    'id': id,
-                    'timestamp': formatted
-                }
-            )
+            try:
+                # Try to save the message
+                username, id, timestamp = await self.save_message(self.user_id, self.room_name, message)
+
+                # Serialize timestamp
+                formatted = timestamp.isoformat()
+
+                await self.channel_layer.group_send( # sends message to all users in a channel group
+                    self.room_group_name,
+                    {
+                        'type': 'chat_message', # Calls chat_message()
+                        'message': message,
+                        'username': username,
+                        'id': id,
+                        'timestamp': formatted
+                    }
+                )
+            except Chat.DoesNotExist:
+                # Handle the error: chat room does not exist
+                print(f"Chat room '{self.room_name}' not found. Message not saved.")
+                # Send an error message back to the user
+                await self.send(text_data=json.dumps({
+                    "error": f"Chat room '{self.room_name}' not found. Message not saved."
+                }))
 
         # Typing status is received
         elif 'typing' in text_data_json:
@@ -130,11 +138,7 @@ def get_username(self, id):
 
     @sync_to_async
     def save_message(self, id, room_name, content):
-        try:
-            user = User.objects.get(id=id)
-            chat = Chat.objects.get(name = room_name)
-            message = Message.objects.create(content=content, chat = chat, sender = user) # Creates and saves message
-            return user.username, message.id, message.timestamp
-        except (Chat.DoesNotExist):
-            print(f"Chat room '{room_name}' not found. Message not saved.")
-            return "Unknown"
+        user = User.objects.get(id=id)
+        chat = Chat.objects.get(name=room_name)
+        message = Message.objects.create(content=content, chat=chat, sender=user)  # Creates and saves message
+        return user.username, message.id, message.timestamp

backend/apps/chat/test_consumers.py

@@ -0,0 +1,142 @@
+from django.contrib.auth.models import User
+from .models import Chat, Message
+from django.contrib.auth import get_user_model
+from .consumers import ChatConsumer
+from channels.testing import WebsocketCommunicator
+from django.conf import settings
+from backend.asgi import application
+from apps.users.models import Profile
+from rest_framework_simplejwt.tokens import RefreshToken
+import asyncio
+import pytest
+import json
+
+@pytest.mark.asyncio
+@pytest.mark.django_db(transaction=True)
+class TestChatWebsockets:
+    @pytest.fixture(autouse=True)
+    def setup(self, db):
+        self.user1 = get_user_model().objects.create_user(
+            username='testuser1',
+            password='password',
+            first_name='John',
+            last_name='Doe',
+            email='testuser1@example.com'
+        )
+        self.profile1 = Profile.objects.create(user=self.user1, role=2)
+        self.token1 = self.get_jwt_token(self.user1)
+
+        self.user2 = get_user_model().objects.create_user(
+            username='testuser2',
+            password='password',
+            first_name='Jane',
+            last_name='Doe',
+            email='testuser2@example.com'
+        )
+        self.profile2 = Profile.objects.create(user=self.user2, role=2)
+        self.token2 = self.get_jwt_token(self.user2)
+
+        # Create test chat
+        self.chat, _ = Chat.objects.get_or_create_chat(self.user1, self.user2)
+        self.chat.name = "room1"
+        self.chat.save()
+
+        # Create test messages
+        self.message1 = Message.objects.create(
+            chat=self.chat,
+            sender=self.user1,
+            content="Test message 1 for chat",
+            status=Message.NOT_SEEN
+        )
+        self.message2 = Message.objects.create(
+            chat=self.chat,
+            sender=self.user1,
+            content="Test message 2 for chat",
+            status=Message.NOT_SEEN
+        )
+
+    def get_jwt_token(self, user):
+        refresh = RefreshToken.for_user(user)
+        return str(refresh.access_token)
+
+    def setup_communicator(self, room_name, token):
+        return WebsocketCommunicator(
+            application,
+            f"/ws/apps/chat/{room_name}/",
+            subprotocols=["chat", token],
+        )
+
+    async def test_websocket_connection(self):
+        communicator = self.setup_communicator("room1", self.token1)
+        connected, _ = await communicator.connect()
+        assert connected
+
+        await asyncio.sleep(0.1)
+
+        await communicator.send_to(text_data=json.dumps({"message": "ping"}))
+        response = await communicator.receive_from()
+        data = json.loads(response)
+
+        assert data["message"] == "successful"
+        assert data["body"] == "ping"
+
+        await communicator.disconnect()
+
+    async def test_websocket_error_when_room_missing(self):
+        # Pick a room name you never created in your fixtures or setup
+        missing_room = "i_dont_exist"
+
+        # Set up communicator exactly the same way
+        communicator = self.setup_communicator(missing_room, self.token1)
+        connected, _ = await communicator.connect()
+        assert connected
+
+        # Give the consumer a split-second to finish any setup
+        await asyncio.sleep(0.1)
+
+        # Send a normal chat payload
+        await communicator.send_to(text_data=json.dumps({"message": "hello"}))
+
+        # The consumer should catch Chat.DoesNotExist and send back an 'error' field
+        response = await communicator.receive_from()
+        data = json.loads(response)
+
+        assert "error" in data, "Expected an 'error' key when room is missing"
+        assert data["error"] == f"Chat room '{missing_room}' not found. Message not saved."
+
+        await communicator.disconnect()
+
+    async def test_websocket_typing_status(self):
+        communicator = self.setup_communicator("room1", self.token1)
+        connected, _ = await communicator.connect()
+        assert connected
+
+        await asyncio.sleep(0.1)
+
+        await communicator.send_to(text_data=json.dumps({"typing": True}))
+        response = await communicator.receive_from()
+        data = json.loads(response)
+
+        assert data["message"] == "successful"
+        assert data["typing"] is True
+        assert data["username"] == self.user1.username
+
+        await communicator.disconnect()
+
+    async def test_websocket_seen_status(self):
+        communicator = self.setup_communicator("room1", self.token2) # User who sees the text
+        connected, _ = await communicator.connect()
+        assert connected
+
+        await asyncio.sleep(0.1)
+
+        seen_ids = [self.message1.id, self.message2.id]
+        await communicator.send_to(text_data=json.dumps({"seen": seen_ids}))
+        response = await communicator.receive_from()
+        data = json.loads(response)
+
+        assert data["message"] == "seen_successful"
+        assert sorted(data["ids"]) == sorted(seen_ids)
+        assert data["username"] == self.user1.username
+
+        await communicator.disconnect()

backend/apps/chat/test_jwt_middleware.py

@@ -0,0 +1,113 @@
+import pytest
+import asyncio
+from django.contrib.auth import get_user_model
+from rest_framework_simplejwt.tokens import AccessToken
+from channels.db import database_sync_to_async
+from channels.testing import ApplicationCommunicator
+from .jwt_middleware import JWTAuthMiddleware
+from apps.users.models import Profile
+from rest_framework_simplejwt.tokens import RefreshToken
+
+User = get_user_model()
+
+class DummyASGIApp:
+    async def __call__(self, scope, receive, send):
+        await send({
+            'type': 'websocket.scope',
+            'scope': scope
+        })
+
+
+@pytest.mark.asyncio
+class TestJWTAuthMiddleware:
+
+    @pytest.fixture(autouse=True)
+    def setup(self, db):
+        self.user = get_user_model().objects.create_user(
+            username='testuser',
+            password='password',
+            first_name='John',
+            last_name='Doe',
+            email='testuser1@example.com'
+        )
+        self.profile = Profile.objects.create(user=self.user, role=2)
+        self.token = self.get_jwt_token(self.user)
+
+    def get_jwt_token(self, user):
+        refresh = RefreshToken.for_user(user)
+        return str(refresh.access_token)
+
+    async def test_jwt_auth_middleware_valid_token(self):
+        scope = {
+            "type": "websocket",
+            "subprotocols": ["chat", self.token],
+        }
+
+        middleware = JWTAuthMiddleware(DummyASGIApp())
+        communicator = ApplicationCommunicator(middleware, scope)
+        await communicator.send_input({'type': 'websocket.connect'})
+        await communicator.wait()
+        response = await communicator.receive_output()
+
+        assert response["scope"]["user_id"] == self.user.id
+        assert "error" not in response["scope"]
+
+    async def test_jwt_auth_middleware_invalid_token(self):
+        scope = {
+            "type": "websocket",
+            "subprotocols": ["chat", "invalid.token.here"],
+        }
+
+        middleware = JWTAuthMiddleware(DummyASGIApp())
+        communicator = ApplicationCommunicator(middleware, scope)
+        await communicator.send_input({'type': 'websocket.connect'})
+        await communicator.wait()
+        response = await communicator.receive_output()
+
+        assert response["scope"].get("error") == "Invalid token"
+        assert "user_id" not in response["scope"]
+
+    async def test_jwt_auth_middleware_missing_token(self):
+        scope = {
+            "type": "websocket",
+            "subprotocols": ["chat"],
+        }
+
+        middleware = JWTAuthMiddleware(DummyASGIApp())
+        communicator = ApplicationCommunicator(middleware, scope)
+        await communicator.send_input({'type': 'websocket.connect'})
+        await communicator.wait()
+        response = await communicator.receive_output()
+
+        assert response["scope"].get("error") == "Provide an auth token"
+        assert "user_id" not in response["scope"]
+
+    async def test_jwt_auth_middleware_single_subprotocol(self):
+        scope = {
+            "type": "websocket",
+            "subprotocols": ["chat"],  # Only 1 subprotocol, should trigger missing token
+        }
+
+        middleware = JWTAuthMiddleware(DummyASGIApp())
+        communicator = ApplicationCommunicator(middleware, scope)
+        await communicator.send_input({'type': 'websocket.connect'})
+        await communicator.wait()
+        response = await communicator.receive_output()
+
+        assert response["scope"].get("error") == "Provide an auth token"
+
+    async def test_jwt_auth_middleware_token_raises_exception(self):
+        # Pass a token that will raise a decoding error
+        broken_token = "a.b.c"  # This will raise a DecodeError
+        scope = {
+            "type": "websocket",
+            "subprotocols": ["chat", broken_token],
+        }
+
+        middleware = JWTAuthMiddleware(DummyASGIApp())
+        communicator = ApplicationCommunicator(middleware, scope)
+        await communicator.send_input({'type': 'websocket.connect'})
+        await communicator.wait()
+        response = await communicator.receive_output()
+
+        assert response["scope"].get("error") == "Invalid token"

backend/apps/submissions/tests.py

@@ -60,7 +60,7 @@ def setUp(self):
 
         url = reverse('token_obtain_pair')  # Update to the correct token URL if necessary
         response = self.client.post(url, {'username': 'student1', 'password': 'password'}, format='json')
-        self.token = response.data['access'] 
+        self.token = response.data['access']
 
 
     ### Testing Models ###

pytest.ini

@@ -1,4 +1,4 @@
 # pytest.ini
 [pytest]
 DJANGO_SETTINGS_MODULE = backend.settings
-python_files = tests.py
+python_files = test_*.py *_tests.py tests.py