You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
188 lines
7.9 KiB
188 lines
7.9 KiB
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)
|