feat: add audit gate to CourseChatView

learning_assistant/api.py

@@ -2,6 +2,7 @@
 Library for the learning_assistant app.
 """
 import logging
+from datetime import datetime, timedelta
 
 from django.conf import settings
 from django.contrib.auth import get_user_model
@@ -10,9 +11,13 @@
 from jinja2 import BaseLoader, Environment
 from opaque_keys import InvalidKeyError
 
-from learning_assistant.constants import ACCEPTED_CATEGORY_TYPES, CATEGORY_TYPE_MAP
+from learning_assistant.constants import ACCEPTED_CATEGORY_TYPES, AUDIT_TRIAL_MAX_DAYS, CATEGORY_TYPE_MAP
 from learning_assistant.data import LearningAssistantCourseEnabledData
-from learning_assistant.models import LearningAssistantCourseEnabled, LearningAssistantMessage
+from learning_assistant.models import (
+    LearningAssistantAuditTrial,
+    LearningAssistantCourseEnabled,
+    LearningAssistantMessage,
+)
 from learning_assistant.platform_imports import (
     block_get_children,
     block_leaf_filter,
@@ -224,3 +229,24 @@ def get_message_history(courserun_key, user, message_count):
     message_history = list(LearningAssistantMessage.objects.filter(
         course_id=courserun_key, user=user).order_by('-created')[:message_count])[::-1]
     return message_history
+
+
+def audit_trial_is_expired(user, upgrade_deadline):
+    """
+    Given a user (User), get or create the corresponding LearningAssistantAuditTrial trial object.
+    """
+    # If the upgrade deadline has passed, return "True" for expired
+    DAYS_SINCE_UPGRADE_DEADLINE = datetime.now() - upgrade_deadline
+    if DAYS_SINCE_UPGRADE_DEADLINE >= timedelta(days=0):
+        return True
+
+    audit_trial, _ = LearningAssistantAuditTrial.objects.get_or_create(
+        user=user,
+        defaults={
+            "start_date": datetime.now(),
+        },
+    )
+
+    # If the user's trial is past its expiry date, return "True" for expired. Else, return False
+    DAYS_SINCE_TRIAL_START_DATE = datetime.now() - audit_trial.start_date
+    return DAYS_SINCE_TRIAL_START_DATE >= timedelta(days=AUDIT_TRIAL_MAX_DAYS)

learning_assistant/constants.py

@@ -14,3 +14,5 @@
     "html": "TEXT",
     "video": "VIDEO",
 }
+
+AUDIT_TRIAL_MAX_DAYS = 14

learning_assistant/platform_imports.py

@@ -8,7 +8,7 @@
 
 def get_text_transcript(video_block):
     """Get the transcript for a video block in text format, or None."""
-    # pylint: disable=import-error, import-outside-toplevel
+    # pylint: disable=import-outside-toplevel
     from xmodule.exceptions import NotFoundError
     from xmodule.video_block.transcripts_utils import get_transcript
     try:
@@ -21,28 +21,28 @@ def get_text_transcript(video_block):
 
 def get_single_block(request, user_id, course_id, usage_key_string, course=None):
     """Load a single xblock."""
-    # pylint: disable=import-error, import-outside-toplevel
+    # pylint: disable=import-outside-toplevel
     from lms.djangoapps.courseware.block_render import load_single_xblock
     return load_single_xblock(request, user_id, course_id, usage_key_string, course)
 
 
 def traverse_block_pre_order(start_node, get_children, filter_func=None):
     """Traverse a DAG or tree in pre-order."""
-    # pylint: disable=import-error, import-outside-toplevel
+    # pylint: disable=import-outside-toplevel
     from openedx.core.lib.graph_traversals import traverse_pre_order
     return traverse_pre_order(start_node, get_children, filter_func)
 
 
 def block_leaf_filter(block):
     """Return only leaf nodes."""
-    # pylint: disable=import-error, import-outside-toplevel
+    # pylint: disable=import-outside-toplevel
     from openedx.core.lib.graph_traversals import leaf_filter
     return leaf_filter(block)
 
 
 def block_get_children(block):
     """Return children of a given block."""
-    # pylint: disable=import-error, import-outside-toplevel
+    # pylint: disable=import-outside-toplevel
     from openedx.core.lib.graph_traversals import get_children
     return get_children(block)
 
@@ -54,7 +54,7 @@ def get_cache_course_run_data(course_run_id, fields):
     This function makes use of the course run cache in the LMS, which caches data from the discovery service. This is
     necessary because only the discovery service stores the relation between courseruns and courses.
     """
-    # pylint: disable=import-error, import-outside-toplevel
+    # pylint: disable=import-outside-toplevel
     from openedx.core.djangoapps.catalog.utils import get_course_run_data
     return get_course_run_data(course_run_id, fields)
 
