diff --git a/config/asgi.py b/config/asgi.py index a850fd6..9952802 100644 --- a/config/asgi.py +++ b/config/asgi.py @@ -1,6 +1,18 @@ import os + from django.core.asgi import get_asgi_application os.environ.setdefault("DJANGO_SETTINGS_MODULE", "config.settings.prod") -application = get_asgi_application() +django_asgi_app = get_asgi_application() + +from channels.routing import ProtocolTypeRouter, URLRouter # noqa: E402 +from training.routing import websocket_urlpatterns # noqa: E402 +from training.middleware import JWTAuthMiddlewareStack # noqa: E402 + +application = ProtocolTypeRouter( + { + "http": django_asgi_app, + "websocket": JWTAuthMiddlewareStack(URLRouter(websocket_urlpatterns)), + } +) diff --git a/config/settings/base.py b/config/settings/base.py index ed5674a..55bca44 100644 --- a/config/settings/base.py +++ b/config/settings/base.py @@ -21,6 +21,8 @@ DJANGO_APPS = [ ] THIRD_PARTY_APPS = [ + "daphne", + "channels", "rest_framework", "rest_framework_simplejwt", "django_filters", @@ -84,6 +86,15 @@ TEMPLATES = [ WSGI_APPLICATION = "config.wsgi.application" ASGI_APPLICATION = "config.asgi.application" +CHANNEL_LAYERS = { + "default": { + "BACKEND": "channels_redis.core.RedisChannelLayer", + "CONFIG": { + "hosts": [env("REDIS_URL", default="redis://localhost:6379/0")], + }, + }, +} + DATABASES = { "default": env.db("DATABASE_URL"), } diff --git a/requirements/base.txt b/requirements/base.txt index bd76410..c815b87 100644 --- a/requirements/base.txt +++ b/requirements/base.txt @@ -3,6 +3,10 @@ Django>=4.2,<5.0 djangorestframework>=3.15 psycopg2-binary>=2.9 +# WebSocket / Channels +channels>=4.0 +channels-redis>=4.0 + # Async task queue celery>=5.3 redis>=5.0 diff --git a/training/consumers.py b/training/consumers.py new file mode 100644 index 0000000..a9d535a --- /dev/null +++ b/training/consumers.py @@ -0,0 +1,361 @@ +import json +import logging + +from channels.db import database_sync_to_async +from channels.generic.websocket import AsyncWebsocketConsumer +from django.utils.timezone import now + +from .models import ( + CallEvent, + CallEventType, + CallParticipant, + CallParticipantStatus, + CallSession, + CallSessionStatus, + Meeting, +) + +logger = logging.getLogger(__name__) + +SIGNALING_MSG_TYPES = {"offer", "answer", "ice_candidate"} +MODERATION_MSG_TYPES = {"mute", "unmute", "kick"} + + +class MeetingCallConsumer(AsyncWebsocketConsumer): + """ + WebSocket consumer for WebRTC signaling in a meeting call. + + URL: /ws/meetings//call/ + + Client sends JSON: + Signaling (P2P relay): + {"type": "offer", "to": "", "sdp": {...}} + {"type": "answer", "to": "", "sdp": {...}} + {"type": "ice_candidate", "to": "", "candidate": {...}} + + Moderation (instructor only): + {"type": "mute", "user_id": ""} + {"type": "unmute", "user_id": ""} + {"type": "kick", "user_id": ""} + + Server → client events: + {"type": "session_state", "session_id": ..., "participants": [...]} + {"type": "participant_joined","user_id":..., "user_name":..., "is_muted":...} + {"type": "participant_left", "user_id":...} + {"type": "offer"|"answer"|"ice_candidate", "from":..., "sdp"|"candidate":...} + {"type": "muted"|"unmuted", "user_id":..., "by":...} + {"type": "kicked", "user_id":..., "kicked_by":...} + {"type": "session_ended"} + """ + + async def connect(self): + user = self.scope.get("user") + if not user or not user.is_authenticated: + await self.close(code=4001) + return + + self.meeting_id = self.scope["url_route"]["kwargs"]["meeting_id"] + self.group_name = f"meeting_call_{self.meeting_id}" + # per-user group for targeted P2P signal relay + self.user_group = f"meeting_call_{self.meeting_id}_user_{user.id}" + self.user = user + + session, allowed = await self._get_or_validate_session() + if not allowed: + await self.close(code=4003) + return + + self.session_id = str(session.id) + + await self.channel_layer.group_add(self.group_name, self.channel_name) + await self.channel_layer.group_add(self.user_group, self.channel_name) + await self.accept() + + participant = await self._record_join() + + await self.channel_layer.group_send( + self.group_name, + { + "type": "call.participant_joined", + "user_id": str(user.id), + "user_name": user.get_full_name() or str(user), + "is_muted": participant.is_muted, + }, + ) + + existing = await self._get_active_participants() + await self.send(text_data=json.dumps({ + "type": "session_state", + "session_id": self.session_id, + "participants": existing, + })) + + async def disconnect(self, close_code): + if not hasattr(self, "group_name"): + return + + await self._record_leave() + + await self.channel_layer.group_send( + self.group_name, + {"type": "call.participant_left", "user_id": str(self.user.id)}, + ) + await self.channel_layer.group_discard(self.group_name, self.channel_name) + await self.channel_layer.group_discard(self.user_group, self.channel_name) + + async def receive(self, text_data): + try: + data = json.loads(text_data) + except json.JSONDecodeError: + await self._send_error("invalid_json") + return + + msg_type = data.get("type") + + if msg_type in SIGNALING_MSG_TYPES: + await self._handle_signaling(msg_type, data) + elif msg_type in MODERATION_MSG_TYPES: + await self._handle_moderation(msg_type, data) + else: + await self._send_error("unknown_message_type") + + # --- signaling relay --- + + async def _handle_signaling(self, msg_type, data): + to_user_id = str(data.get("to", "")) + if not to_user_id: + await self._send_error("missing_to") + return + + target_group = f"meeting_call_{self.meeting_id}_user_{to_user_id}" + payload: dict = { + "type": "call.signal", + "signal_type": msg_type, + "from_user_id": str(self.user.id), + } + if msg_type in ("offer", "answer"): + payload["sdp"] = data.get("sdp") + else: + payload["candidate"] = data.get("candidate") + + await self.channel_layer.group_send(target_group, payload) + + # --- moderation (instructor only) --- + + async def _handle_moderation(self, msg_type, data): + if not await self._is_instructor(): + await self._send_error("not_authorized") + return + + target_user_id = str(data.get("user_id", "")) + if not target_user_id: + await self._send_error("missing_user_id") + return + + if msg_type == "kick": + await self._handle_kick(target_user_id) + elif msg_type in ("mute", "unmute"): + await self._handle_mute_toggle(msg_type, target_user_id) + + async def _handle_kick(self, target_user_id): + await self._log_moderation_event(CallEventType.KICK, target_user_id) + await self.channel_layer.group_send( + self.group_name, + { + "type": "call.kicked", + "user_id": target_user_id, + "kicked_by": str(self.user.id), + }, + ) + + async def _handle_mute_toggle(self, action, target_user_id): + is_muted = action == "mute" + await self._set_participant_muted(target_user_id, is_muted) + event_type = CallEventType.MUTE if is_muted else CallEventType.UNMUTE + await self._log_moderation_event(event_type, target_user_id) + + broadcast_type = "call.muted" if is_muted else "call.unmuted" + await self.channel_layer.group_send( + self.group_name, + { + "type": broadcast_type, + "user_id": target_user_id, + "by": str(self.user.id), + }, + ) + + # --- channel message handlers (called by channel layer) --- + + async def call_participant_joined(self, event): + await self.send(text_data=json.dumps({ + "type": "participant_joined", + "user_id": event["user_id"], + "user_name": event["user_name"], + "is_muted": event["is_muted"], + })) + + async def call_participant_left(self, event): + await self.send(text_data=json.dumps({ + "type": "participant_left", + "user_id": event["user_id"], + })) + + async def call_signal(self, event): + payload: dict = { + "type": event["signal_type"], + "from": event["from_user_id"], + } + if event["signal_type"] in ("offer", "answer"): + payload["sdp"] = event.get("sdp") + else: + payload["candidate"] = event.get("candidate") + await self.send(text_data=json.dumps(payload)) + + async def call_muted(self, event): + await self.send(text_data=json.dumps({ + "type": "muted", + "user_id": event["user_id"], + "by": event["by"], + })) + + async def call_unmuted(self, event): + await self.send(text_data=json.dumps({ + "type": "unmuted", + "user_id": event["user_id"], + "by": event["by"], + })) + + async def call_kicked(self, event): + await self.send(text_data=json.dumps({ + "type": "kicked", + "user_id": event["user_id"], + "kicked_by": event["kicked_by"], + })) + if event["user_id"] == str(self.user.id): + await self.close(code=4004) + + async def call_session_ended(self, event): + await self.send(text_data=json.dumps({"type": "session_ended"})) + await self.close(code=1000) + + # --- database helpers --- + + @database_sync_to_async + def _get_or_validate_session(self): + try: + meeting = Meeting.objects.get(pk=self.meeting_id) + except Meeting.DoesNotExist: + return None, False + + is_participant = meeting.participants.filter(user=self.user).exists() + if not (is_participant or _user_is_instructor(self.user)): + return None, False + + session, _ = CallSession.objects.get_or_create( + meeting=meeting, + defaults={"started_by": self.user, "status": CallSessionStatus.ACTIVE}, + ) + if session.status == CallSessionStatus.ENDED: + return session, False + + return session, True + + @database_sync_to_async + def _record_join(self): + session = CallSession.objects.get(pk=self.session_id) + participant, created = CallParticipant.objects.get_or_create( + session=session, + user=self.user, + defaults={"status": CallParticipantStatus.JOINED}, + ) + if not created: + participant.status = CallParticipantStatus.JOINED + participant.is_muted = False + participant.left_at = None + participant.save(update_fields=["status", "is_muted", "left_at"]) + + CallEvent.objects.create( + session=session, + event_type=CallEventType.JOIN, + actor=self.user, + target_user=self.user, + ) + return participant + + @database_sync_to_async + def _record_leave(self): + try: + session = CallSession.objects.get(pk=self.session_id) + participant = CallParticipant.objects.get(session=session, user=self.user) + participant.status = CallParticipantStatus.LEFT + participant.left_at = now() + participant.save(update_fields=["status", "left_at"]) + CallEvent.objects.create( + session=session, + event_type=CallEventType.LEAVE, + actor=self.user, + target_user=self.user, + ) + except Exception: + pass + + @database_sync_to_async + def _get_active_participants(self): + session = CallSession.objects.get(pk=self.session_id) + return [ + { + "user_id": str(p.user_id), + "user_name": p.user.get_full_name() or str(p.user), + "is_muted": p.is_muted, + } + for p in session.call_participants.filter( + status=CallParticipantStatus.JOINED, + ).select_related("user") + ] + + @database_sync_to_async + def _log_moderation_event(self, event_type, target_user_id): + from django.contrib.auth import get_user_model + + User = get_user_model() + session = CallSession.objects.get(pk=self.session_id) + try: + target = User.objects.get(pk=target_user_id) + except User.DoesNotExist: + target = None + CallEvent.objects.create( + session=session, + event_type=event_type, + actor=self.user, + target_user=target, + ) + + @database_sync_to_async + def _set_participant_muted(self, user_id, is_muted): + try: + session = CallSession.objects.get(pk=self.session_id) + p = CallParticipant.objects.get( + session=session, + user_id=user_id, + status=CallParticipantStatus.JOINED, + ) + p.is_muted = is_muted + p.save(update_fields=["is_muted"]) + except CallParticipant.DoesNotExist: + pass + + @database_sync_to_async + def _is_instructor(self): + return _user_is_instructor(self.user) + + async def _send_error(self, code): + await self.send(text_data=json.dumps({"type": "error", "code": code})) + + +def _user_is_instructor(user): + try: + from accounts.services import get_effective_capabilities + + return "training:signoff" in get_effective_capabilities(user) + except Exception: + return False diff --git a/training/middleware.py b/training/middleware.py new file mode 100644 index 0000000..f9f2964 --- /dev/null +++ b/training/middleware.py @@ -0,0 +1,38 @@ +from urllib.parse import parse_qs + +from channels.auth import AuthMiddlewareStack +from channels.db import database_sync_to_async +from channels.middleware import BaseMiddleware +from django.contrib.auth import get_user_model +from django.contrib.auth.models import AnonymousUser +from rest_framework_simplejwt.tokens import AccessToken + +User = get_user_model() + + +@database_sync_to_async +def get_user_from_token(token_key): + try: + token = AccessToken(token_key) + user_id = token["user_id"] + return User.objects.get(pk=user_id) + except Exception: + return AnonymousUser() + + +class JWTAuthMiddleware(BaseMiddleware): + async def __call__(self, scope, receive, send): + query_string = scope.get("query_string", b"").decode() + params = parse_qs(query_string) + token_list = params.get("token", []) + + if token_list: + scope["user"] = await get_user_from_token(token_list[0]) + else: + scope["user"] = AnonymousUser() + + return await super().__call__(scope, receive, send) + + +def JWTAuthMiddlewareStack(inner): + return JWTAuthMiddleware(AuthMiddlewareStack(inner)) diff --git a/training/migrations/0004_call_session_moderation.py b/training/migrations/0004_call_session_moderation.py new file mode 100644 index 0000000..b2480a3 --- /dev/null +++ b/training/migrations/0004_call_session_moderation.py @@ -0,0 +1,107 @@ +import django.db.models.deletion +import uuid +from django.conf import settings +from django.db import migrations, models + + +class Migration(migrations.Migration): + + dependencies = [ + ("training", "0003_meetingparticipant_attendance_session"), + migrations.swappable_dependency(settings.AUTH_USER_MODEL), + ] + + operations = [ + migrations.CreateModel( + name="CallSession", + fields=[ + ("id", models.UUIDField(default=uuid.uuid4, editable=False, primary_key=True, serialize=False)), + ("status", models.CharField( + choices=[("active", "Active"), ("ended", "Ended")], + default="active", + max_length=20, + )), + ("started_at", models.DateTimeField(auto_now_add=True)), + ("ended_at", models.DateTimeField(blank=True, null=True)), + ("meeting", models.OneToOneField( + on_delete=django.db.models.deletion.CASCADE, + related_name="call_session", + to="training.meeting", + )), + ("started_by", models.ForeignKey( + null=True, + on_delete=django.db.models.deletion.SET_NULL, + related_name="started_call_sessions", + to=settings.AUTH_USER_MODEL, + )), + ], + options={"db_table": "training_call_session"}, + ), + migrations.CreateModel( + name="CallParticipant", + fields=[ + ("id", models.UUIDField(default=uuid.uuid4, editable=False, primary_key=True, serialize=False)), + ("status", models.CharField( + choices=[("joined", "Joined"), ("left", "Left")], + default="joined", + max_length=20, + )), + ("is_muted", models.BooleanField(default=False)), + ("screen_share_active", models.BooleanField(default=False)), + ("joined_at", models.DateTimeField(auto_now_add=True)), + ("left_at", models.DateTimeField(blank=True, null=True)), + ("session", models.ForeignKey( + on_delete=django.db.models.deletion.CASCADE, + related_name="call_participants", + to="training.callsession", + )), + ("user", models.ForeignKey( + on_delete=django.db.models.deletion.CASCADE, + related_name="call_participations", + to=settings.AUTH_USER_MODEL, + )), + ], + options={"db_table": "training_call_participant"}, + ), + migrations.CreateModel( + name="CallEvent", + fields=[ + ("id", models.UUIDField(default=uuid.uuid4, editable=False, primary_key=True, serialize=False)), + ("event_type", models.CharField( + choices=[ + ("join", "Join"), + ("leave", "Leave"), + ("mute", "Mute"), + ("unmute", "Unmute"), + ("kick", "Kick"), + ("screen_share_start", "Screen Share Start"), + ("screen_share_stop", "Screen Share Stop"), + ("session_started", "Session Started"), + ("session_ended", "Session Ended"), + ], + max_length=30, + )), + ("metadata", models.JSONField(blank=True, default=dict)), + ("occurred_at", models.DateTimeField(auto_now_add=True)), + ("session", models.ForeignKey( + on_delete=django.db.models.deletion.CASCADE, + related_name="events", + to="training.callsession", + )), + ("actor", models.ForeignKey( + null=True, + on_delete=django.db.models.deletion.SET_NULL, + related_name="call_events_as_actor", + to=settings.AUTH_USER_MODEL, + )), + ("target_user", models.ForeignKey( + blank=True, + null=True, + on_delete=django.db.models.deletion.SET_NULL, + related_name="call_events_as_target", + to=settings.AUTH_USER_MODEL, + )), + ], + options={"db_table": "training_call_event", "ordering": ["occurred_at"]}, + ), + ] diff --git a/training/models.py b/training/models.py index 101c98b..ff2f7e4 100644 --- a/training/models.py +++ b/training/models.py @@ -24,20 +24,10 @@ class SignoffDecision(models.TextChoices): class TrainingRecord(models.Model): - """ - Completion state machine for one enrollment. - Extends tracking.Enrollment with completion mode and trainer workflow. - """ id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False) - enrollment = models.OneToOneField( - "tracking.Enrollment", on_delete=models.CASCADE, related_name="training_record" - ) - completion_mode = models.CharField( - max_length=20, choices=CompletionMode.choices, default=CompletionMode.ONLINE - ) - status = models.CharField( - max_length=20, choices=CompletionStatus.choices, default=CompletionStatus.NOT_STARTED - ) + enrollment = models.OneToOneField("tracking.Enrollment", on_delete=models.CASCADE, related_name="training_record") + completion_mode = models.CharField(max_length=20, choices=CompletionMode.choices, default=CompletionMode.ONLINE) + status = models.CharField(max_length=20, choices=CompletionStatus.choices, default=CompletionStatus.NOT_STARTED) online_passed_at = models.DateTimeField(null=True, blank=True) completed_at = models.DateTimeField(null=True, blank=True) created_at = models.DateTimeField(auto_now_add=True) @@ -46,18 +36,11 @@ class TrainingRecord(models.Model): class Meta: db_table = "training_record" - def __str__(self): - return f"{self.enrollment_id} [{self.completion_mode}] {self.status}" - class TrainerSignoff(models.Model): id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False) - training_record = models.ForeignKey( - TrainingRecord, on_delete=models.CASCADE, related_name="signoffs" - ) - trainer = models.ForeignKey( - settings.AUTH_USER_MODEL, on_delete=models.CASCADE, related_name="given_signoffs" - ) + training_record = models.ForeignKey(TrainingRecord, on_delete=models.CASCADE, related_name="signoffs") + trainer = models.ForeignKey(settings.AUTH_USER_MODEL, on_delete=models.CASCADE, related_name="given_signoffs") decision = models.CharField(max_length=20, choices=SignoffDecision.choices) notes = models.TextField(blank=True) signed_off_at = models.DateTimeField(auto_now_add=True) @@ -66,24 +49,156 @@ class TrainerSignoff(models.Model): db_table = "training_trainer_signoff" ordering = ["-signed_off_at"] - def __str__(self): - return f"Signoff {self.decision} by {self.trainer_id} on {self.training_record_id}" - class SignoffEvidence(models.Model): id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False) - training_record = models.ForeignKey( - TrainingRecord, on_delete=models.CASCADE, related_name="evidence" - ) + training_record = models.ForeignKey(TrainingRecord, on_delete=models.CASCADE, related_name="evidence") file_name = models.CharField(max_length=255) file_path = models.CharField(max_length=1000) uploaded_at = models.DateTimeField(auto_now_add=True) - uploaded_by = models.ForeignKey( - settings.AUTH_USER_MODEL, on_delete=models.CASCADE, related_name="uploaded_evidence" - ) + uploaded_by = models.ForeignKey(settings.AUTH_USER_MODEL, on_delete=models.CASCADE, related_name="uploaded_evidence") class Meta: db_table = "training_signoff_evidence" - def __str__(self): - return f"{self.file_name} for {self.training_record_id}" + +class AttendanceStatus(models.TextChoices): + PRESENT = "present", "Present" + ABSENT = "absent", "Absent" + LATE = "late", "Late" + + +class Meeting(models.Model): + id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False) + title = models.CharField(max_length=255) + starts_at = models.DateTimeField() + created_at = models.DateTimeField(auto_now_add=True) + updated_at = models.DateTimeField(auto_now=True) + + class Meta: + db_table = "training_meeting" + ordering = ["starts_at", "created_at"] + + +class MeetingParticipant(models.Model): + id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False) + meeting = models.ForeignKey(Meeting, on_delete=models.CASCADE, related_name="participants") + user = models.ForeignKey(settings.AUTH_USER_MODEL, on_delete=models.CASCADE, related_name="meeting_participations") + attendance_status = models.CharField(max_length=20, choices=AttendanceStatus.choices, null=True, blank=True) + attendance_recorded_at = models.DateTimeField(null=True, blank=True) + attendance_recorded_by = models.ForeignKey(settings.AUTH_USER_MODEL, on_delete=models.SET_NULL, null=True, blank=True, related_name="recorded_meeting_attendance") + attendance_session = models.ForeignKey( + "attendance.AttendanceSession", + on_delete=models.SET_NULL, + null=True, + blank=True, + related_name="meeting_attendance_links", + ) + created_at = models.DateTimeField(auto_now_add=True) + + class Meta: + db_table = "training_meeting_participant" + constraints = [models.UniqueConstraint(fields=["meeting", "user"], name="training_unique_meeting_participant")] + + +class CallSessionStatus(models.TextChoices): + ACTIVE = "active", "Active" + ENDED = "ended", "Ended" + + +class CallSession(models.Model): + id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False) + meeting = models.OneToOneField(Meeting, on_delete=models.CASCADE, related_name="call_session") + status = models.CharField(max_length=20, choices=CallSessionStatus.choices, default=CallSessionStatus.ACTIVE) + started_by = models.ForeignKey( + settings.AUTH_USER_MODEL, + on_delete=models.SET_NULL, + null=True, + related_name="started_call_sessions", + ) + started_at = models.DateTimeField(auto_now_add=True) + ended_at = models.DateTimeField(null=True, blank=True) + + class Meta: + db_table = "training_call_session" + + +class CallParticipantStatus(models.TextChoices): + JOINED = "joined", "Joined" + LEFT = "left", "Left" + + +class CallParticipant(models.Model): + id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False) + session = models.ForeignKey(CallSession, on_delete=models.CASCADE, related_name="call_participants") + user = models.ForeignKey( + settings.AUTH_USER_MODEL, + on_delete=models.CASCADE, + related_name="call_participations", + ) + status = models.CharField(max_length=20, choices=CallParticipantStatus.choices, default=CallParticipantStatus.JOINED) + is_muted = models.BooleanField(default=False) + screen_share_active = models.BooleanField(default=False) + joined_at = models.DateTimeField(auto_now_add=True) + left_at = models.DateTimeField(null=True, blank=True) + + class Meta: + db_table = "training_call_participant" + + +class CallEventType(models.TextChoices): + JOIN = "join", "Join" + LEAVE = "leave", "Leave" + MUTE = "mute", "Mute" + UNMUTE = "unmute", "Unmute" + KICK = "kick", "Kick" + SCREEN_SHARE_START = "screen_share_start", "Screen Share Start" + SCREEN_SHARE_STOP = "screen_share_stop", "Screen Share Stop" + SESSION_STARTED = "session_started", "Session Started" + SESSION_ENDED = "session_ended", "Session Ended" + + +class CallEvent(models.Model): + id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False) + session = models.ForeignKey(CallSession, on_delete=models.CASCADE, related_name="events") + event_type = models.CharField(max_length=30, choices=CallEventType.choices) + actor = models.ForeignKey( + settings.AUTH_USER_MODEL, + on_delete=models.SET_NULL, + null=True, + related_name="call_events_as_actor", + ) + target_user = models.ForeignKey( + settings.AUTH_USER_MODEL, + on_delete=models.SET_NULL, + null=True, + blank=True, + related_name="call_events_as_target", + ) + metadata = models.JSONField(default=dict, blank=True) + occurred_at = models.DateTimeField(auto_now_add=True) + + class Meta: + db_table = "training_call_event" + ordering = ["occurred_at"] + + +class EmployeeAttendance(models.Model): + id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False) + user = models.ForeignKey(settings.AUTH_USER_MODEL, on_delete=models.CASCADE, related_name="employee_attendance") + attendance_date = models.DateField() + status = models.CharField(max_length=20, choices=AttendanceStatus.choices) + meeting = models.ForeignKey(Meeting, on_delete=models.SET_NULL, null=True, blank=True, related_name="employee_attendance_entries") + source = models.CharField(max_length=64, default="meeting") + updated_by = models.ForeignKey(settings.AUTH_USER_MODEL, on_delete=models.SET_NULL, null=True, blank=True, related_name="updated_employee_attendance") + updated_at = models.DateTimeField(auto_now=True) + created_at = models.DateTimeField(auto_now_add=True) + + class Meta: + db_table = "training_employee_attendance" + constraints = [ + models.UniqueConstraint( + fields=["user", "attendance_date", "meeting"], + name="training_unique_employee_attendance_for_meeting", + ) + ] diff --git a/training/permissions.py b/training/permissions.py index 0a39b76..33df143 100644 --- a/training/permissions.py +++ b/training/permissions.py @@ -7,3 +7,14 @@ class IsTrainer(BasePermission): return False from accounts.services import get_effective_capabilities return "training:signoff" in get_effective_capabilities(request.user) + + +class IsMeetingModerator(BasePermission): + """Grants access to users who can moderate meeting calls (trainers).""" + + def has_permission(self, request, view): + if not request.user or not request.user.is_authenticated: + return False + from accounts.services import get_effective_capabilities + caps = get_effective_capabilities(request.user) + return "training:signoff" in caps or "meeting:moderate" in caps diff --git a/training/routing.py b/training/routing.py new file mode 100644 index 0000000..8c84246 --- /dev/null +++ b/training/routing.py @@ -0,0 +1,7 @@ +from django.urls import re_path + +from .consumers import MeetingCallConsumer + +websocket_urlpatterns = [ + re_path(r"^ws/meetings/(?P[0-9a-f-]+)/call/$", MeetingCallConsumer.as_asgi()), +] diff --git a/training/serializers.py b/training/serializers.py index 683d75c..2edc1f3 100644 --- a/training/serializers.py +++ b/training/serializers.py @@ -2,6 +2,9 @@ from rest_framework import serializers from .models import ( AttendanceStatus, + CallEvent, + CallParticipant, + CallSession, EmployeeAttendance, Meeting, MeetingParticipant, @@ -14,10 +17,7 @@ from .models import ( class TrainingRecordSerializer(serializers.ModelSerializer): class Meta: model = TrainingRecord - fields = [ - "id", "enrollment", "completion_mode", "status", - "online_passed_at", "completed_at", "created_at", "updated_at", - ] + fields = ["id", "enrollment", "completion_mode", "status", "online_passed_at", "completed_at", "created_at", "updated_at"] read_only_fields = fields @@ -41,24 +41,19 @@ class MeetingSerializer(serializers.ModelSerializer): class MeetingParticipantSerializer(serializers.ModelSerializer): + attendance_session_id = serializers.UUIDField(source="attendance_session.id", read_only=True) + class Meta: model = MeetingParticipant - fields = [ - "id", - "meeting", - "user", - "attendance_status", - "attendance_recorded_at", - "attendance_recorded_by", - "created_at", - ] - read_only_fields = [ - "id", - "attendance_status", - "attendance_recorded_at", - "attendance_recorded_by", - "created_at", - ] + fields = ["id", "meeting", "user", "attendance_status", "attendance_recorded_at", "attendance_recorded_by", "attendance_session_id", "created_at"] + read_only_fields = ["id", "attendance_status", "attendance_recorded_at", "attendance_recorded_by", "attendance_session_id", "created_at"] + + +class EmployeeAttendanceSerializer(serializers.ModelSerializer): + class Meta: + model = EmployeeAttendance + fields = ["id", "user", "attendance_date", "status", "meeting", "source", "updated_by", "updated_at", "created_at"] + read_only_fields = ["id", "updated_at", "created_at"] class MeetingAttendanceRequestSerializer(serializers.Serializer): @@ -66,16 +61,40 @@ class MeetingAttendanceRequestSerializer(serializers.Serializer): status = serializers.ChoiceField(choices=AttendanceStatus.choices) -class EmployeeAttendanceSerializer(serializers.ModelSerializer): - meeting_title = serializers.SerializerMethodField() +class CallParticipantSerializer(serializers.ModelSerializer): + user_id = serializers.UUIDField(source="user.id", read_only=True) + user_name = serializers.SerializerMethodField() class Meta: - model = EmployeeAttendance - fields = [ - "id", "user", "attendance_date", "status", - "meeting", "meeting_title", "source", "updated_at", - ] + model = CallParticipant + fields = ["id", "user_id", "user_name", "status", "is_muted", "screen_share_active", "joined_at", "left_at"] read_only_fields = fields - def get_meeting_title(self, obj): - return obj.meeting.title if obj.meeting else None + def get_user_name(self, obj): + return obj.user.get_full_name() or str(obj.user) + + +class CallSessionSerializer(serializers.ModelSerializer): + active_participants = serializers.SerializerMethodField() + + class Meta: + model = CallSession + fields = ["id", "meeting", "status", "started_by", "started_at", "ended_at", "active_participants"] + read_only_fields = fields + + def get_active_participants(self, obj): + from .models import CallParticipantStatus + qs = obj.call_participants.filter(status=CallParticipantStatus.JOINED).select_related("user") + return CallParticipantSerializer(qs, many=True).data + + +class CallEventSerializer(serializers.ModelSerializer): + class Meta: + model = CallEvent + fields = ["id", "event_type", "actor", "target_user", "metadata", "occurred_at"] + read_only_fields = fields + + +class CallModerationRequestSerializer(serializers.Serializer): + action = serializers.ChoiceField(choices=["mute", "unmute", "kick"]) + user_id = serializers.UUIDField() diff --git a/training/services.py b/training/services.py index 69bf461..e1d2402 100644 --- a/training/services.py +++ b/training/services.py @@ -1,10 +1,22 @@ from __future__ import annotations +from django.db.models import Q from django.utils.timezone import now +from attendance.models import AttendanceSession + from .models import ( + CallEvent, + CallEventType, + CallParticipant, + CallParticipantStatus, + CallSession, + CallSessionStatus, CompletionMode, CompletionStatus, + EmployeeAttendance, + Meeting, + MeetingParticipant, SignoffDecision, SignoffEvidence, TrainerSignoff, @@ -13,15 +25,12 @@ from .models import ( class InvalidTransitionError(Exception): - """Raised when a state transition is not allowed from the current status.""" + pass def _require_status(record: TrainingRecord, *allowed: str) -> None: if record.status not in allowed: - raise InvalidTransitionError( - f"Cannot perform action from status '{record.status}'. " - f"Allowed: {', '.join(allowed)}" - ) + raise InvalidTransitionError(f"Cannot perform action from status '{record.status}'. Allowed: {', '.join(allowed)}") def mark_in_progress(record: TrainingRecord) -> TrainingRecord: @@ -32,11 +41,6 @@ def mark_in_progress(record: TrainingRecord) -> TrainingRecord: def mark_online_passed(record: TrainingRecord) -> TrainingRecord: - """ - Signal that the online (quiz) component has been passed. - - Online mode → completed - - Blended mode → pending_signoff (trainer still needed) - """ _require_status(record, CompletionStatus.IN_PROGRESS) record.online_passed_at = now() if record.completion_mode == CompletionMode.ONLINE: @@ -45,51 +49,24 @@ def mark_online_passed(record: TrainingRecord) -> TrainingRecord: elif record.completion_mode == CompletionMode.BLENDED: record.status = CompletionStatus.PENDING_SIGNOFF else: - raise InvalidTransitionError( - "mark_online_passed is only valid for online or blended modes." - ) + raise InvalidTransitionError("mark_online_passed is only valid for online or blended modes.") record.save(update_fields=["online_passed_at", "status", "completed_at", "updated_at"]) return record -def submit_trainer_signoff( - record: TrainingRecord, - trainer, - decision: str, - notes: str = "", - evidence_paths: list[tuple[str, str]] | None = None, -) -> TrainerSignoff: - """ - Record a trainer's signoff decision. - - Offline mode: can be called from IN_PROGRESS; approved → completed - - Blended mode: must be called from PENDING_SIGNOFF; approved → completed - - Rejected: returns to IN_PROGRESS in either mode. - - evidence_paths: list of (file_name, file_path) tuples. - """ +def submit_trainer_signoff(record: TrainingRecord, trainer, decision: str, notes: str = "", evidence_paths=None): if record.completion_mode == CompletionMode.ONLINE: raise InvalidTransitionError("Online-mode training does not require trainer signoff.") - if record.completion_mode == CompletionMode.OFFLINE: _require_status(record, CompletionStatus.IN_PROGRESS, CompletionStatus.PENDING_SIGNOFF) - else: # blended + else: _require_status(record, CompletionStatus.PENDING_SIGNOFF) - signoff = TrainerSignoff.objects.create( - training_record=record, - trainer=trainer, - decision=decision, - notes=notes, - ) + signoff = TrainerSignoff.objects.create(training_record=record, trainer=trainer, decision=decision, notes=notes) if evidence_paths: SignoffEvidence.objects.bulk_create([ - SignoffEvidence( - training_record=record, - file_name=fname, - file_path=fpath, - uploaded_by=trainer, - ) + SignoffEvidence(training_record=record, file_name=fname, file_path=fpath, uploaded_by=trainer) for fname, fpath in evidence_paths ]) @@ -102,3 +79,123 @@ def submit_trainer_signoff( record.save(update_fields=["status", "updated_at"]) return signoff + + +def _find_linked_attendance_session(meeting: Meeting, user_id): + return ( + AttendanceSession.objects.filter(user_id=user_id) + .filter(checked_in_at__date=meeting.starts_at.date()) + .filter(Q(checked_out_at__isnull=True) | Q(checked_out_at__gte=meeting.starts_at)) + .order_by("-checked_in_at") + .first() + ) + + +def record_meeting_attendance(meeting: Meeting, *, user_id, status_value: str, actor): + participant = MeetingParticipant.objects.get(meeting=meeting, user_id=user_id) + participant.attendance_status = status_value + participant.attendance_recorded_at = now() + participant.attendance_recorded_by = actor + participant.attendance_session = _find_linked_attendance_session(meeting, user_id) + participant.save(update_fields=["attendance_status", "attendance_recorded_at", "attendance_recorded_by", "attendance_session"]) + + EmployeeAttendance.objects.update_or_create( + user_id=user_id, + attendance_date=meeting.starts_at.date(), + meeting=meeting, + defaults={"status": status_value, "source": "meeting", "updated_by": actor}, + ) + return participant + + +# --------------------------------------------------------------------------- +# Call session management +# --------------------------------------------------------------------------- + + +class CallSessionError(Exception): + pass + + +def get_or_create_call_session(meeting: Meeting, actor) -> CallSession: + session, _ = CallSession.objects.get_or_create( + meeting=meeting, + defaults={"started_by": actor, "status": CallSessionStatus.ACTIVE}, + ) + if session.status == CallSessionStatus.ENDED: + raise CallSessionError("This meeting's call session has already ended.") + return session + + +def end_call_session(session: CallSession, actor) -> CallSession: + if session.status == CallSessionStatus.ENDED: + raise CallSessionError("Call session is already ended.") + session.status = CallSessionStatus.ENDED + session.ended_at = now() + session.save(update_fields=["status", "ended_at"]) + + CallParticipant.objects.filter( + session=session, + status=CallParticipantStatus.JOINED, + ).update(status=CallParticipantStatus.LEFT, left_at=now()) + + CallEvent.objects.create( + session=session, + event_type=CallEventType.SESSION_ENDED, + actor=actor, + ) + return session + + +def apply_moderation_action(session: CallSession, action: str, target_user_id, actor) -> None: + from django.contrib.auth import get_user_model + + User = get_user_model() + + try: + target = User.objects.get(pk=target_user_id) + except User.DoesNotExist: + raise CallSessionError("Target user not found.") + + if action == "kick": + CallParticipant.objects.filter( + session=session, + user=target, + status=CallParticipantStatus.JOINED, + ).update(status=CallParticipantStatus.LEFT, left_at=now()) + CallEvent.objects.create( + session=session, + event_type=CallEventType.KICK, + actor=actor, + target_user=target, + ) + elif action in ("mute", "unmute"): + is_muted = action == "mute" + CallParticipant.objects.filter( + session=session, + user=target, + status=CallParticipantStatus.JOINED, + ).update(is_muted=is_muted) + event_type = CallEventType.MUTE if is_muted else CallEventType.UNMUTE + CallEvent.objects.create( + session=session, + event_type=event_type, + actor=actor, + target_user=target, + ) + else: + raise CallSessionError(f"Unknown action: {action}") + + +def toggle_screen_share(session: CallSession, actor, *, start: bool) -> CallParticipant: + """Start or stop the actor's own screen share. Caller must hold meeting:moderate.""" + cp, _ = CallParticipant.objects.get_or_create( + session=session, + user=actor, + defaults={"status": CallParticipantStatus.JOINED}, + ) + cp.screen_share_active = start + cp.save(update_fields=["screen_share_active"]) + event_type = CallEventType.SCREEN_SHARE_START if start else CallEventType.SCREEN_SHARE_STOP + CallEvent.objects.create(session=session, event_type=event_type, actor=actor) + return cp diff --git a/training/tests/test_groupcall_integration.py b/training/tests/test_groupcall_integration.py new file mode 100644 index 0000000..0474af1 --- /dev/null +++ b/training/tests/test_groupcall_integration.py @@ -0,0 +1,469 @@ +""" +Integration tests for TRA-372: Live Groupcall Backend. + +Covers: +- REST: start/get/end call session +- REST: moderation actions (mute, unmute, kick) +- REST: call event audit log +- WebSocket: connect/disconnect, P2P signaling relay, instructor moderation +""" + +import json +import uuid + +import pytest +from channels.db import database_sync_to_async +from channels.testing import WebsocketCommunicator +from django.utils import timezone +from rest_framework.test import APIClient + +from accounts.tests.factories import AccountUserFactory, RoleFactory, UserRoleBindingFactory +from training.models import ( + CallEvent, + CallEventType, + CallParticipant, + CallParticipantStatus, + CallSession, + CallSessionStatus, + Meeting, + MeetingParticipant, +) + +pytestmark = pytest.mark.django_db(transaction=True) + + +# --------------------------------------------------------------------------- +# helpers +# --------------------------------------------------------------------------- + + +def make_trainer(): + user = AccountUserFactory() + role = RoleFactory(slug="trainer", name="Trainer") + UserRoleBindingFactory(user=user, role=role) + return user + + +def make_meeting(title="Test Meeting"): + return Meeting.objects.create(title=title, starts_at=timezone.now()) + + +def add_participant(meeting, user): + return MeetingParticipant.objects.create(meeting=meeting, user=user) + + +def trainer_client(trainer): + c = APIClient() + c.force_authenticate(user=trainer) + return c + + +# --------------------------------------------------------------------------- +# REST: call session lifecycle +# --------------------------------------------------------------------------- + + +class TestCallSessionREST: + def test_trainer_can_start_session(self): + trainer = make_trainer() + meeting = make_meeting() + client = trainer_client(trainer) + + resp = client.post(f"/api/v1/training/meetings/{meeting.id}/call/") + assert resp.status_code == 201 + data = resp.json() + assert data["status"] == "active" + assert data["meeting"] == str(meeting.id) + + def test_non_trainer_cannot_start_session(self): + learner = AccountUserFactory() + meeting = make_meeting() + add_participant(meeting, learner) + client = APIClient() + client.force_authenticate(user=learner) + + resp = client.post(f"/api/v1/training/meetings/{meeting.id}/call/") + assert resp.status_code == 403 + + def test_start_session_idempotent(self): + trainer = make_trainer() + meeting = make_meeting() + client = trainer_client(trainer) + + resp1 = client.post(f"/api/v1/training/meetings/{meeting.id}/call/") + resp2 = client.post(f"/api/v1/training/meetings/{meeting.id}/call/") + assert resp1.status_code == 201 + assert resp2.status_code == 201 + assert resp1.json()["id"] == resp2.json()["id"] + + def test_get_session_returns_state(self): + trainer = make_trainer() + meeting = make_meeting() + client = trainer_client(trainer) + client.post(f"/api/v1/training/meetings/{meeting.id}/call/") + + resp = client.get(f"/api/v1/training/meetings/{meeting.id}/call/") + assert resp.status_code == 200 + assert resp.json()["status"] == "active" + + def test_get_nonexistent_session_returns_404(self): + trainer = make_trainer() + meeting = make_meeting() + client = trainer_client(trainer) + + resp = client.get(f"/api/v1/training/meetings/{meeting.id}/call/") + assert resp.status_code == 404 + + def test_trainer_can_end_session(self): + trainer = make_trainer() + meeting = make_meeting() + client = trainer_client(trainer) + client.post(f"/api/v1/training/meetings/{meeting.id}/call/") + + resp = client.delete(f"/api/v1/training/meetings/{meeting.id}/call/") + assert resp.status_code == 200 + assert resp.json()["status"] == "ended" + + def test_ended_session_cannot_be_ended_again(self): + trainer = make_trainer() + meeting = make_meeting() + client = trainer_client(trainer) + client.post(f"/api/v1/training/meetings/{meeting.id}/call/") + client.delete(f"/api/v1/training/meetings/{meeting.id}/call/") + + resp = client.delete(f"/api/v1/training/meetings/{meeting.id}/call/") + assert resp.status_code == 409 + + def test_end_session_sets_participants_left(self): + trainer = make_trainer() + learner = AccountUserFactory() + meeting = make_meeting() + add_participant(meeting, learner) + client = trainer_client(trainer) + resp = client.post(f"/api/v1/training/meetings/{meeting.id}/call/") + session = CallSession.objects.get(pk=resp.json()["id"]) + CallParticipant.objects.create(session=session, user=learner, status=CallParticipantStatus.JOINED) + + client.delete(f"/api/v1/training/meetings/{meeting.id}/call/") + + assert not CallParticipant.objects.filter( + session=session, status=CallParticipantStatus.JOINED + ).exists() + + +# --------------------------------------------------------------------------- +# REST: moderation +# --------------------------------------------------------------------------- + + +class TestCallModerationREST: + def _setup(self): + trainer = make_trainer() + learner = AccountUserFactory() + meeting = make_meeting() + add_participant(meeting, learner) + client = trainer_client(trainer) + resp = client.post(f"/api/v1/training/meetings/{meeting.id}/call/") + session = CallSession.objects.get(pk=resp.json()["id"]) + CallParticipant.objects.create(session=session, user=learner, status=CallParticipantStatus.JOINED) + return trainer, learner, meeting, session, client + + def test_trainer_can_mute_participant(self): + trainer, learner, meeting, session, client = self._setup() + + resp = client.post( + f"/api/v1/training/meetings/{meeting.id}/call/moderate/", + {"action": "mute", "user_id": str(learner.id)}, + format="json", + ) + assert resp.status_code == 200 + assert CallParticipant.objects.get(session=session, user=learner).is_muted is True + assert CallEvent.objects.filter(session=session, event_type=CallEventType.MUTE).exists() + + def test_trainer_can_unmute_participant(self): + trainer, learner, meeting, session, client = self._setup() + CallParticipant.objects.filter(session=session, user=learner).update(is_muted=True) + + resp = client.post( + f"/api/v1/training/meetings/{meeting.id}/call/moderate/", + {"action": "unmute", "user_id": str(learner.id)}, + format="json", + ) + assert resp.status_code == 200 + assert CallParticipant.objects.get(session=session, user=learner).is_muted is False + + def test_trainer_can_kick_participant(self): + trainer, learner, meeting, session, client = self._setup() + + resp = client.post( + f"/api/v1/training/meetings/{meeting.id}/call/moderate/", + {"action": "kick", "user_id": str(learner.id)}, + format="json", + ) + assert resp.status_code == 200 + assert CallParticipant.objects.get(session=session, user=learner).status == CallParticipantStatus.LEFT + assert CallEvent.objects.filter(session=session, event_type=CallEventType.KICK).exists() + + def test_non_trainer_cannot_moderate(self): + trainer, learner, meeting, session, _ = self._setup() + other = AccountUserFactory() + client = APIClient() + client.force_authenticate(user=other) + + resp = client.post( + f"/api/v1/training/meetings/{meeting.id}/call/moderate/", + {"action": "mute", "user_id": str(learner.id)}, + format="json", + ) + assert resp.status_code == 403 + + def test_moderate_unknown_user_returns_400(self): + trainer, learner, meeting, session, client = self._setup() + + resp = client.post( + f"/api/v1/training/meetings/{meeting.id}/call/moderate/", + {"action": "mute", "user_id": str(uuid.uuid4())}, + format="json", + ) + assert resp.status_code == 400 + + +# --------------------------------------------------------------------------- +# REST: audit events +# --------------------------------------------------------------------------- + + +class TestCallEventsREST: + def test_trainer_can_retrieve_audit_events(self): + trainer = make_trainer() + learner = AccountUserFactory() + meeting = make_meeting() + add_participant(meeting, learner) + client = trainer_client(trainer) + client.post(f"/api/v1/training/meetings/{meeting.id}/call/") + + resp = client.get(f"/api/v1/training/meetings/{meeting.id}/call/events/") + assert resp.status_code == 200 + assert isinstance(resp.json(), list) + + def test_non_trainer_cannot_retrieve_audit_events(self): + trainer = make_trainer() + learner = AccountUserFactory() + meeting = make_meeting() + add_participant(meeting, learner) + trainer_client(trainer).post(f"/api/v1/training/meetings/{meeting.id}/call/") + + client = APIClient() + client.force_authenticate(user=learner) + resp = client.get(f"/api/v1/training/meetings/{meeting.id}/call/events/") + assert resp.status_code == 403 + + +# --------------------------------------------------------------------------- +# WebSocket: signaling and moderation +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +class TestMeetingCallConsumer: + async def _make_app(self): + from config.asgi import application + return application + + async def _connect(self, app, user, meeting_id): + from rest_framework_simplejwt.tokens import AccessToken + token = str(AccessToken.for_user(user)) + url = f"/ws/meetings/{meeting_id}/call/?token={token}" + comm = WebsocketCommunicator(app, url) + connected, _ = await comm.connect() + return comm, connected + + @database_sync_to_async + def _create_fixtures(self): + trainer = make_trainer() + learner = AccountUserFactory() + meeting = make_meeting() + add_participant(meeting, learner) + return trainer, learner, meeting + + async def test_unauthenticated_connection_rejected(self): + app = await self._make_app() + meeting = await database_sync_to_async(make_meeting)() + comm = WebsocketCommunicator(app, f"/ws/meetings/{meeting.id}/call/") + connected, code = await comm.connect() + assert not connected + assert code == 4001 + await comm.disconnect() + + async def test_non_participant_connection_rejected(self): + app = await self._make_app() + stranger = await database_sync_to_async(AccountUserFactory)() + meeting = await database_sync_to_async(make_meeting)() + + comm, connected = await self._connect(app, stranger, meeting.id) + assert not connected + await comm.disconnect() + + async def test_participant_can_connect_and_receives_session_state(self): + app = await self._make_app() + trainer, learner, meeting = await self._create_fixtures() + + comm, connected = await self._connect(app, learner, meeting.id) + assert connected + + msg = json.loads(await comm.receive_from()) + # first message from self is own join broadcast + # second message is session_state + if msg["type"] == "participant_joined": + msg = json.loads(await comm.receive_from()) + assert msg["type"] == "session_state" + assert "participants" in msg + + await comm.disconnect() + + async def test_participant_join_broadcast_to_group(self): + app = await self._make_app() + trainer, learner, meeting = await self._create_fixtures() + + learner2 = await database_sync_to_async(AccountUserFactory)() + await database_sync_to_async(add_participant)(meeting, learner2) + + comm1, _ = await self._connect(app, learner, meeting.id) + # drain initial messages for comm1 + await comm1.receive_from() # own join broadcast + await comm1.receive_from() # session_state + + comm2, _ = await self._connect(app, learner2, meeting.id) + + # comm1 should receive a participant_joined event for learner2 + msg = json.loads(await comm1.receive_from()) + assert msg["type"] == "participant_joined" + assert msg["user_id"] == str(learner2.id) + + await comm1.disconnect() + await comm2.disconnect() + + async def test_signaling_relay_offer_forwarded_to_target(self): + app = await self._make_app() + trainer, learner, meeting = await self._create_fixtures() + learner2 = await database_sync_to_async(AccountUserFactory)() + await database_sync_to_async(add_participant)(meeting, learner2) + + comm1, _ = await self._connect(app, learner, meeting.id) + comm2, _ = await self._connect(app, learner2, meeting.id) + + # drain setup messages + for _ in range(3): + await comm1.receive_from() + for _ in range(2): + await comm2.receive_from() + + await comm1.send_to(text_data=json.dumps({ + "type": "offer", + "to": str(learner2.id), + "sdp": {"type": "offer", "sdp": "v=0..."}, + })) + + msg = json.loads(await comm2.receive_from()) + assert msg["type"] == "offer" + assert msg["from"] == str(learner.id) + assert msg["sdp"]["type"] == "offer" + + await comm1.disconnect() + await comm2.disconnect() + + async def test_instructor_mute_broadcasts_to_group(self): + app = await self._make_app() + trainer, learner, meeting = await self._create_fixtures() + + comm_trainer, _ = await self._connect(app, trainer, meeting.id) + comm_learner, _ = await self._connect(app, learner, meeting.id) + + # drain setup messages + for _ in range(3): + await comm_trainer.receive_from() + for _ in range(2): + await comm_learner.receive_from() + + await comm_trainer.send_to(text_data=json.dumps({ + "type": "mute", + "user_id": str(learner.id), + })) + + msg = json.loads(await comm_learner.receive_from()) + assert msg["type"] == "muted" + assert msg["user_id"] == str(learner.id) + + await comm_trainer.disconnect() + await comm_learner.disconnect() + + async def test_instructor_kick_closes_target_connection(self): + app = await self._make_app() + trainer, learner, meeting = await self._create_fixtures() + + comm_trainer, _ = await self._connect(app, trainer, meeting.id) + comm_learner, _ = await self._connect(app, learner, meeting.id) + + # drain setup messages + for _ in range(3): + await comm_trainer.receive_from() + for _ in range(2): + await comm_learner.receive_from() + + await comm_trainer.send_to(text_data=json.dumps({ + "type": "kick", + "user_id": str(learner.id), + })) + + msg = json.loads(await comm_learner.receive_from()) + assert msg["type"] == "kicked" + assert msg["user_id"] == str(learner.id) + + await comm_trainer.disconnect() + await comm_learner.disconnect() + + async def test_non_instructor_mute_rejected(self): + app = await self._make_app() + trainer, learner, meeting = await self._create_fixtures() + learner2 = await database_sync_to_async(AccountUserFactory)() + await database_sync_to_async(add_participant)(meeting, learner2) + + comm1, _ = await self._connect(app, learner, meeting.id) + # drain + for _ in range(2): + await comm1.receive_from() + + await comm1.send_to(text_data=json.dumps({ + "type": "mute", + "user_id": str(learner2.id), + })) + + msg = json.loads(await comm1.receive_from()) + assert msg["type"] == "error" + assert msg["code"] == "not_authorized" + + await comm1.disconnect() + + async def test_disconnect_records_leave_event(self): + app = await self._make_app() + trainer, learner, meeting = await self._create_fixtures() + + comm, connected = await self._connect(app, learner, meeting.id) + # drain + for _ in range(2): + await comm.receive_from() + + session = await database_sync_to_async( + lambda: meeting.call_session + )() + await comm.disconnect() + + left = await database_sync_to_async( + lambda: CallParticipant.objects.get(session=session, user=learner).status + )() + assert left == CallParticipantStatus.LEFT + + event_exists = await database_sync_to_async( + lambda: CallEvent.objects.filter(session=session, event_type=CallEventType.LEAVE).exists() + )() + assert event_exists diff --git a/training/urls.py b/training/urls.py index 2c04225..06c178b 100644 --- a/training/urls.py +++ b/training/urls.py @@ -5,6 +5,10 @@ from .views import ( MarkInProgressView, MarkOnlinePassedView, MeetingAttendanceView, + MeetingCallEventsView, + MeetingCallModerationView, + MeetingCallScreenShareView, + MeetingCallSessionView, MeetingDetailView, MeetingListCreateView, MeetingParticipantsCreateView, @@ -26,4 +30,8 @@ urlpatterns = [ path("meetings//participants/", MeetingParticipantsCreateView.as_view(), name="meeting-participants-create"), path("meetings//attendance/", MeetingAttendanceView.as_view(), name="meeting-attendance"), path("attendance/", EmployeeAttendanceListView.as_view(), name="employee-attendance-list"), + path("meetings//call/", MeetingCallSessionView.as_view(), name="meeting-call-session"), + path("meetings//call/moderate/", MeetingCallModerationView.as_view(), name="meeting-call-moderate"), + path("meetings//call/screen-share/", MeetingCallScreenShareView.as_view(), name="meeting-call-screen-share"), + path("meetings//call/events/", MeetingCallEventsView.as_view(), name="meeting-call-events"), ] diff --git a/training/views.py b/training/views.py index a2959e6..36a5893 100644 --- a/training/views.py +++ b/training/views.py @@ -4,9 +4,13 @@ from rest_framework.permissions import IsAuthenticated from rest_framework.response import Response from rest_framework.views import APIView -from .models import EmployeeAttendance, Meeting, MeetingParticipant, TrainingRecord -from .permissions import IsTrainer +from .models import CallSession, EmployeeAttendance, Meeting, MeetingParticipant, TrainingRecord +from .permissions import IsMeetingModerator, IsTrainer from .serializers import ( + CallEventSerializer, + CallModerationRequestSerializer, + CallParticipantSerializer, + CallSessionSerializer, EmployeeAttendanceSerializer, MeetingAttendanceRequestSerializer, MeetingParticipantSerializer, @@ -16,11 +20,16 @@ from .serializers import ( TrainingRecordSerializer, ) from .services import ( + CallSessionError, InvalidTransitionError, + apply_moderation_action, + end_call_session, + get_or_create_call_session, mark_in_progress, mark_online_passed, record_meeting_attendance, submit_trainer_signoff, + toggle_screen_share, ) @@ -160,6 +169,168 @@ class MeetingDetailView(APIView): return Response(data) +class MeetingCallSessionView(APIView): + """ + GET /meetings/{id}/call/ — get current call session state (participants + trainer) + POST /meetings/{id}/call/ — start or resume a call session (trainer only) + DELETE /meetings/{id}/call/ — end a call session (trainer only); sends session_ended to all WS clients + """ + + def get_permissions(self): + if self.request.method in ("POST", "DELETE"): + return [IsAuthenticated(), IsMeetingModerator()] + return [IsAuthenticated()] + + def get(self, request, meeting_id): + meeting = get_object_or_404(Meeting, pk=meeting_id) + try: + session = meeting.call_session + except CallSession.DoesNotExist: + return Response({"detail": "No call session exists for this meeting."}, status=status.HTTP_404_NOT_FOUND) + return Response(CallSessionSerializer(session).data) + + def post(self, request, meeting_id): + meeting = get_object_or_404(Meeting, pk=meeting_id) + try: + session = get_or_create_call_session(meeting, actor=request.user) + except CallSessionError as exc: + return Response({"detail": str(exc)}, status=status.HTTP_409_CONFLICT) + return Response(CallSessionSerializer(session).data, status=status.HTTP_201_CREATED) + + def delete(self, request, meeting_id): + meeting = get_object_or_404(Meeting, pk=meeting_id) + try: + session = meeting.call_session + except CallSession.DoesNotExist: + return Response({"detail": "No call session found."}, status=status.HTTP_404_NOT_FOUND) + try: + session = end_call_session(session, actor=request.user) + except CallSessionError as exc: + return Response({"detail": str(exc)}, status=status.HTTP_409_CONFLICT) + + _broadcast_session_ended(str(meeting.id)) + return Response(CallSessionSerializer(session).data) + + +class MeetingCallModerationView(APIView): + """ + POST /meetings/{id}/call/moderate/ + Body: {"action": "mute"|"unmute"|"kick", "user_id": ""} + Moderator-only (meeting:moderate capability). Writes audit event + broadcasts WS event. + """ + + permission_classes = [IsAuthenticated, IsMeetingModerator] + + def post(self, request, meeting_id): + meeting = get_object_or_404(Meeting, pk=meeting_id) + try: + session = meeting.call_session + except CallSession.DoesNotExist: + return Response({"detail": "No active call session."}, status=status.HTTP_404_NOT_FOUND) + + serializer = CallModerationRequestSerializer(data=request.data) + serializer.is_valid(raise_exception=True) + + action = serializer.validated_data["action"] + target_user_id = serializer.validated_data["user_id"] + + try: + apply_moderation_action(session, action, target_user_id, actor=request.user) + except CallSessionError as exc: + return Response({"detail": str(exc)}, status=status.HTTP_400_BAD_REQUEST) + + _broadcast_moderation(str(meeting.id), action, str(target_user_id), str(request.user.id)) + return Response({"detail": f"{action} applied."}) + + +class MeetingCallScreenShareView(APIView): + """ + POST /meetings/{id}/call/screen-share/ + Body: {"action": "start"|"stop"} + Moderator-only. Only users with meeting:moderate can start/stop screen sharing. + """ + + permission_classes = [IsAuthenticated, IsMeetingModerator] + + def post(self, request, meeting_id): + meeting = get_object_or_404(Meeting, pk=meeting_id) + try: + session = meeting.call_session + except CallSession.DoesNotExist: + return Response({"detail": "No active call session."}, status=status.HTTP_404_NOT_FOUND) + + action = request.data.get("action") + if action not in ("start", "stop"): + return Response({"detail": "action must be 'start' or 'stop'."}, status=status.HTTP_400_BAD_REQUEST) + + try: + cp = toggle_screen_share(session, actor=request.user, start=(action == "start")) + except CallSessionError as exc: + return Response({"detail": str(exc)}, status=status.HTTP_400_BAD_REQUEST) + + _broadcast_screen_share(str(meeting.id), action, str(request.user.id)) + return Response(CallParticipantSerializer(cp).data) + + +class MeetingCallEventsView(APIView): + """ + GET /meetings/{id}/call/events/ — audit log for moderation/session events (moderator only) + """ + + permission_classes = [IsAuthenticated, IsMeetingModerator] + + def get(self, request, meeting_id): + meeting = get_object_or_404(Meeting, pk=meeting_id) + try: + session = meeting.call_session + except CallSession.DoesNotExist: + return Response({"detail": "No call session found."}, status=status.HTTP_404_NOT_FOUND) + events = session.events.select_related("actor", "target_user") + return Response(CallEventSerializer(events, many=True).data) + + +def _broadcast_session_ended(meeting_id: str): + from asgiref.sync import async_to_sync + from channels.layers import get_channel_layer + + channel_layer = get_channel_layer() + if channel_layer is None: + return + group = f"meeting_call_{meeting_id}" + async_to_sync(channel_layer.group_send)(group, {"type": "call.session_ended"}) + + +def _broadcast_moderation(meeting_id: str, action: str, user_id: str, by: str): + from asgiref.sync import async_to_sync + from channels.layers import get_channel_layer + + channel_layer = get_channel_layer() + if channel_layer is None: + return + group = f"meeting_call_{meeting_id}" + + if action == "kick": + msg = {"type": "call.kicked", "user_id": user_id, "kicked_by": by} + elif action == "mute": + msg = {"type": "call.muted", "user_id": user_id, "by": by} + else: + msg = {"type": "call.unmuted", "user_id": user_id, "by": by} + + async_to_sync(channel_layer.group_send)(group, msg) + + +def _broadcast_screen_share(meeting_id: str, action: str, by: str): + from asgiref.sync import async_to_sync + from channels.layers import get_channel_layer + + channel_layer = get_channel_layer() + if channel_layer is None: + return + group = f"meeting_call_{meeting_id}" + msg_type = "call.screen_share_started" if action == "start" else "call.screen_share_stopped" + async_to_sync(channel_layer.group_send)(group, {"type": msg_type, "by": by}) + + class EmployeeAttendanceListView(APIView): """ GET /training/attendance/ — own attendance history