Merge pull request #190 from UNLV-CS472-672/mute_block_report_tests

PR 7: Refactor Mute/Blocking Logic and Tests for User Moderation System

Author: aviendha-andrus

backend/apps/chat/consumers.py

@@ -3,6 +3,7 @@
 from asgiref.sync import sync_to_async
 from .models import Message, Chat, BlockedUser, MutedUser
 from django.contrib.auth.models import User
+from channels.db import database_sync_to_async
 
 # https://medium.com/@farad.dev/how-to-build-a-real-time-chat-app-using-django-channels-2ba2621ea972
 class ChatConsumer(AsyncWebsocketConsumer):
@@ -28,6 +29,10 @@ async def disconnect(self, close_code):
             self.channel_name
         )
 
+    @database_sync_to_async
+    def user_blocked_or_muted(self, chat, sender):
+        Chat.objects.user_blocked_or_muted(chat, sender)
+
 # This function receive messages from WebSocket.
     async def receive(self, text_data):
         # Message is received
@@ -38,8 +43,7 @@ async def receive(self, text_data):
             try:
                 chat = await sync_to_async(Chat.objects.get)(name=self.room_name)
                 sender = await sync_to_async(User.objects.get)(id=self.user_id)
-                receiver = await sync_to_async(Chat.objects.get_other_user)(chat, sender)
-                await self.blocked_or_muted(sender, receiver)
+                await self.user_blocked_or_muted(chat, sender)
 
                 # Try to save the message
                 username, id, timestamp = await self.save_message(self.user_id, self.room_name, message)
@@ -146,17 +150,4 @@ def save_message(self, id, room_name, content):
         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
-        
-    @staticmethod
-    async def blocked_or_muted(sender, receiver):
-        sender_profile = await sync_to_async(lambda: sender.profile)()
-        receiver_profile = await sync_to_async(lambda: receiver.profile)()
-
-        blocked = await sync_to_async(BlockedUser.objects.filter(blocked_by=receiver_profile, blocked_user=sender_profile).exists)()
-        if blocked:
-            raise PermissionError("You are blocked by this user.")
-
-        muted = await sync_to_async(MutedUser.objects.filter(muted_by=receiver_profile, muted_user=sender_profile).exists)()
-        if muted:
-            raise PermissionError("You are muted by this user.")
+        return user.username, message.id, message.timestamp

backend/apps/chat/managers.py

@@ -12,6 +12,16 @@ def get_or_create_chat(self, user1, user2):
 
     def get_other_user(self, chat, user):
         return chat.user2 if chat.user1 == user else chat.user1
+    
+    def user_blocked_or_muted(self, chat, sender):
+        receiver = self.get_other_user(chat, sender)
+        from .models import BlockedUser, MutedUser # avoid circular import
+
+        if BlockedUser.objects.filter(blocked_by=receiver.profile, blocked_user=sender.profile).exists():
+            raise PermissionError("You are blocked by this user.")
+
+        if MutedUser.objects.filter(muted_by=receiver.profile,  muted_user=sender.profile).exists():
+            raise PermissionError("You are muted by this user.")
 
 class MessageManager(models.Manager):
     def create_message(self, chat, sender, content):

backend/apps/chat/test_chat_manager.py

