Dovodi_Backend/apps/course/tests/test_webhook_live_session.py

import hashlib
import hmac
import json
import tempfile
from datetime import timedelta
from unittest import mock

from django.core.management import call_command
from django.core.files.uploadedfile import SimpleUploadedFile
from django.test import override_settings
from django.urls import reverse
from django.utils import timezone
from dj_language.models import Language
from rest_framework import status
from rest_framework.test import APITestCase

from apps.account.models import ProfessorUser, StudentUser
from apps.course.models import (
    Course,
    CourseCategory,
    CourseLiveSession,
    LiveSessionUser,
)


@override_settings(
    PLUGNMEET_API_SECRET='test-secret',
    MEDIA_ROOT=tempfile.gettempdir(),
)
class PlugNMeetWebhookLiveSessionTests(APITestCase):
    def setUp(self):
        Language.objects.update_or_create(
            id=69,
            defaults={
                'name': 'English',
                'code': 'en',
                'status': True,
                'countries': ['US'],
            },
        )
        self.professor = ProfessorUser.objects.create(
            email='webhook-prof@example.com',
            fullname='Webhook Professor',
            experience_years=8,
        )
        self.student = StudentUser.objects.create(
            email='webhook-student@example.com',
            fullname='Webhook Student',
        )
        self.category = CourseCategory.objects.create(
            name='Webhook Category',
            slug='webhook-category',
        )
        thumbnail = SimpleUploadedFile(
            'thumb.jpg',
            b'filecontent',
            content_type='image/jpeg',
        )
        self.course = Course.objects.create(
            title='Webhook Course',
            slug='webhook-course',
            category=self.category,
            professor=self.professor,
            thumbnail=thumbnail,
            video_type=Course.VedioTypeChoices.YOUTUBE_LINK,
            video_link='https://example.com/video',
            is_online=True,
            online_link='https://example.com/live',
            level=Course.LevelChoices.BEGINNER,
            duration=10,
            lessons_count=1,
            description='Description',
            short_description='Short',
            status=Course.StatusChoices.ONGOING,
            is_free=True,
        )
        self.session = CourseLiveSession.objects.create(
            course=self.course,
            room_id='room-16-1781160705',
            subject='Webhook Course Live Session',
            started_at=timezone.now(),
        )
        LiveSessionUser.objects.create(
            session=self.session,
            user=self.student,
            role='participant',
            entered_at=timezone.now(),
            is_online=True,
        )

    def _post_webhook(self, payload: dict):
        body = json.dumps(payload, ensure_ascii=False).encode('utf-8')
        signature = hmac.new(
            b'test-secret',
            body,
            hashlib.sha256,
        ).hexdigest()
        url = reverse('plugnmeet-webhook')
        return self.client.post(
            url,
            data=body,
            content_type='application/webhook+json',
            HTTP_HASH_TOKEN=signature,
        )

    def test_room_finished_with_room_name_closes_live_session(self):
        response = self._post_webhook(
            {
                'event': 'room_finished',
                'room': {
                    'name': self.session.room_id,
                },
            }
        )

        self.assertEqual(response.status_code, status.HTTP_200_OK)
        self.session.refresh_from_db()
        self.assertIsNotNone(self.session.ended_at)
        self.assertFalse(
            LiveSessionUser.objects.get(session=self.session, user=self.student).is_online
        )

    def test_session_ended_event_closes_live_session(self):
        second_session = CourseLiveSession.objects.create(
            course=self.course,
            room_id='room-16-1781160999',
            subject='Second Live Session',
            started_at=timezone.now(),
        )

        response = self._post_webhook(
            {
                'event': 'session_ended',
                'room': {
                    'name': second_session.room_id,
                },
            }
        )

        self.assertEqual(response.status_code, status.HTTP_200_OK)
        second_session.refresh_from_db()
        self.assertIsNotNone(second_session.ended_at)

    def test_last_moderator_leave_schedules_auto_close(self):
        LiveSessionUser.objects.create(
            session=self.session,
            user=self.professor,
            role='moderator',
            entered_at=timezone.now(),
            is_online=True,
        )

        response = self._post_webhook(
            {
                'event': 'participant_left',
                'room': {
                    'name': self.session.room_id,
                },
                'participant': {
                    'identity': str(self.professor.id),
                },
            }
        )

        self.assertEqual(response.status_code, status.HTTP_200_OK)
        self.session.refresh_from_db()
        self.assertIsNotNone(self.session.last_moderator_left_at)
        self.assertIsNotNone(self.session.auto_close_after_moderator_exit_at)

    @mock.patch('apps.course.management.commands.auto_close_professorless_live_sessions.PlugNMeetClient')
    def test_auto_close_command_closes_professorless_session(self, mock_client_cls):
        LiveSessionUser.objects.create(
            session=self.session,
            user=self.professor,
            role='moderator',
            entered_at=timezone.now() - timedelta(minutes=20),
            exited_at=timezone.now() - timedelta(minutes=15),
            is_online=False,
        )
        self.session.last_moderator_left_at = timezone.now() - timedelta(minutes=15)
        self.session.auto_close_after_moderator_exit_at = timezone.now() - timedelta(minutes=5)
        self.session.save(
            update_fields=[
                'last_moderator_left_at',
                'auto_close_after_moderator_exit_at',
                'updated_at',
            ]
        )

        mock_client = mock_client_cls.return_value
        mock_client.end_room.return_value = {'status': True, 'msg': 'room ended successfully'}

        call_command('auto_close_professorless_live_sessions')

        self.session.refresh_from_db()
        self.assertIsNotNone(self.session.ended_at)
        self.assertIsNone(self.session.last_moderator_left_at)
        self.assertIsNone(self.session.auto_close_after_moderator_exit_at)
        self.assertFalse(
            LiveSessionUser.objects.get(session=self.session, user=self.student).is_online
        )