[BD-24] [TNL-7661] [BB-3172] LTI Improvements - Use declarative grading model on XBlock launch

lti_consumer/admin.py

@@ -2,7 +2,7 @@
 Admin views for LTI related models.
 """
 from django.contrib import admin
-from lti_consumer.models import LtiAgsLineItem, LtiConfiguration
+from lti_consumer.models import LtiAgsLineItem, LtiConfiguration, LtiAgsScore
 
 
 class LtiConfigurationAdmin(admin.ModelAdmin):
@@ -16,3 +16,4 @@ class LtiConfigurationAdmin(admin.ModelAdmin):
 
 admin.site.register(LtiConfiguration, LtiConfigurationAdmin)
 admin.site.register(LtiAgsLineItem)
+admin.site.register(LtiAgsScore)

lti_consumer/apps.py

@@ -25,3 +25,7 @@ class LTIConsumerApp(AppConfig):
             }
         }
     }
+
+    def ready(self):
+        # pylint: disable=unused-import,import-outside-toplevel
+        import lti_consumer.signals

lti_consumer/lti_1p3/ags.py

@@ -19,6 +19,7 @@ class LtiAgs:
     def __init__(
         self,
         lineitems_url,
+        lineitem_url=None,
         allow_creating_lineitems=True,
         results_service_enabled=True,
         scores_service_enabled=True
@@ -37,6 +38,8 @@ def __init__(
         # Lineitems urls
         self.lineitems_url = lineitems_url
 
+        self.lineitem_url = lineitem_url
+
     def get_available_scopes(self):
         """
         Retrieves list of available token scopes in this instance.
@@ -60,11 +63,15 @@ def get_lti_ags_launch_claim(self):
         """
         Returns LTI AGS Claim to be injected in the LTI launch message.
         """
-        ags_claim = {
-            "https://purl.imsglobal.org/spec/lti-ags/claim/endpoint": {
-                "scope": self.get_available_scopes(),
-                "lineitems": self.lineitems_url,
-            }
+
+        claim_values = {
+            "scope": self.get_available_scopes(),
+            "lineitems": self.lineitems_url,
         }
 
-        return ags_claim
+        if self.lineitem_url:
+            claim_values["lineitem"] = self.lineitem_url
+
+        return {
+            "https://purl.imsglobal.org/spec/lti-ags/claim/endpoint": claim_values,
+        }

lti_consumer/lti_1p3/consumer.py

@@ -475,16 +475,20 @@ def lti_ags(self):
     def enable_ags(
         self,
         lineitems_url,
+        lineitem_url=None,
+        allow_programatic_grade_interaction=False
     ):
         """
         Enable LTI Advantage Assignments and Grades Service.
 
         This will include the LTI AGS Claim in the LTI message
         and set up the required class.
         """
+
         self.ags = LtiAgs(
             lineitems_url=lineitems_url,
-            allow_creating_lineitems=True,
+            lineitem_url=lineitem_url,
+            allow_creating_lineitems=allow_programatic_grade_interaction,
             results_service_enabled=True,
             scores_service_enabled=True
         )

lti_consumer/lti_1p3/tests/test_consumer.py

@@ -647,11 +647,11 @@ def test_enable_ags(self):
             {
                 'https://purl.imsglobal.org/spec/lti-ags/claim/endpoint': {
                     'scope': [
-                        'https://purl.imsglobal.org/spec/lti-ags/scope/lineitem',
+                        'https://purl.imsglobal.org/spec/lti-ags/scope/lineitem.readonly',
                         'https://purl.imsglobal.org/spec/lti-ags/scope/result.readonly',
-                        'https://purl.imsglobal.org/spec/lti-ags/scope/score'
+                        'https://purl.imsglobal.org/spec/lti-ags/scope/score',
                     ],
-                    'lineitems': 'http://example.com/lineitems'
+                    'lineitems': 'http://example.com/lineitems',
                 }
             }
         )

lti_consumer/models.py

@@ -12,7 +12,10 @@
 # LTI 1.3
 from lti_consumer.lti_1p3.consumer import LtiAdvantageConsumer
 from lti_consumer.plugin import compat
-from lti_consumer.utils import get_lms_base, get_lti_ags_lineitems_url
+from lti_consumer.utils import (
+    get_lms_base,
+    get_lti_ags_lineitems_url,
+)
 
 
 class LtiConfiguration(models.Model):
@@ -129,8 +132,29 @@ def _get_lti_1p3_consumer(self):
 
             # Check if enabled and setup LTI-AGS
             if self.block.has_score:
+
+                default_values = {
+                    'resource_id': self.block.location,
+                    'score_maximum': self.block.weight,
+                    'label': self.block.display_name
+                }
+
+                if hasattr(self.block, 'start'):
+                    default_values['start_date_time'] = self.block.start
+
+                if hasattr(self.block, 'due'):
+                    default_values['end_date_time'] = self.block.due
+
+                # create LineItem if there is none for current lti configuration
+                lineitem, _ = LtiAgsLineItem.objects.get_or_create(
+                    lti_configuration=self,
+                    resource_link_id=self.block.location,
+                    defaults=default_values
+                )
+
                 consumer.enable_ags(
-                    lineitems_url=get_lti_ags_lineitems_url(self.id)
+                    lineitems_url=get_lti_ags_lineitems_url(self.id),
+                    lineitem_url=get_lti_ags_lineitems_url(self.id, lineitem.id)
                 )
 
             return consumer

lti_consumer/plugin/compat.py

@@ -1,6 +1,8 @@
 """
 Compatibility layer to isolate core-platform method calls from implementation.
 """
+from django.core.exceptions import ValidationError
+from lti_consumer.exceptions import LtiError
 
 
 def run_xblock_handler(*args, **kwargs):
@@ -30,22 +32,65 @@ def load_block_as_anonymous_user(location):
     handler.
     """
     # pylint: disable=import-error,import-outside-toplevel
+    from crum import impersonate
     from django.contrib.auth.models import AnonymousUser
     from xmodule.modulestore.django import modulestore
     from lms.djangoapps.courseware.module_render import get_module_for_descriptor_internal
 
     # Retrieve descriptor from modulestore
     descriptor = modulestore().get_item(location)
 
-    # Load block, attaching it to AnonymousUser
-    get_module_for_descriptor_internal(
-        user=AnonymousUser(),
-        descriptor=descriptor,
-        student_data=None,
-        course_id=location.course_key,
-        track_function=None,
-        xqueue_callback_url_prefix="",
-        request_token="",
-    )
+    # ensure `crum.get_current_user` returns AnonymousUser. It returns None when outside
+    # of request scope which causes error during block loading.
+    user = AnonymousUser()
+    with impersonate(user):
+        # Load block, attaching it to AnonymousUser
+        get_module_for_descriptor_internal(
+            user=user,
+            descriptor=descriptor,
+            student_data=None,
+            course_id=location.course_key,
+            track_function=None,
+            xqueue_callback_url_prefix="",
+            request_token="",
+        )
+
+        return descriptor
+
+
+def get_user_from_external_user_id(external_user_id):
+    """
+    Import ExternalId model and find user by external_user_id
+    """
+    # pylint: disable=import-error,import-outside-toplevel
+    from openedx.core.djangoapps.external_user_ids.models import ExternalId
+    try:
+        external_id = ExternalId.objects.get(
+            external_user_id=external_user_id,
+            external_id_type__name='lti'
+        )
+        return external_id.user
+    except ExternalId.DoesNotExist as exception:
+        raise LtiError('Invalid User') from exception
+    except ValidationError as exception:
+        raise LtiError('Invalid userID') from exception
 
-    return descriptor
+
+def publish_grade(block, user, score, possible, only_if_higher=False, score_deleted=None, comment=None):
+    """
+    Import grades signals and publishes score by triggering SCORE_PUBLISHED signal.
+    """
+    # pylint: disable=import-error,import-outside-toplevel
+    from lms.djangoapps.grades.api import signals as grades_signals
+
+    # publish score
+    grades_signals.SCORE_PUBLISHED.send(
+        sender=None,
+        block=block,
+        user=user,
+        raw_earned=score,
+        raw_possible=possible,
+        only_if_higher=only_if_higher,
+        score_deleted=score_deleted,
+        grader_response=comment
+    )

lti_consumer/signals.py

@@ -0,0 +1,29 @@
+"""
+LTI Consumer related Signal handlers
+"""
+
+from django.db.models.signals import post_save
+from django.dispatch import receiver
+
+from lti_consumer.models import LtiAgsScore
+from lti_consumer.plugin import compat
+
+
+@receiver(post_save, sender=LtiAgsScore, dispatch_uid='publish_grade_on_score_update')
+def publish_grade_on_score_update(sender, instance, **kwargs):  # pylint: disable=unused-argument
+    """
+    Publish grade to xblock whenever score saved/updated and its grading_progress is set to FullyGraded.
+    """
+    if instance.grading_progress == LtiAgsScore.FULLY_GRADED:
+        block = compat.load_block_as_anonymous_user(instance.line_item.resource_link_id)
+        if not block.is_past_due() or block.accept_grades_past_due:
+            user = compat.get_user_from_external_user_id(instance.user_id)
+            # check if score_given is larger than score_maximum
+            score = instance.score_given if instance.score_given < instance.score_maximum else instance.score_maximum
+            compat.publish_grade(
+                block,
+                user,
+                score,
+                instance.score_maximum,
+                comment=instance.comment
+            )