@@ -0,0 +1,53 @@
+import pytest
+from django.contrib.auth.models import User
+from django.contrib.auth import get_user_model
+from .models import Chat, BlockedUser, MutedUser
+from apps.users.models import Profile
+
+@pytest.mark.django_db
+def test_ensure_user_allowed_blocks_as_expected():
+    User = get_user_model()
+    u1 = User.objects.create_user("u1","p")
+    u2 = User.objects.create_user("u2","p")
+    Profile.objects.create(user=u1, role=1)
+    Profile.objects.create(user=u2, role=1)
+    chat, _ = Chat.objects.get_or_create_chat(u1, u2)
+
+    BlockedUser.objects.create(blocked_by=u2.profile, blocked_user=u1.profile)
+    with pytest.raises(PermissionError, match="You are blocked by this user."):
+        Chat.objects.user_blocked_or_muted(chat, u1)
+
+@pytest.mark.django_db
+def test_user_blocked_or_muted_mutes_as_expected():
+    u1 = User.objects.create_user("u1_muted", password="pass")
+    u2 = User.objects.create_user("u2_muted", password="pass")
+    Profile.objects.create(user=u1, role=1)
+    Profile.objects.create(user=u2, role=1)
+    chat, _ = Chat.objects.get_or_create_chat(u1, u2)
+
+    # mute u1 by u2
+    MutedUser.objects.create(muted_by=u2.profile, muted_user=u1.profile)
+    with pytest.raises(PermissionError, match="You are muted by this user."):
+        Chat.objects.user_blocked_or_muted(chat, u1)
+
+@pytest.mark.django_db
+def test_user_blocked_or_muted_allows_when_clear():
+    u1 = User.objects.create_user("u1_clear", password="pass")
+    u2 = User.objects.create_user("u2_clear", password="pass")
+    Profile.objects.create(user=u1, role=1)
+    Profile.objects.create(user=u2, role=1)
+    chat, _ = Chat.objects.get_or_create_chat(u1, u2)
+    Chat.objects.user_blocked_or_muted(chat, u1) # should not raise error
+
+@pytest.mark.django_db
+def test_get_other_user_returns_correct_partner():
+    u1 = User.objects.create_user("u1_other", password="pass")
+    u2 = User.objects.create_user("u2_other", password="pass")
+    # profiles must exist but not used here
+    Profile.objects.create(user=u1, role=1)
+    Profile.objects.create(user=u2, role=1)
+
+    chat, _ = Chat.objects.get_or_create_chat(u1, u2)
+    # whoever you pass in, you get the other one back
+    assert Chat.objects.get_other_user(chat, u1) == u2
+    assert Chat.objects.get_other_user(chat, u2) == u1

backend/apps/chat/tests.py

@@ -1,16 +1,20 @@
 from django.contrib.auth.models import User
+from apps.users.models import Profile
 from django.urls import reverse
 from rest_framework import status
 from rest_framework.test import APITestCase, APIClient
-from .models import Chat, Message
+from django.test import TestCase
+from apps.chat.models import Chat, Message, MutedUser, BlockedUser, ReportedUser
 from apps.chat.consumers import ChatConsumer
+from django.utils.timezone import now
+import time
+from django.core.exceptions import ValidationError
 
 def test_chat_consumer_init():
     consumer = ChatConsumer()
     assert consumer is not None
 
 class ChatAPITestCase(APITestCase):
-
     def setUp(self):
         self.user1 = User.objects.create_user(username='alice', password='pass1234')
         self.user2 = User.objects.create_user(username='bob', password='pass1234')
@@ -63,21 +67,197 @@ def test_send_message(self):
         self.assertEqual(message.content, 'Hello Bob!')
         self.assertEqual(message.sender, self.user1)
 
+    def test_create_chat_missing_user2(self):
+        url = reverse('chat-list')
+        response = self.client.post(url, {})  # Missing "user2"
+        self.assertEqual(response.status_code, 400)
+        self.assertIn("user2", response.data)
 
-    def test_list_chat_messages(self):
-        chat, created = Chat.objects.get_or_create_chat(self.user1, self.user2)
-        Message.objects.create(chat=chat, sender=self.user1, content='Hi Bob!')
-        Message.objects.create(chat=chat, sender=self.user2, content='Hey Alice!')
+    def test_create_chat_user2_not_found(self):
+        url = reverse('chat-list')
+        response = self.client.post(url, {"user2": 9999})  # non-existent ID
+        self.assertEqual(response.status_code, 400)
+        self.assertIn("user2", response.data)
 