@@ -66,7 +66,7 @@ def get_cache_course_data(course_id, fields):
     This function makes use of the course cache in the LMS, which caches data from the discovery service. This is
     necessary because only the discovery service stores course skills data.
     """
-    # pylint: disable=import-error, import-outside-toplevel
+    # pylint: disable=import-outside-toplevel
     from openedx.core.djangoapps.catalog.utils import get_course_data
     return get_course_data(course_id, fields)
 
@@ -82,6 +82,6 @@ def get_user_role(user, course_key):
     Returns:
         * str: the user's role
     """
-    # pylint: disable=import-error, import-outside-toplevel
+    # pylint: disable=import-outside-toplevel
     from lms.djangoapps.courseware.access import get_user_role as platform_get_user_role
     return platform_get_user_role(user, course_key)

learning_assistant/views.py

@@ -21,6 +21,7 @@
     pass
 
 from learning_assistant.api import (
+    audit_trial_is_expired,
     get_course_id,
     get_message_history,
     learning_assistant_enabled,
@@ -43,46 +44,10 @@
     authentication_classes = (SessionAuthentication, JwtAuthentication,)
     permission_classes = (IsAuthenticated,)
 
-    def post(self, request, course_run_id):
+    def _get_next_message(self, request, courserun_key, course_run_id):
         """
-        Given a course run ID, retrieve a chat response for that course.
-
-        Expected POST data: {
-            [
-                {'role': 'user', 'content': 'What is 2+2?'},
-                {'role': 'assistant', 'content': '4'}
-            ]
-        }
+        Generate the next message to be returned by the learning assistant.
         """
-        try:
-            courserun_key = CourseKey.from_string(course_run_id)
-        except InvalidKeyError:
-            return Response(
-                status=http_status.HTTP_400_BAD_REQUEST,
-                data={'detail': 'Course ID is not a valid course ID.'}
-            )
-
-        if not learning_assistant_enabled(courserun_key):
-            return Response(
-                status=http_status.HTTP_403_FORBIDDEN,
-                data={'detail': 'Learning assistant not enabled for course.'}
-            )
-
-        # If user does not have an enrollment record, or is not staff, they should not have access
-        user_role = get_user_role(request.user, courserun_key)
-        enrollment_object = CourseEnrollment.get_enrollment(request.user, courserun_key)
-        enrollment_mode = enrollment_object.mode if enrollment_object else None
-        if (
-            (enrollment_mode not in CourseMode.VERIFIED_MODES)
-            and not user_role_is_staff(user_role)
-        ):
-            return Response(
-                status=http_status.HTTP_403_FORBIDDEN,
-                data={'detail': 'Must be staff or have valid enrollment.'}
-            )
-
-        unit_id = request.query_params.get('unit_id')
-
         message_list = request.data
 
         # Check that the last message in the list corresponds to a user
@@ -117,8 +82,8 @@
         )
 
         course_id = get_course_id(course_run_id)
-
         template_string = getattr(settings, 'LEARNING_ASSISTANT_PROMPT_TEMPLATE', '')
