import datetime from django.utils import timezone from django.test import TestCase from unittest.mock import patch from django.core.files.uploadedfile import SimpleUploadedFile from dj_language.models import Language from apps.account.models import StudentUser, User, Notification from apps.course.models.course import Course, CourseCategory from apps.course.models.participant import Participant from apps.chat.models import RoomMessage, ChatMessage class CommunicationNotificationsFlowTests(TestCase): def setUp(self): # Create languages self.lang_fa, _ = Language.objects.update_or_create( code='fa', defaults={'name': 'Persian', 'status': True, 'countries': []} ) self.lang_en, _ = Language.objects.update_or_create( code='en', defaults={'name': 'English', 'status': True, 'countries': []} ) # Create category self.category = CourseCategory.objects.create( name=[{"title": "Category", "language_code": "en"}], slug=[{"title": "category", "language_code": "en"}] ) # Create student user self.student = StudentUser.objects.create( email='student_comm@example.com', username='student_comm', fullname='Comm Student', language=self.lang_fa, fcm='token_comm_student' ) # Create teacher user self.teacher = User.objects.create( email='teacher_comm@example.com', username='teacher_comm', fullname='Teacher Comm', user_type='professor', is_active=True, language=self.lang_fa ) # Create support user (admin) self.support = User.objects.create( email='support_comm@example.com', username='support_comm', fullname='Support Comm', user_type='admin', is_active=True, language=self.lang_fa ) # Create Course thumbnail = SimpleUploadedFile('thumb.jpg', b'filecontent', content_type='image/jpeg') self.course = Course.objects.create( title=[ {"title": "Course 1", "language_code": "en"}, {"title": "دوره ۱", "language_code": "fa"} ], slug=[{"title": "course-1", "language_code": "en"}], category=self.category, thumbnail=thumbnail, video_type=Course.VedioTypeChoices.YOUTUBE_LINK, video_link='https://example.com/video', is_online=False, level=Course.LevelChoices.BEGINNER, duration=10, lessons_count=0, description='Description', short_description='Short description', status=[{"title": "ongoing", "language_code": "en"}], is_free=True, ) # Enroll student in the course self.participant = Participant.objects.create( student=self.student, course=self.course, is_active=True ) @patch('apps.account.notification_service.send_notification') def test_teacher_replied_private_chat(self, mock_send): # Create a private chat room between student and teacher room = RoomMessage.objects.create( name="Private chat with teacher", initiator=self.student, recipient=self.teacher, room_type=RoomMessage.RoomTypeChoices.PRIVATE ) # Teacher sends a message ChatMessage.objects.create( room=room, sender=self.teacher, content="Hello from teacher", content_type=ChatMessage.ChatTypeChoices.TEXT ) # Verify notification sent to the student notif = Notification.objects.filter(user=self.student, title="پیام جدید از طرف استاد").first() self.assertIsNotNone(notif) self.assertIn("Hello from teacher", notif.message) self.assertTrue(mock_send.called) @patch('apps.account.notification_service.send_notification') def test_support_replied_private_chat(self, mock_send): # Create a private chat room between student and support room = RoomMessage.objects.create( name="Support chat", initiator=self.student, recipient=self.support, room_type=RoomMessage.RoomTypeChoices.PRIVATE ) # Support agent sends a message ChatMessage.objects.create( room=room, sender=self.support, content="How can I help you?", content_type=ChatMessage.ChatTypeChoices.TEXT ) # Verify notification sent to the student notif = Notification.objects.filter(user=self.student, title="پاسخ پشتیبانی").first() self.assertIsNotNone(notif) self.assertIn("How can I help you?", notif.message) self.assertTrue(mock_send.called) @patch('apps.account.notification_service.send_notification') def test_teacher_posted_in_course_chat(self, mock_send): # Create a course chat room room = RoomMessage.objects.create( name="Course group chat", course=self.course, initiator=self.teacher, room_type=RoomMessage.RoomTypeChoices.GROUP ) # Teacher sends a message to course chat ChatMessage.objects.create( room=room, sender=self.teacher, content="Hello class, remember the homework", content_type=ChatMessage.ChatTypeChoices.TEXT ) # Verify notification sent to the student enrolled notif = Notification.objects.filter(user=self.student, title="پیام استاد در گفتگوی دوره").first() self.assertIsNotNone(notif) self.assertIn("homework", notif.message) self.assertIn("دوره ۱", notif.message) self.assertTrue(mock_send.called) @patch('apps.account.notification_service.send_notification') def test_student_message_no_notification(self, mock_send): # Create a private chat room room = RoomMessage.objects.create( name="Private chat", initiator=self.student, recipient=self.teacher, room_type=RoomMessage.RoomTypeChoices.PRIVATE ) # Clear notifications Notification.objects.filter(user=self.student).delete() mock_send.reset_mock() # Student sends a message ChatMessage.objects.create( room=room, sender=self.student, content="Hello professor", content_type=ChatMessage.ChatTypeChoices.TEXT ) # Student should not receive a notification for their own message self.assertFalse(Notification.objects.filter(user=self.student).exists()) self.assertFalse(mock_send.called) @patch('apps.account.notification_service.send_notification') def test_webhook_unauthorized(self, mock_send): from django.urls import reverse url = reverse('chat-notify-message') response = self.client.post(url, {'message_id': 1}, HTTP_X_INTERNAL_SECRET='wrong-token') self.assertEqual(response.status_code, 403) @patch('apps.account.notification_service.send_notification') def test_webhook_authorized_success(self, mock_send): from django.urls import reverse from django.conf import settings from django.db.models.signals import post_save from apps.chat.signals import notify_on_new_chat_message # Create a private chat room room = RoomMessage.objects.create( name="Private chat with teacher", initiator=self.student, recipient=self.teacher, room_type=RoomMessage.RoomTypeChoices.PRIVATE ) # Temporarily disconnect the signal to avoid trigger on save, so we only test webhook execution post_save.disconnect(notify_on_new_chat_message, sender=ChatMessage) try: # Teacher sends a message msg = ChatMessage.objects.create( room=room, sender=self.teacher, content="How are you?", content_type=ChatMessage.ChatTypeChoices.TEXT ) # Clear notifications created (if any) Notification.objects.all().delete() mock_send.reset_mock() url = reverse('chat-notify-message') secret = getattr(settings, 'INTERNAL_WEBHOOK_SECRET', 'super-secret-key-12345') response = self.client.post( url, {'message_id': msg.id}, HTTP_X_INTERNAL_SECRET=secret ) self.assertEqual(response.status_code, 200) # Verify notification was sent notif = Notification.objects.filter(user=self.student).first() self.assertIsNotNone(notif) self.assertIn("How are you?", notif.message) self.assertTrue(mock_send.called) finally: # Reconnect signal post_save.connect(notify_on_new_chat_message, sender=ChatMessage)