-        url = reverse('chat-messages', kwargs={'pk': chat.id})
+    def test_create_chat_already_exists(self):
+        Chat.objects.get_or_create_chat(self.user1, self.user2)
+        url = reverse('chat-list')
+        response = self.client.post(url, {"user2": self.user2.id})
+        self.assertEqual(response.status_code, 400)
+        self.assertIn("Chat already exists", str(response.data))
+
+    def test_get_messages_custom_action(self):
+        chat, _ = Chat.objects.get_or_create_chat(self.user1, self.user2)
+        Message.objects.create(chat=chat, sender=self.user1, content="Hello!")
+        url = reverse('chat-messages', args=[chat.id])  # Must match router name
+        response = self.client.get(url)
+        self.assertEqual(response.status_code, 200)
+        self.assertEqual(len(response.data), 1)
+        self.assertEqual(response.data[0]['content'], "Hello!")
+
+# https://chatgpt.com/share/681180bd-8930-800c-9f58-8f4de378d8e2
+class ChatModelTests(TestCase):
+    def setUp(self):
+        self.user1 = User.objects.create_user(username='alice', password='pass1234')
+        self.user2 = User.objects.create_user(username='bob', password='pass1234')
+        self.profile1 = Profile.objects.create(user=self.user1, role=1)
+        self.profile2 = Profile.objects.create(user=self.user2, role=1)
+
+    def test_chat_str(self):
+        chat = Chat.objects.create(user1=self.user1, user2=self.user2)
+        expected_str = f"Chat between {self.user1} and {self.user2}"
+        self.assertEqual(str(chat), expected_str)
+
+    def test_message_str(self):
+        chat = Chat.objects.create(user1=self.user1, user2=self.user2)
+        message = Message.objects.create(chat=chat, sender=self.user1, content="Hello!", timestamp=now())
+        expected_str = f"From {self.user1} in Chat {chat.id} at {message.timestamp}"
+        self.assertEqual(str(message), expected_str)
+
+    def test_muted_user_str(self):
+        muted = MutedUser.objects.create(muted_user=self.profile2, muted_by=self.profile1)
+        expected_str = f"{self.profile1} muted {self.profile2}"
+        self.assertEqual(str(muted), expected_str)
+
+    def test_blocked_user_str(self):
+        blocked = BlockedUser.objects.create(blocked_user=self.profile2, blocked_by=self.profile1)
+        expected_str = f"{self.profile1} blocked {self.profile2}"
+        self.assertEqual(str(blocked), expected_str)
+
+    def test_reported_user_str(self):
+        reported = ReportedUser.objects.create(reported_user=self.profile2, reported_by=self.profile1, reason="Spam")
+        expected_str = f"{self.profile1} reported {self.profile2}"
+        self.assertEqual(str(reported), expected_str)
+
+    def test_muted_user_clean_valid(self):
+        muted = MutedUser(muted_user=self.profile2, muted_by=self.profile1)
+        muted.clean()  # Should not raise
+
+    def test_muted_user_clean_invalid(self):
+        muted = MutedUser(muted_user=self.profile1, muted_by=self.profile1)
+        with self.assertRaises(ValidationError) as context:
+            muted.clean()
+        self.assertIn("You cannot mute yourself.", str(context.exception))
+
+    def test_blocked_user_clean_valid(self):
+        blocked = BlockedUser(blocked_user=self.profile2, blocked_by=self.profile1)
+        blocked.clean()  # should not raise
+
+    def test_blocked_user_clean_invalid(self):
+        blocked = BlockedUser(blocked_user=self.profile1, blocked_by=self.profile1)
+        with self.assertRaises(ValidationError) as context:
+            blocked.clean()
+        self.assertIn("You cannot block yourself.", str(context.exception))
+    
+class ChatManagerTests(TestCase):
+    def setUp(self):
+        self.user1 = User.objects.create_user(username='alice', password='pass1234')
+        self.user2 = User.objects.create_user(username='bob', password='pass1234')
+        self.profile1 = Profile.objects.create(user=self.user1, role=1)
+        self.profile2 = Profile.objects.create(user=self.user2, role=1)
+        self.chat, _ = Chat.objects.get_or_create_chat(self.user1, self.user2)
+
+    def test_get_or_create_chat_orders_users(self):
+        Chat.objects.all().delete()
+        chat1, created1 = Chat.objects.get_or_create_chat(self.user2, self.user1)
+        self.assertTrue(created1)
+        self.assertEqual(chat1.user1, min(self.user1, self.user2, key=lambda u: u.id))
+        self.assertEqual(chat1.user2, max(self.user1, self.user2, key=lambda u: u.id))
+
+        # should not create a new one
+        chat2, created2 = Chat.objects.get_or_create_chat(self.user1, self.user2)
+        self.assertFalse(created2)
+        self.assertEqual(chat1.id, chat2.id)
+
+    def test_get_other_user(self):
+        chat, _ = Chat.objects.get_or_create_chat(self.user1, self.user2)
+        self.assertEqual(Chat.objects.get_other_user(chat, self.user1), self.user2)
+        self.assertEqual(Chat.objects.get_other_user(chat, self.user2), self.user1)
+
+    def test_user_blocked_or_muted_blocked(self):
+        chat, _ = Chat.objects.get_or_create_chat(self.user1, self.user2)
+        BlockedUser.objects.create(blocked_by=self.profile2, blocked_user=self.profile1)
+
+        with self.assertRaises(PermissionError) as ctx:
+            Chat.objects.user_blocked_or_muted(chat, self.user1)
+        self.assertIn("blocked", str(ctx.exception))
+
+    def test_user_blocked_or_muted_muted(self):
+        chat, _ = Chat.objects.get_or_create_chat(self.user1, self.user2)
+        MutedUser.objects.create(muted_by=self.profile2, muted_user=self.profile1)
+
+        with self.assertRaises(PermissionError) as ctx:
+            Chat.objects.user_blocked_or_muted(chat, self.user1)
+        self.assertIn("muted", str(ctx.exception))
+
+    def test_create_message_and_update_timestamp(self):
+        old_updated_at = self.chat.updatedAt
+        time.sleep(0.01)  # Make sure timestamp will differ
+        msg = Message.objects.create_message(chat=self.chat, sender=self.user1, content="hello!")
+        
+        self.chat.refresh_from_db()
+        self.assertEqual(msg.content, "hello!")
+        self.assertEqual(msg.chat, self.chat)
+        self.assertGreater(self.chat.updatedAt, old_updated_at)
+
+class MuteBlockReportAPITests(APITestCase):
+    def setUp(self):
+        self.user1 = User.objects.create_user(username='alice', password='pass1234')
+        self.user2 = User.objects.create_user(username='bob', password='pass1234')
+        self.profile1 = Profile.objects.create(user=self.user1, role=1)
+        self.profile2 = Profile.objects.create(user=self.user2, role=1)
+
+        self.client = APIClient()
+        self.client.force_authenticate(user=self.user1)
+
+    def test_mute_user(self):
+        url = reverse('mute-list')
+        payload = {"muted_user": self.profile2.id}
+        response = self.client.post(url, payload)
+        self.assertEqual(response.status_code, status.HTTP_201_CREATED)
+        self.assertEqual(MutedUser.objects.count(), 1)
+
+    def test_duplicate_mute_user(self):
+        MutedUser.objects.create(muted_user=self.profile2, muted_by=self.profile1)
+        url = reverse('mute-list')
+        payload = {"muted_user": self.profile2.id}
+        response = self.client.post(url, payload)
+        self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
+
+    def test_block_user(self):
+        url = reverse('block-list')
+        payload = {"blocked_user": self.profile2.id}
+        response = self.client.post(url, payload)
+        self.assertEqual(response.status_code, status.HTTP_201_CREATED)
+        self.assertEqual(BlockedUser.objects.count(), 1)
+
+    def test_duplicate_block_user(self):
+        BlockedUser.objects.create(blocked_user=self.profile2, blocked_by=self.profile1)
+        url = reverse('block-list')
+        payload = {"blocked_user": self.profile2.id}
+        response = self.client.post(url, payload)
+        self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
+
+    def test_report_user(self):
+        url = reverse('report-list')
+        payload = {
+            "reported_user": self.profile2.id,
+            "reason": "Spam"
+        }
+        response = self.client.post(url, payload)
+        self.assertEqual(response.status_code, status.HTTP_201_CREATED)
+        self.assertEqual(ReportedUser.objects.count(), 1)
+
+    def test_list_blocked_users(self):
+        BlockedUser.objects.create(blocked_user=self.profile2, blocked_by=self.profile1)
+        url = reverse('block-list')
         response = self.client.get(url)
         self.assertEqual(response.status_code, status.HTTP_200_OK)
-        self.assertEqual(len(response.data), 2)
-        self.assertEqual(response.data[0]['content'], 'Hi Bob!')
-        self.assertEqual(response.data[1]['content'], 'Hey Alice!')
+        self.assertEqual(len(response.data), 1)
 
-    def test_unauthenticated_access(self):
-        self.client.logout()
-        url = reverse('chat-list')
+    def test_list_reported_users(self):
+        ReportedUser.objects.create(reported_user=self.profile2, reported_by=self.profile1, reason="Spam")
+        url = reverse('report-list')
         response = self.client.get(url)
-        self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED)
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+        self.assertEqual(len(response.data), 1)