New dahboard endpoint to sync grades of a given async

lms/routes.py

@@ -284,6 +284,10 @@ def includeme(config):  # noqa: PLR0915
     config.add_route(
         "api.dashboard.assignment", r"/api/dashboard/assignments/{assignment_id}"
     )
+    config.add_route(
+        "api.dashboard.assignments.grading.sync",
+        "/api/dashboard/assignments/{assignment_id}/grading/sync",
+    )
     config.add_route("api.dashboard.assignments", "/api/dashboard/assignments")
     config.add_route(
         "api.dashboard.course.assignments.metrics",

lms/services/dashboard.py

@@ -1,8 +1,8 @@
 from pyramid.httpexceptions import HTTPNotFound, HTTPUnauthorized
 from sqlalchemy import select
 
+from lms.models import Assignment, Organization
 from lms.models.dashboard_admin import DashboardAdmin
-from lms.models.organization import Organization
 from lms.security import Permissions
 from lms.services.organization import OrganizationService
 
@@ -21,7 +21,7 @@ def __init__(
         self._course_service = course_service
         self._organization_service = organization_service
 
-    def get_request_assignment(self, request):
+    def get_request_assignment(self, request) -> Assignment:
         """Get and authorize an assignment for the given request."""
         assigment_id = request.matchdict.get(
             "assignment_id"

lms/services/lti_grading/_v13.py

@@ -92,15 +92,38 @@ def read_result(self, grading_id) -> GradingResult:
     def get_score_maximum(self, resource_link_id) -> float | None:
         return self._read_grading_configuration(resource_link_id).get("scoreMaximum")
 
+    def sync_grade(  # noqa: PLR0913
+        self,
+        lti_registration: LTIRegistration,
+        lis_outcome_service_url: str,
+        grade_timestamp: datetime,
+        user_grading_id: str,
+        score: float,
+    ):
+        """
+        Send a grade to the LMS.
+
+        This is very similar to `record_result` but not scoped to the request context,
+        taking all the necessary information as parameters.
+        """
+        payload = self._record_score_payload(
+            score, user_grading_id, grade_timestamp.isoformat()
+        )
+        return self._ltia_service.request(
+            lti_registration,
+            "POST",
+            self._service_url(lis_outcome_service_url, "/scores"),
+            scopes=self.LTIA_SCOPES,
+            json=payload,
+            headers={"Content-Type": "application/vnd.ims.lis.v1.score+json"},
+        )
+
     def record_result(self, grading_id, score=None, pre_record_hook=None, comment=None):
-        payload = {
-            "scoreMaximum": 1,
-            "scoreGiven": score,
-            "userId": grading_id,
-            "timestamp": datetime.now(timezone.utc).isoformat(),
-            "activityProgress": "Completed",
-            "gradingProgress": "FullyGraded",
-        }
+        payload = self._record_score_payload(
+            score,
+            grading_id,
+            datetime.now(timezone.utc).isoformat(),
+        )
         if comment:
             payload["comment"] = self._misc_plugin.format_grading_comment_for_lms(
                 comment
@@ -192,6 +215,19 @@ def _read_grading_configuration(self, resource_link_id) -> dict:
 
         return {}
 
+    @staticmethod
+    def _record_score_payload(
+        score: float | None, user_grading_id: str, timestamp: str
+    ):
+        return {
+            "scoreMaximum": 1,
+            "scoreGiven": score,
+            "userId": user_grading_id,
+            "timestamp": timestamp,
+            "activityProgress": "Completed",
+            "gradingProgress": "FullyGraded",
+        }
+
     @staticmethod
     def _service_url(base_url, endpoint):
         """

lms/views/dashboard/api/grading.py

@@ -0,0 +1,86 @@
+import logging
+from datetime import datetime
+
+from marshmallow import Schema, fields
+from pyramid.view import view_config
+from sqlalchemy import select
+
+from lms.models import (
+    ApplicationInstance,
+    Grouping,
+    LMSUser,
+    LTIRegistration,
+)
+from lms.security import Permissions
+from lms.services import LTIAHTTPService
+from lms.services.dashboard import DashboardService
+from lms.services.lti_grading.factory import LTI13GradingService
+from lms.validation._base import JSONPyramidRequestSchema
+
+LOG = logging.getLogger(__name__)
+
+
+class _GradeSchema(Schema):
+    h_userid = fields.Str(required=True)
+    grade = fields.Float(required=True)
+
+
+class AutoGradeSyncSchema(JSONPyramidRequestSchema):
+    grades = fields.List(fields.Nested(_GradeSchema), required=True)
+
+
+class DashboardGradingViews:
+    def __init__(self, request) -> None:
+        self.request = request
+        self.db = request.db
+        self.dashboard_service: DashboardService = request.find_service(
+            name="dashboard"
+        )
+
+    @view_config(
+        route_name="api.dashboard.assignments.grading.sync",
+        request_method="POST",
+        renderer="json",
+        permission=Permissions.GRADE_ASSIGNMENT,
+        schema=AutoGradeSyncSchema,
+    )
+    def auto_grading_sync(self):
+        assignment = self.dashboard_service.get_request_assignment(self.request)
+        assert assignment.lis_outcome_service_url, "Assignment without grading URL"
+        lti_registration = self.db.scalars(
+            select(LTIRegistration)
+            .join(ApplicationInstance)
+            .join(Grouping)
+            .where(Grouping.id == assignment.course_id)
+            .order_by(LTIRegistration.updated.desc())
+        ).first()
+        assert lti_registration, "No LTI registraion for LTI1.3 assignment"
+
+        sync_h_user_ids = [g["h_userid"] for g in self.request.parsed_params["grades"]]
+
+        sync_lms_users = self.db.execute(
+            select(LMSUser.h_userid, LMSUser.lti_v13_user_id).where(
+                LMSUser.h_userid.in_(sync_h_user_ids)
+            )
+        ).all()
+        # Organize the data in a dict h_userid -> lti_user_id for easier access
+        sync_lms_users_by_h_userid = {r[0]: r[1] for r in sync_lms_users}
+
+        grading_service = LTI13GradingService(
+            ltia_service=self.request.find_service(LTIAHTTPService),
+            line_item_url=None,
+            line_item_container_url=None,
+            product_family=None,  # type: ignore
+            misc_plugin=None,  # type: ignore
+            lti_registration=None,  # type: ignore
+        )
+        # Use the same timestamp for all grades of the same sync
+        grade_sync_time_stamp = datetime.now()
+        for grade in self.request.parsed_params["grades"]:
+            grading_service.sync_grade(
+                lti_registration,
+                assignment.lis_outcome_service_url,
+                grade_sync_time_stamp,
+                sync_lms_users_by_h_userid[grade["h_userid"]],
+                grade["grade"],
+            )

tests/unit/lms/services/lti_grading/_v13_test.py

@@ -1,3 +1,4 @@
+from datetime import datetime
 from unittest.mock import Mock, sentinel
 
 import pytest
@@ -148,6 +149,33 @@ def test_get_score_maximum_no_line_item(self, svc, ltia_http_service):
 
         assert not svc.get_score_maximum(sentinel.resource_link_id)
 
+    def test_sync_grade(self, svc, ltia_http_service, lti_registration):
+        response = svc.sync_grade(
+            lti_registration,
+            "LIS_OUTCOME_SERVICE_URL",
+            datetime(2022, 4, 4),
+            sentinel.user_id,
+            sentinel.grade,
+        )
+
+        payload = {
+            "scoreMaximum": 1,
+            "scoreGiven": sentinel.grade,
+            "userId": sentinel.user_id,
+            "timestamp": "2022-04-04T00:00:00",
+            "activityProgress": "Completed",
+            "gradingProgress": "FullyGraded",
+        }
+        ltia_http_service.request.assert_called_once_with(
+            lti_registration,
+            "POST",
+            "LIS_OUTCOME_SERVICE_URL/scores",
+            scopes=svc.LTIA_SCOPES,
+            json=payload,
+            headers={"Content-Type": "application/vnd.ims.lis.v1.score+json"},
+        )
+        assert response == ltia_http_service.request.return_value
+
     @freeze_time("2022-04-04")
     @pytest.mark.parametrize("comment", [sentinel.comment, None])
     def test_record_result(

tests/unit/lms/views/dashboard/api/grading_test.py

@@ -0,0 +1,72 @@
+from datetime import datetime
+
+import pytest
+from freezegun import freeze_time
+
+from lms.views.dashboard.api.grading import DashboardGradingViews
+from tests import factories
+
+
+@pytest.mark.usefixtures("dashboard_service", "ltia_http_service")
+class TestDashboardGradingViews:
+    @freeze_time("2022-06-21 12:00:00")
+    def test_auto_grading_sync(
+        self,
+        lti_v13_application_instance,
+        pyramid_request,
+        views,
+        DashboardGradingView,
+        ltia_http_service,
+        dashboard_service,
+        assignment,
+    ):
+        dashboard_service.get_request_assignment.return_value = assignment
+
+        views.auto_grading_sync()
+
+        dashboard_service.get_request_assignment.assert_called_once_with(
+            pyramid_request
+        )
+        DashboardGradingView.assert_called_once_with(
+            ltia_service=ltia_http_service,
+            line_item_url=None,
+            line_item_container_url=None,
+            product_family=None,
+            misc_plugin=None,
+            lti_registration=None,
+        )
+        DashboardGradingView.return_value.sync_grade.assert_called_once_with(
+            lti_v13_application_instance.lti_registration,
+            "LIS_OUTCOME_SERVICE_URL",
+            datetime(2022, 6, 21, 12, 0, 0),
+            "LTI_V13_USER_ID",
+            1,
+        )
+
+    @pytest.fixture
+    def assignment(self, lti_v13_application_instance, db_session):
+        course = factories.Course(application_instance=lti_v13_application_instance)
+        assignment = factories.Assignment(
+            lis_outcome_service_url="LIS_OUTCOME_SERVICE_URL", course=course
+        )
+        db_session.flush()
+        return assignment
+
+    @pytest.fixture
+    def lms_user(self):
+        return factories.LMSUser(lti_v13_user_id="LTI_V13_USER_ID")
+
+    @pytest.fixture
+    def pyramid_request(self, pyramid_request, lms_user):
+        pyramid_request.parsed_params = {
+            "grades": [{"h_userid": lms_user.h_userid, "grade": 1}]
+        }
+        return pyramid_request
+
+    @pytest.fixture
+    def views(self, pyramid_request):
+        return DashboardGradingViews(pyramid_request)
+
+    @pytest.fixture
+    def DashboardGradingView(self, patch):
+        return patch("lms.views.dashboard.api.grading.LTI13GradingService")