Add Update and Delete capabilities for LineItems

pylti1p3/assignments_grades.py

@@ -89,7 +89,7 @@ def put_grade(
         return self._service_connector.make_service_request(
             self._service_data["scope"],
             score_url,
-            is_post=True,
+            method='POST',
             data=grade.get_value(),
             content_type="application/vnd.ims.lis.v1.score+json",
         )
@@ -115,6 +115,46 @@ def get_lineitem(self, lineitem_url: t.Optional[str] = None):
         )
         return LineItem(t.cast(TLineItem, lineitem_response["body"]))
 
+    def update_lineitem(self, lineitem: LineItem):
+        """
+        Update an individual lineitem. Lineitem to be updated is identified by the lineitem ID.
+
+        :param lineitem: LineItem instance to be updated
+        :return: LineItem instance (updated, based on response from the LTI platform)
+        """
+        if not self.can_create_lineitem():
+            raise LtiException("Can't update lineitem: Missing required scope")
+
+        lineitem_url = lineitem.get_id()
+
+        lineitem_response = self._service_connector.make_service_request(
+            self._service_data["scope"],
+            lineitem_url,
+            method='PUT',
+            data=lineitem.get_value(),
+            content_type="application/vnd.ims.lis.v2.lineitem+json",
+            accept="application/vnd.ims.lis.v2.lineitem+json",
+        )
+        return LineItem(t.cast(TLineItem, lineitem_response["body"]))
+
+    def delete_lineitem(self, lineitem_url: t.Optional[str]):
+        """
+        Delete an individual lineitem.
+
+        :param lineitem_url: endpoint for LTI line item
+        :return: None
+        """
+        if not self.can_create_lineitem():
+            raise LtiException("Can't update lineitem: Missing required scope")
+
+        self._service_connector.make_service_request(
+            self._service_data["scope"],
+            lineitem_url,
+            method='DELETE',
+            content_type="application/vnd.ims.lis.v2.lineitem+json",
+            accept="application/vnd.ims.lis.v2.lineitem+json",
+        )
+
     def get_lineitems_page(
         self, lineitems_url: t.Optional[str] = None
     ) -> t.Tuple[list, t.Optional[str]]:
@@ -252,7 +292,7 @@ def find_or_create_lineitem(
         created_lineitem = self._service_connector.make_service_request(
             self._service_data["scope"],
             self._service_data["lineitems"],
-            is_post=True,
+            method='POST',
             data=new_lineitem.get_value(),
             content_type="application/vnd.ims.lis.v2.lineitem+json",
             accept="application/vnd.ims.lis.v2.lineitem+json",

pylti1p3/service_connector.py

@@ -7,7 +7,7 @@
 import jwt  # type: ignore
 import requests
 import typing_extensions as te
-from .exception import LtiServiceException
+from .exception import LtiException, LtiServiceException
 from .registration import Registration
 
 TServiceConnectorResponse = te.TypedDict(
@@ -108,7 +108,7 @@ def make_service_request(
         self,
         scopes: t.Sequence[str],
         url: str,
-        is_post: bool = False,
+        method: str = 'GET',
         data: t.Optional[str] = None,
         content_type: str = "application/json",
         accept: str = "application/json",
@@ -117,12 +117,20 @@ def make_service_request(
         access_token = self.get_access_token(scopes)
         headers = {"Authorization": "Bearer " + access_token, "Accept": accept}
 
-        if is_post:
-            headers["Content-Type"] = content_type
-            post_data = data or None
-            r = self._requests_session.post(url, data=post_data, headers=headers)
-        else:
+        if method == 'GET':
             r = self._requests_session.get(url, headers=headers)
+        elif method == 'DELETE':
+            r = self._requests_session.delete(url, headers=headers)
+        else:
+            headers["Content-Type"] = content_type
+            request_data = data or None
+            if method == 'PUT':
+                r = self._requests_session.put(url, data=request_data, headers=headers)
+            elif method == 'POST':
+                r = self._requests_session.post(url, data=request_data, headers=headers)
+            else:
+                raise LtiException(f'Unsupported method: {method}. Available methods are: '
+                                   '"GET", "PUT", "POST", "DELETE".')
 
         if not r.ok:
             raise LtiServiceException(r)

tests/test_grades.py

@@ -4,7 +4,7 @@
 import requests_mock
 from parameterized import parameterized
 from pylti1p3.grade import Grade
-from pylti1p3.lineitem import LineItem
+from pylti1p3.lineitem import LineItem, TLineItem
 from .request import FakeRequest
 from .tool_config import get_test_tool_conf
 from .base import TestServicesBase
@@ -163,3 +163,110 @@ def test_send_scores(self):
 
                     resp = ags.put_grade(sc, sc_line_item)
                     self.assertEqual(expected_result, resp["body"])
+
+    def test_delete_lineitem(self):
+        from pylti1p3.contrib.django import DjangoMessageLaunch
+
+        tool_conf = get_test_tool_conf()
+
+        with patch.object(
+            DjangoMessageLaunch, "_get_jwt_body", autospec=True
+        ) as get_jwt_body:
+            message_launch = DjangoMessageLaunch(FakeRequest(), tool_conf)
+            line_items_url = "http://canvas.docker/api/lti/courses/1/line_items"
+            get_jwt_body.side_effect = lambda x: self._get_jwt_body()
+            with patch("socket.gethostbyname", return_value="127.0.0.1"):
+                with requests_mock.Mocker() as m:
+                    m.post(
+                        self._get_auth_token_url(),
+                        text=json.dumps(self._get_auth_token_response()),
+                    )
+
+                    line_item_url = "http://canvas.docker/api/lti/courses/1/line_items/1"
+                    line_items_response = [
+                        {
+                            "scoreMaximum": 100.0,
+                            "tag": "test",
+                            "id": line_item_url,
+                            "label": "Test",
+                        },
+                    ]
+                    m.get(line_items_url, text=json.dumps(line_items_response))
+                    m.delete(line_item_url, text='', status_code=204)
+
+                    ags = message_launch.validate_registration().get_ags()
+
+                    test_line_item = LineItem()
+                    test_line_item.set_tag("test").set_score_maximum(100).set_label(
+                        "Test"
+                    )
+                    line_item = ags.find_or_create_lineitem(test_line_item)
+                    self.assertIsNotNone(line_item)
+
+                    ags.delete_lineitem(line_item.get_id())
+
+                    # assert DELETE was called for specific URL
+                    self.assertEqual(len(m.request_history), 3)  # Auth, GET Line items, DELETE Line item
+                    self.assertEqual(m.request_history[2].method, 'DELETE')
+                    self.assertEqual(m.request_history[2].url, line_item_url)
+
+    def test_update_lineitem(self):
+        from pylti1p3.contrib.django import DjangoMessageLaunch
+
+        tool_conf = get_test_tool_conf()
+
+        with patch.object(
+            DjangoMessageLaunch, "_get_jwt_body", autospec=True
+        ) as get_jwt_body:
+            message_launch = DjangoMessageLaunch(FakeRequest(), tool_conf)
+            line_items_url = "http://canvas.docker/api/lti/courses/1/line_items"
+            get_jwt_body.side_effect = lambda x: self._get_jwt_body()
+            with patch("socket.gethostbyname", return_value="127.0.0.1"):
+                with requests_mock.Mocker() as m:
+                    m.post(
+                        self._get_auth_token_url(),
+                        text=json.dumps(self._get_auth_token_response()),
+                    )
+
+                    line_item_url = "http://canvas.docker/api/lti/courses/1/line_items/1"
+                    line_items_response = [
+                        {
+                            "id": line_item_url,
+                            "scoreMaximum": 100.0,
+                            "tag": "test",
+                            "label": "Test",
+                        },
+                    ]
+                    line_items_update_response = {
+                        "id": line_item_url,
+                        "scoreMaximum": 60.0,  # we changed the maximum score
+                        "tag": "test",
+                        "label": "Test",
+                    }
+
+                    m.get(line_items_url, text=json.dumps(line_items_response))
+                    m.put(line_item_url, text=json.dumps(line_items_update_response))
+
+                    ags = message_launch.validate_registration().get_ags()
+
+                    test_line_item = LineItem()
+                    test_line_item.set_tag("test").set_score_maximum(100).set_label(
+                        "Test"
+                    )
+
+                    line_item = ags.find_or_create_lineitem(test_line_item)
+                    self.assertIsNotNone(line_item)
+
+                    line_item.set_score_maximum(60)
+
+                    new_lineitem = ags.update_lineitem(line_item)
+
+                    self.assertEqual(new_lineitem.get_id(), line_item_url)
+                    self.assertEqual(new_lineitem.get_score_maximum(), 60.0)
+                    self.assertEqual(new_lineitem.get_tag(), "test")
+                    self.assertEqual(new_lineitem.get_label(), "Test")
+
+                    # assert PUT was called for specific URL
+                    self.assertEqual(len(m.request_history), 3)  # Auth, GET Line items, DELETE Line item
+                    self.assertEqual(m.request_history[2].method, 'PUT')
+                    self.assertEqual(m.request_history[2].url, line_item_url)