+        unit_id = request.query_params.get('unit_id')
 
         prompt_template = render_prompt_template(
             request, request.user.id, course_run_id, unit_id, course_id, template_string
@@ -130,6 +95,68 @@
 
         return Response(status=status_code, data=message)
 
+    def post(self, request, course_run_id):
+        """
+        Given a course run ID, retrieve a chat response for that course.
+
+        Expected POST data: {
+            [
+                {'role': 'user', 'content': 'What is 2+2?'},
+                {'role': 'assistant', 'content': '4'}
+            ]
+        }
+        """
+        try:
+            courserun_key = CourseKey.from_string(course_run_id)
+        except InvalidKeyError:
+            return Response(
+                status=http_status.HTTP_400_BAD_REQUEST,
+                data={'detail': 'Course ID is not a valid course ID.'}
+            )
+
+        if not learning_assistant_enabled(courserun_key):
+            return Response(
+                status=http_status.HTTP_403_FORBIDDEN,
+                data={'detail': 'Learning assistant not enabled for course.'}
+            )
+
+        # If user does not have a verified enrollment record, or is not staff, they should not have full access
+        user_role = get_user_role(request.user, courserun_key)
+        enrollment_object = CourseEnrollment.get_enrollment(request.user, courserun_key)
+        enrollment_mode = enrollment_object.mode if enrollment_object else None
+
+        # If the user is in a verified course mode or is staff, return the next message
+        if (
+            # Here we include CREDIT_MODE and NO_ID_PROFESSIONAL_MODE, as CourseMode.VERIFIED_MODES on its own
+            # doesn't match what we count as "verified modes" in the frontend component.
+            enrollment_mode in CourseMode.VERIFIED_MODES + CourseMode.CREDIT_MODE + CourseMode.NO_ID_PROFESSIONAL_MODE
+            or user_role_is_staff(user_role)
+        ):
+            return self._get_next_message(request, courserun_key, course_run_id)
+
+        # If user has an audit enrollment record, get or create their trial. If the trial is not expired, return the
+        # next message. Otherwise, return 403
+        elif enrollment_mode in CourseMode.UPSELL_TO_VERIFIED_MODES:  # AUDIT, HONOR
+            course_mode = CourseMode.objects.get(course=courserun_key)
+            upgrade_deadline = course_mode.expiration_datetime()
+
+            user_audit_trial_expired = audit_trial_is_expired(request.user, upgrade_deadline)
+            if user_audit_trial_expired:
+                return Response(
+                    status=http_status.HTTP_403_FORBIDDEN,
+                    data={'detail': 'The audit trial for this user has expired.'}
+                )
+            else:
+                return self._get_next_message(request, courserun_key, course_run_id)
+
+        # If user has a course mode that is not verified & not meant to access to the learning assistant, return 403
+        # This covers the other course modes: UNPAID_EXECUTIVE_EDUCATION & UNPAID_BOOTCAMP
+        else:
+            return Response(
+                status=http_status.HTTP_403_FORBIDDEN,
+                data={'detail': 'Must be staff or have valid enrollment.'}
+            )
+
 
 class LearningAssistantEnabledView(APIView):
     """

tests/test_api.py

@@ -2,6 +2,7 @@
 Test cases for the learning-assistant api module.
 """
 import itertools
+from datetime import datetime, timedelta
 from unittest.mock import MagicMock, patch
 
 import ddt
@@ -16,6 +17,7 @@
     _extract_block_contents,
     _get_children_contents,
     _leaf_filter,
+    audit_trial_is_expired,
     get_block_content,
     get_message_history,
     learning_assistant_available,
@@ -24,8 +26,13 @@
     save_chat_message,
     set_learning_assistant_enabled,
 )
+from learning_assistant.constants import AUDIT_TRIAL_MAX_DAYS
 from learning_assistant.data import LearningAssistantCourseEnabledData
-from learning_assistant.models import LearningAssistantCourseEnabled, LearningAssistantMessage
+from learning_assistant.models import (
+    LearningAssistantAuditTrial,
+    LearningAssistantCourseEnabled,
+    LearningAssistantMessage,
+)
 
 fake_transcript = 'This is the text version from the transcript'
 User = get_user_model()
@@ -241,6 +248,7 @@ class TestLearningAssistantCourseEnabledApi(TestCase):
     """
     Test suite for save_chat_message.
     """
+
     def setUp(self):
         super().setUp()
 
@@ -473,3 +481,37 @@ def test_get_message_course_id_differences(self):
             self.assertEqual(return_value.user, expected_value[i].user)
             self.assertEqual(return_value.role, expected_value[i].role)
             self.assertEqual(return_value.content, expected_value[i].content)
+
+
+@ddt.ddt
+class CheckIfAuditTrialIsExpiredTests(TestCase):
+    """
+    Test suite for audit_trial_is_expired.
+    """
+
+    def setUp(self):
+        super().setUp()
+        self.course_key = CourseKey.from_string('course-v1:edx+fake+1')
+        self.user = User(username='tester', email='[email protected]')
+        self.user.save()
+
+        self.role = 'verified'
+        self.upgrade_deadline = datetime.now() + timedelta(days=1)  # 1 day from now
+
+    def test_check_if_past_upgrade_deadline(self):
+        upgrade_deadline = datetime.now() - timedelta(days=1)  # yesterday
+        self.assertEqual(audit_trial_is_expired(self.user, upgrade_deadline), True)
+
+    def test_audit_trial_is_expired_audit_trial_expired(self):
+        LearningAssistantAuditTrial.objects.create(
+            user=self.user,
+            start_date=datetime.now() - timedelta(days=AUDIT_TRIAL_MAX_DAYS + 1),  # 1 day more than trial deadline
+        )
+        self.assertEqual(audit_trial_is_expired(self.user, self.upgrade_deadline), True)
+
+    def test_audit_trial_is_expired_audit_trial_unexpired(self):
+        LearningAssistantAuditTrial.objects.create(
+            user=self.user,
+            start_date=datetime.now() - timedelta(days=AUDIT_TRIAL_MAX_DAYS - 0.99),  # 0.99 days less than deadline
+        )
+        self.assertEqual(audit_trial_is_expired(self.user, self.upgrade_deadline), False)