# Run from /backend with: python manage.py test chat
import json
from channels.testing import WebsocketCommunicator
from django.test import TestCase, TransactionTestCase
from django.urls import reverse
from chat.consumers import ChatConsumer
from chat.models import Message
from matching.services import mentee_unmatch, mentor_unmatch
from users.models import User
PASSWORD = 'testpassword12345'
[docs]
def make_mentee(email):
return User.objects.create_user(email=email, password=PASSWORD, role='MENTEE')
[docs]
def make_mentor(email):
return User.objects.create_user(email=email, password=PASSWORD, role='MENTOR')
[docs]
def match(mentee, mentor):
"""Wire up both sides of a mentee-mentor match."""
mentee.isMatched = True
mentee.matchedMentorEmail = mentor.email
mentee.save()
mentor.current_mentees.append(mentee.email)
mentor.save()
[docs]
def send(sender_email, receiver_email, content='hello'):
return Message.objects.create(
sender_email=sender_email,
receiver_email=receiver_email,
content=content,
)
# ---------- unmatch tests ----------
[docs]
class UnmatchChatDeletionTests(TestCase):
[docs]
def setUp(self):
self.mentee = make_mentee('mentee1@ucla.edu')
self.mentor = make_mentor('mentor1@ucla.edu')
match(self.mentee, self.mentor)
send(self.mentee.email, self.mentor.email, 'hi mentor')
send(self.mentor.email, self.mentee.email, 'hi mentee')
[docs]
def test_mentee_unmatch_deletes_messages(self):
self.assertEqual(Message.objects.count(), 2)
mentee_unmatch(self.mentee, self.mentor.email)
self.assertEqual(Message.objects.count(), 0)
[docs]
def test_mentor_unmatch_deletes_messages(self):
self.assertEqual(Message.objects.count(), 2)
mentor_unmatch(self.mentor, self.mentee.email)
self.assertEqual(Message.objects.count(), 0)
[docs]
def test_unmatch_only_deletes_between_pair(self):
mentee2 = make_mentee('mentee2@ucla.edu')
match(mentee2, self.mentor)
send(mentee2.email, self.mentor.email, 'msg from mentee2')
send(self.mentor.email, mentee2.email, 'reply to mentee2')
self.assertEqual(Message.objects.count(), 4)
mentee_unmatch(self.mentee, self.mentor.email)
# Only mentee2's messages remain
self.assertEqual(Message.objects.count(), 2)
remaining = Message.objects.all()
for msg in remaining:
self.assertIn(mentee2.email, [msg.sender_email, msg.receiver_email])
# ---------- account deletion tests ----------
[docs]
class AccountDeletionChatDeletionTests(TestCase):
[docs]
def test_mentee_deletion_deletes_sent_messages(self):
mentee = make_mentee('mentee3@ucla.edu')
mentor = make_mentor('mentor3@ucla.edu')
match(mentee, mentor)
send(mentee.email, mentor.email, 'sent by mentee')
self.assertEqual(Message.objects.count(), 1)
mentee.request_deletion()
self.assertEqual(Message.objects.count(), 0)
[docs]
def test_mentee_deletion_deletes_received_messages(self):
mentee = make_mentee('mentee4@ucla.edu')
mentor = make_mentor('mentor4@ucla.edu')
match(mentee, mentor)
send(mentor.email, mentee.email, 'sent by mentor')
self.assertEqual(Message.objects.count(), 1)
mentee.request_deletion()
self.assertEqual(Message.objects.count(), 0)
[docs]
def test_mentor_deletion_deletes_all_messages(self):
mentor = make_mentor('mentor5@ucla.edu')
mentee_a = make_mentee('mentee5a@ucla.edu')
mentee_b = make_mentee('mentee5b@ucla.edu')
match(mentee_a, mentor)
match(mentee_b, mentor)
send(mentee_a.email, mentor.email)
send(mentor.email, mentee_a.email)
send(mentee_b.email, mentor.email)
send(mentor.email, mentee_b.email)
self.assertEqual(Message.objects.count(), 4)
mentor.request_deletion()
self.assertEqual(Message.objects.count(), 0)
[docs]
def test_deletion_only_deletes_own_messages(self):
# Pair 1
mentee1 = make_mentee('mentee6@ucla.edu')
mentor1 = make_mentor('mentor6@ucla.edu')
match(mentee1, mentor1)
send(mentee1.email, mentor1.email)
send(mentor1.email, mentee1.email)
# Pair 2
mentee2 = make_mentee('mentee7@ucla.edu')
mentor2 = make_mentor('mentor7@ucla.edu')
match(mentee2, mentor2)
send(mentee2.email, mentor2.email)
send(mentor2.email, mentee2.email)
self.assertEqual(Message.objects.count(), 4)
mentee1.request_deletion()
# Only pair 2's messages remain
self.assertEqual(Message.objects.count(), 2)
for msg in Message.objects.all():
self.assertNotIn(mentee1.email, [msg.sender_email, msg.receiver_email])
[docs]
class ChatModelMethodTests(TestCase):
[docs]
def setUp(self):
self.mentee = make_mentee('mentee_model@ucla.edu')
self.mentor = make_mentor('mentor_model@ucla.edu')
match(self.mentee, self.mentor)
# Send 3 messages: 2 from mentor (unread), 1 from mentee
send(self.mentor.email, self.mentee.email, 'Hello 1')
send(self.mentor.email, self.mentee.email, 'Hello 2')
send(self.mentee.email, self.mentor.email, 'Reply 1')
[docs]
def test_get_unread_count(self):
"""Verifies unread counts are calculated correctly per user"""
# Mentee should have 2 unread messages from mentor
self.assertEqual(Message.get_unread_count(self.mentee.email, self.mentor.email), 2)
# Mentor should have 1 unread message from mentee
self.assertEqual(Message.get_unread_count(self.mentor.email, self.mentee.email), 1)
[docs]
def test_mark_as_read(self):
"""Verifies marking messages as read updates the database"""
# Mentee reads mentor's messages
updated_count = Message.mark_as_read(self.mentee.email, self.mentor.email)
self.assertEqual(updated_count, 2)
# Verify count is now 0
self.assertEqual(Message.get_unread_count(self.mentee.email, self.mentor.email), 0)
[docs]
class ChatAPIEndpointTests(TestCase):
[docs]
def setUp(self):
self.mentee = make_mentee('mentee_api@ucla.edu')
self.mentor = make_mentor('mentor_api@ucla.edu')
self.hacker = make_mentee('hacker_api@ucla.edu') # Not matched with anyone
match(self.mentee, self.mentor)
# Ensure verified so they can log in
self.mentee.is_verified = True
self.mentor.is_verified = True
self.hacker.is_verified = True
self.mentee.save()
self.mentor.save()
self.hacker.save()
[docs]
def test_get_chat_partners_success(self):
"""Verifies a matched user can fetch their chat partners list"""
self.client.login(email=self.mentee.email, password=PASSWORD)
response = self.client.get(reverse('chat_partners'))
self.assertEqual(response.status_code, 200)
data = response.json()
self.assertEqual(len(data['partners']), 1)
self.assertEqual(data['partners'][0]['email'], self.mentor.email)
[docs]
def test_unauthorized_message_sending_blocked(self):
"""Ensures an unmatched user gets a 403 Forbidden when trying to send a message"""
self.client.login(email=self.hacker.email, password=PASSWORD)
payload = {
'receiver_email': self.mentor.email,
'content': 'I am hacking your chat!'
}
response = self.client.post(reverse('send_message'), data=json.dumps(payload), content_type='application/json')
# Hacker is not matched with the mentor, should be blocked
self.assertEqual(response.status_code, 403)
self.assertEqual(Message.objects.count(), 0)
[docs]
def test_authorized_message_sending_success(self):
"""Ensures a matched user can successfully send a message via the API"""
self.client.login(email=self.mentee.email, password=PASSWORD)
payload = {
'receiver_email': self.mentor.email,
'content': 'Hello from the API test!'
}
response = self.client.post(reverse('send_message'), data=json.dumps(payload), content_type='application/json')
self.assertEqual(response.status_code, 200)
self.assertEqual(Message.objects.count(), 1)
self.assertEqual(Message.objects.first().content, 'Hello from the API test!')
[docs]
def test_unauthenticated_access_blocked(self):
"""Ensures all endpoints return 401 Unauthorized if not logged in"""
# No login performed
response = self.client.get(reverse('chat_partners'))
self.assertEqual(response.status_code, 401)
[docs]
def test_get_chat_history_success(self):
"""Verifies a matched user can fetch their paginated chat history in chronological order"""
# 1. Create a quick conversation using your existing helper
send(self.mentee.email, self.mentor.email, 'First message')
send(self.mentor.email, self.mentee.email, 'Second message')
send(self.mentee.email, self.mentor.email, 'Third message')
# 2. Log in and fetch the full history
self.client.login(email=self.mentee.email, password=PASSWORD)
url = reverse('chat_history', kwargs={'partner_email': self.mentor.email})
response = self.client.get(url)
self.assertEqual(response.status_code, 200)
data = response.json()
# Verify the partner's profile data was bundled correctly
self.assertEqual(data['partner']['email'], self.mentor.email)
# Verify messages are returned in chronological order (oldest first)
self.assertEqual(len(data['messages']), 3)
self.assertEqual(data['messages'][0]['content'], 'First message')
self.assertEqual(data['messages'][2]['content'], 'Third message')
self.assertFalse(data['has_more']) # We fetched all 3, so has_more should be False
# 3. Test the Pagination (limit and offset)
response_paged = self.client.get(url, {'limit': 2, 'offset': 0})
data_paged = response_paged.json()
# The database query fetches the 2 newest messages (Third and Second),
# and the view reverses them for the frontend, so we expect [Second, Third]
self.assertEqual(len(data_paged['messages']), 2)
self.assertEqual(data_paged['messages'][0]['content'], 'Second message')
self.assertEqual(data_paged['messages'][1]['content'], 'Third message')
self.assertTrue(data_paged['has_more']) # There is 1 older message left, so has_more is True
[docs]
class ChatWebSocketTests(TransactionTestCase):
[docs]
def setUp(self):
# Create users using your existing helper functions
self.mentee = make_mentee('mentee_ws@ucla.edu')
self.mentor = make_mentor('mentor_ws@ucla.edu')
self.hacker = make_mentee('hacker_ws@ucla.edu') # Unmatched user
# Match the mentee and mentor
match(self.mentee, self.mentor)
[docs]
async def test_valid_connection_and_messaging(self):
"""Ensures matched users can connect and send live messages"""
# 1. Connect the Mentee
mentee_client = WebsocketCommunicator(ChatConsumer.as_asgi(), f"/ws/chat/{self.mentor.email}/")
mentee_client.scope['user'] = self.mentee
mentee_client.scope['url_route'] = {'kwargs': {'partner_email': self.mentor.email}}
connected1, _ = await mentee_client.connect()
self.assertTrue(connected1)
# Flush the initial "Connected to chat" welcome message
welcome_msg = await mentee_client.receive_json_from()
self.assertEqual(welcome_msg['type'], 'connection_established')
# 2. Connect the Mentor to the exact same room
mentor_client = WebsocketCommunicator(ChatConsumer.as_asgi(), f"/ws/chat/{self.mentee.email}/")
mentor_client.scope['user'] = self.mentor
mentor_client.scope['url_route'] = {'kwargs': {'partner_email': self.mentee.email}}
connected2, _ = await mentor_client.connect()
self.assertTrue(connected2)
await mentor_client.receive_json_from() # Flush mentor's welcome message
# 3. Mentee sends a live message
await mentee_client.send_json_to({
'type': 'chat_message',
'content': 'Hello Mentor! This is a live WebSocket test.'
})
# 4. Mentor instantly receives the broadcasted message
received = await mentor_client.receive_json_from()
self.assertEqual(received['type'], 'chat_message')
self.assertEqual(received['content'], 'Hello Mentor! This is a live WebSocket test.')
self.assertEqual(received['sender_email'], self.mentee.email)
# 5. Cleanup connections
await mentee_client.disconnect()
await mentor_client.disconnect()
[docs]
async def test_unauthorized_connection_rejected(self):
"""Ensures unmatched users are instantly rejected and disconnected"""
hacker_client = WebsocketCommunicator(ChatConsumer.as_asgi(), f"/ws/chat/{self.mentor.email}/")
hacker_client.scope['user'] = self.hacker
hacker_client.scope['url_route'] = {'kwargs': {'partner_email': self.mentor.email}}
# Attempt to connect
connected, _ = await hacker_client.connect()
# The consumer should immediately drop the connection (code 4003)
self.assertFalse(connected)