diff --git a/pylti1p3/assignments_grades.py b/pylti1p3/assignments_grades.py index d0db759..b9da8f2 100644 --- a/pylti1p3/assignments_grades.py +++ b/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", diff --git a/pylti1p3/service_connector.py b/pylti1p3/service_connector.py index d50149f..695b7ce 100644 --- a/pylti1p3/service_connector.py +++ b/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) diff --git a/tests/test_grades.py b/tests/test_grades.py index e83a0d4..f7ad83e 100644 --- a/tests/test_grades.py +++ b/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)