#106: test case

src/xapi/tests/tests.py

 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)