diff --git a/src/xapi/tests/tests.py b/src/xapi/tests/tests.py index d57b107761124dd4de5dc351d75148b09647b857..674d4ef7351bfb0085b2821256a13aa8f6addfdf 100644 --- a/src/xapi/tests/tests.py +++ b/src/xapi/tests/tests.py @@ -1,9 +1,10 @@ import json import os +import copy from datetime import datetime from io import StringIO from unittest import mock -from unittest.mock import patch +from unittest.mock import patch, MagicMock from django.core.exceptions import ObjectDoesNotExist from django.test import TestCase @@ -892,3 +893,148 @@ class TestxAPIObjectMatchingDefinitiondId(BaseTestCase): self.assertEqual( response.json()["message"], "xAPI statements couldn't be stored in LRS" ) + +class TextxAPIAdditionalLrs(BaseTestCase): + provider_schema = { + "id": "h5p-0", + "name": "H5P", + "description": "Open-source content collaboration framework", + "groups": [ + { + "id": "default_group", + "label": "Default group", + "description": "default", + "showVerbDetails": True, + "purposeOfCollection": "Lorem Ipsum", + "verbs": [ + { + "id": "https://xapi.elearn.rwth-aachen.de/definitions/lms/verbs/unlocked", + "label": "Unlocked", + "description": "Actor unlocked an object", + "defaultConsent": True, + "objects": [ + { + "id": "https://xapi.elearn.rwth-aachen.de/definitions/lms/activities/my-random-object-id", + "label": "Course", + "defaultConsent": True, + "matching": "id", + "definition": { + "type": "https://xapi.elearn.rwth-aachen.de/definitions/lms/activities/course", + "name": { + "enUS": "A course within an LMS. Contains learning materials and activities" + }, + }, + }, + ], + } + ], + "isDefault": True, + }, + ], + "essentialVerbs": [], + "additionalLrs": [ + { + "url": "http://localhost:5555/xapi/statements", + "token": "token_to_check" + } + ] + } + + statement = { + "actor": {"mbox": "mailto:test@mail.com"}, + "verb": {"id": "https://xapi.elearn.rwth-aachen.de/definitions/lms/verbs/unlocked"}, + "object": { + "objectType": "Activity", + "definition": { + "type": "https://xapi.elearn.rwth-aachen.de/definitions/lms/activities/course", + "name": {"de": "Testkurs KI:edu.nrw "}, + }, + "id": "https://xapi.elearn.rwth-aachen.de/definitions/lms/activities/my-random-object-id", + }, + "timestamp": datetime.now().strftime("%Y-%m-%dT%H:%M:%SZ"), + } + + additional_lrs_auth_headers = {'Authorization': 'Bearer token_to_check'} + + @patch("xapi.views.store_in_db", mock_store_in_lrs) + @patch("xapi.views.requests.post") + def test_xapi_additional_lrs(self, mock_post): + """ + Ensure xAPI statement is matched by object definition type. + """ + + # Create provider + with StringIO(json.dumps(self.provider_schema)) as fp: + response = self.provider_client.put( + "/api/v1/consents/provider/create", + {"provider-schema": fp}, + format="multipart", + ) + self.assertEqual(response.status_code, 201) + + # Create user consent for test user + user_consent = [ + { + "providerId": 1, + "providerSchemaId": 1, + "verbs": [ + { + "provider": 1, + "id": "https://xapi.elearn.rwth-aachen.de/definitions/lms/verbs/unlocked", + "consented": True, + "objects": json.dumps( + [ + { + "id": "https://xapi.elearn.rwth-aachen.de/definitions/lms/activities/my-random-object-id", + "label": "Course", + "defaultConsent": True, + "matching": "id", + "definition": { + "type": "https://xapi.elearn.rwth-aachen.de/definitions/lms/activities/course", + "name": { + "enUS": "A course within an LMS. Contains learning materials and activities" + }, + }, + "consented": True, + } + ] + ), + } + ], + } + ] + response = self.user_client.post( + "/api/v1/consents/user/save", user_consent, format="json" + ) + self.assertEqual(response.status_code, 200) + + access_token_h5p = ProviderAuthorization.objects.get(provider__name="H5P").key + + client = APIClient() + client.credentials(HTTP_AUTHORIZATION="Basic " + access_token_h5p) + + # side effect to immediately create a copy of the object to preserve original properties (since, apparently, the magic mock passes the json object by reference) + def capture_args(*args, **kwargs): + self.captured_json = copy.deepcopy(kwargs.get('json')) + self.captured_headers = copy.deepcopy(kwargs.get('headers')) + + # create mock response for external LRS + mock_response = MagicMock() + mock_response.status_code = 200 + mock_response.json.return_value = {'message': "xAPI statements successfully stored in LRS"} + mock_post.return_value = mock_response + mock_post.side_effect = capture_args + + # Send xAPI statement + response = client.post( + "/xapi/statements", + self.statement, + format="json", + ) + self.assertEqual(response.status_code, 200) + self.assertEqual( + response.json()["message"], "xAPI statements successfully stored in LRS" + ) + mock_post.assert_called_once() + self.assertEqual(self.captured_json, self.statement) + self.assertEqual(self.captured_headers, self.additional_lrs_auth_headers)