import jwt from django.conf import settings from django.test import TestCase from django.core.files.uploadedfile import SimpleUploadedFile from django.contrib.auth import get_user_model from rest_framework.test import APIClient from rest_framework import status from apps.chat.models.chat import ChatRoom, ChatMessage from apps.profiles.models.institution import Institution, InstitutionMember User = get_user_model() class Phase6ChatTests(TestCase): """ Automated test suite for Phase 6: - Centrifugo v5 HMAC-SHA256 WebSocket Token Issuance - Direct & Group Chat Room Creation and Reusability - Conversation Listing with Unread Counters & Last Message - Real-Time Message Dispatch & Centrifugo Event Publishing - Message File Attachments - Read Receipts & Unread Count Synchronization - Strict Room Participant Permissions """ def setUp(self): self.client = APIClient() # Users self.user_a = User.objects.create_user( email="hassan@mashhad.org", password="SecurePassword123!", fullname="Hassan Razavi", user_type=User.UserType.INSTITUTION_ADMIN ) self.user_b = User.objects.create_user( email="fatima@berlin.de", password="SecurePassword123!", fullname="Fatima Al-Berlini", user_type=User.UserType.EDITOR ) self.outsider = User.objects.create_user( email="outsider@community.org", password="SecurePassword123!", fullname="Outsider User", user_type=User.UserType.CLIENT ) # Institutions self.institution = Institution.objects.create( name="Imam Reza Holy Shrine Center", type=Institution.InstitutionType.INSTITUTE, country="Iran", city="Mashhad" ) InstitutionMember.objects.create( institution=self.institution, user=self.user_a, role=InstitutionMember.MemberRole.ADMIN ) def test_centrifugo_token_issuance(self): self.client.force_authenticate(user=self.user_a) response = self.client.get("/api/v1/chat/centrifugo-token/") self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertIn("token", response.data) self.assertIn("ws_url", response.data) self.assertEqual(response.data["user_id"], self.user_a.id) # Verify decoded token signature token_str = response.data["token"] decoded = jwt.decode( token_str, settings.CENTRIFUGO_SECRET, algorithms=["HS256"] ) self.assertEqual(decoded["sub"], str(self.user_a.id)) self.assertEqual(decoded["info"]["email"], self.user_a.email) def test_chat_room_creation_and_direct_chat_reuse(self): self.client.force_authenticate(user=self.user_a) # 1. Create direct chat with user_b payload = { "room_type": "direct", "target_user_id": self.user_b.id, "initial_message": "Salam Fatima, how is the Berlin conference preparation going?" } create_resp = self.client.post("/api/v1/chat/rooms/", payload, format="json") self.assertEqual(create_resp.status_code, status.HTTP_201_CREATED) room_id = create_resp.data["id"] self.assertEqual(create_resp.data["room_type"], "direct") # 2. Call again with the same target user -> should reuse existing room (status 200) reuse_payload = { "room_type": "direct", "target_user_id": self.user_b.id, } reuse_resp = self.client.post("/api/v1/chat/rooms/", reuse_payload, format="json") self.assertEqual(reuse_resp.status_code, status.HTTP_200_OK) self.assertEqual(reuse_resp.data["id"], room_id) def test_chat_room_listing_and_unread_counts(self): # Create a room between user_a and user_b room = ChatRoom.objects.create(room_type=ChatRoom.RoomType.DIRECT, created_by=self.user_a) room.participant_users.add(self.user_a, self.user_b) # user_a sends 2 messages msg1 = ChatMessage.objects.create(room=room, sender=self.user_a, content="First message") msg2 = ChatMessage.objects.create(room=room, sender=self.user_a, content="Second message") # user_b lists rooms -> should show unread_count = 2 self.client.force_authenticate(user=self.user_b) list_resp = self.client.get("/api/v1/chat/rooms/") self.assertEqual(list_resp.status_code, status.HTTP_200_OK) self.assertEqual(len(list_resp.data["results"]), 1) self.assertEqual(list_resp.data["results"][0]["unread_count"], 2) self.assertEqual(list_resp.data["results"][0]["last_message"]["content"], "Second message") def test_message_dispatch_and_centrifugo_publish(self): room = ChatRoom.objects.create(room_type=ChatRoom.RoomType.DIRECT, created_by=self.user_a) room.participant_users.add(self.user_a, self.user_b) self.client.force_authenticate(user=self.user_b) send_resp = self.client.post( f"/api/v1/chat/rooms/{room.id}/messages/", {"content": "Alhamdulillah, conference agenda is finalized!"}, format="json" ) self.assertEqual(send_resp.status_code, status.HTTP_201_CREATED) self.assertEqual(send_resp.data["content"], "Alhamdulillah, conference agenda is finalized!") self.assertEqual(send_resp.data["sender"]["email"], self.user_b.email) # Message count in DB self.assertEqual(ChatMessage.objects.filter(room=room).count(), 1) def test_message_attachment_upload(self): room = ChatRoom.objects.create(room_type=ChatRoom.RoomType.DIRECT, created_by=self.user_a) room.participant_users.add(self.user_a, self.user_b) self.client.force_authenticate(user=self.user_a) mock_pdf = SimpleUploadedFile("agenda_schedule.pdf", b"%PDF-1.4 Mock Schedule", content_type="application/pdf") upload_resp = self.client.post( f"/api/v1/chat/rooms/{room.id}/messages/", { "content": "Attached is the latest schedule.", "attachment": mock_pdf }, format="multipart" ) self.assertEqual(upload_resp.status_code, status.HTTP_201_CREATED) self.assertIn("agenda_schedule", upload_resp.data["attachment"]) def test_mark_room_messages_as_read(self): room = ChatRoom.objects.create(room_type=ChatRoom.RoomType.DIRECT, created_by=self.user_a) room.participant_users.add(self.user_a, self.user_b) ChatMessage.objects.create(room=room, sender=self.user_a, content="Unread message 1") ChatMessage.objects.create(room=room, sender=self.user_a, content="Unread message 2") # user_b marks room as read self.client.force_authenticate(user=self.user_b) read_resp = self.client.post(f"/api/v1/chat/rooms/{room.id}/read/") self.assertEqual(read_resp.status_code, status.HTTP_200_OK) # Verify unread count is now 0 list_resp = self.client.get("/api/v1/chat/rooms/") self.assertEqual(list_resp.data["results"][0]["unread_count"], 0) def test_room_participant_permissions(self): room = ChatRoom.objects.create(room_type=ChatRoom.RoomType.DIRECT, created_by=self.user_a) room.participant_users.add(self.user_a, self.user_b) # Outsider attempts to read messages -> 403 Forbidden self.client.force_authenticate(user=self.outsider) forbidden_get = self.client.get(f"/api/v1/chat/rooms/{room.id}/messages/") self.assertEqual(forbidden_get.status_code, status.HTTP_403_FORBIDDEN) # Outsider attempts to send message -> 403 Forbidden forbidden_post = self.client.post( f"/api/v1/chat/rooms/{room.id}/messages/", {"content": "Intruder trying to send message"}, format="json" ) self.assertEqual(forbidden_post.status_code, status.HTTP_403_FORBIDDEN)