Skip to content
Snippets Groups Projects
Commit a2ad7937 authored by Lennard Strohmeyer's avatar Lennard Strohmeyer :penguin:
Browse files

#110: implemented timestamp validation and test case 

#91: bugfix
parent cea40f4c
No related branches found
No related tags found
No related merge requests found
Hide whitespace changes
Inline Side-by-side
src/xapi/tests/tests.py
+ 155
− 14
View file @ a2ad7937
import json
import os
import copy
from datetime import datetime
from datetime import datetime, timedelta
from io import StringIO
from unittest import mock
from unittest.mock import patch, MagicMock
......@@ -134,7 +134,7 @@ class XAPITestCase(TestCase):
"objectType": "Activity",
"definition": accepted_verb["objects"][0]["definition"],
},
"timestamp": datetime.now().strftime("%Y-%m-%dT%H:%M:%SZ"),
"timestamp": (datetime.now() + timedelta(0,30)).strftime("%Y-%m-%dT%H:%M:%SZ"), # timedelta is used to ensure the statement is "after" the consent
},
format="json",
)
......@@ -208,7 +208,7 @@ class XAPITestCase(TestCase):
}
},
},
"timestamp": datetime.now().strftime("%Y-%m-%dT%H:%M:%SZ"),
"timestamp": (datetime.now() + timedelta(0,30)).strftime("%Y-%m-%dT%H:%M:%SZ"), # timedelta is used to ensure the statement is "after" the consent
},
format="json",
)
......@@ -291,7 +291,7 @@ class TestxAPIWithDataRecordingPause(BaseTestCase):
"name": {"enUS": "2.3.1 Funktion Zirkulationsleitung"},
},
},
"timestamp": datetime.now().strftime("%Y-%m-%dT%H:%M:%SZ"),
"timestamp": (datetime.now() + timedelta(0,30)).strftime("%Y-%m-%dT%H:%M:%SZ"), # timedelta is used to ensure the statement is "after" the consent
},
format="json",
)
......@@ -318,7 +318,7 @@ class TestxAPIWithDataRecordingPause(BaseTestCase):
"name": {"enUS": "2.3.1 Funktion Zirkulationsleitung"},
},
},
"timestamp": datetime.now().strftime("%Y-%m-%dT%H:%M:%SZ"),
"timestamp": (datetime.now() + timedelta(0,30)).strftime("%Y-%m-%dT%H:%M:%SZ"), # timedelta is used to ensure the statement is "after" the consent
},
format="json",
)
......@@ -415,7 +415,7 @@ class TestxAPIStatementActorAccount(BaseTestCase):
"name": {"enUS": "2.3.1 Funktion Zirkulationsleitung"},
},
},
"timestamp": datetime.now().strftime("%Y-%m-%dT%H:%M:%SZ"),
"timestamp": (datetime.now() + timedelta(0,30)).strftime("%Y-%m-%dT%H:%M:%SZ"), # timedelta is used to ensure the statement is "after" the consent
},
format="json",
)
......@@ -500,7 +500,7 @@ class TestxAPIStatementActorAccount(BaseTestCase):
"name": "2.3.1 Funktion Zirkulationsleitung",
},
},
"timestamp": datetime.now().strftime("%Y-%m-%dT%H:%M:%SZ"),
"timestamp": (datetime.now() + timedelta(0,30)).strftime("%Y-%m-%dT%H:%M:%SZ"), # timedelta is used to ensure the statement is "after" the consent
},
format="json",
)
......@@ -583,7 +583,7 @@ class TestxAPIStatementActorAccount(BaseTestCase):
"name": {"enUS": "2.3.1 Funktion Zirkulationsleitung"},
},
},
"timestamp": datetime.now().strftime("%Y-%m-%dT%H:%M:%SZ"),
"timestamp": (datetime.now() + timedelta(0,30)).strftime("%Y-%m-%dT%H:%M:%SZ"), # timedelta is used to ensure the statement is "after" the consent
},
{
"actor": {
......@@ -601,7 +601,7 @@ class TestxAPIStatementActorAccount(BaseTestCase):
"name": {"enUS": "1.2.3 Kappenventil"},
},
},
"timestamp": datetime.now().strftime("%Y-%m-%dT%H:%M:%SZ"),
"timestamp": (datetime.now() + timedelta(0,30)).strftime("%Y-%m-%dT%H:%M:%SZ"), # timedelta is used to ensure the statement is "after" the consent
},
],
format="json",
......@@ -721,7 +721,7 @@ class TestxAPIObjectMatchingDefinitionType(BaseTestCase):
},
"id": "https://moodle-analytics.ruhr-uni-bochum.de/course/view.php?id=127",
},
"timestamp": datetime.now().strftime("%Y-%m-%dT%H:%M:%SZ"),
"timestamp": (datetime.now() + timedelta(0,30)).strftime("%Y-%m-%dT%H:%M:%SZ"), # timedelta is used to ensure the statement is "after" the consent
},
format="json",
)
......@@ -744,7 +744,7 @@ class TestxAPIObjectMatchingDefinitionType(BaseTestCase):
},
"id": "https://moodle-analytics.ruhr-uni-bochum.de/course/view.php?id=127",
},
"timestamp": datetime.now().strftime("%Y-%m-%dT%H:%M:%SZ"),
"timestamp": (datetime.now() + timedelta(0,30)).strftime("%Y-%m-%dT%H:%M:%SZ"), # timedelta is used to ensure the statement is "after" the consent
},
format="json",
)
......@@ -863,7 +863,7 @@ class TestxAPIObjectMatchingDefinitiondId(BaseTestCase):
},
"id": "https://xapi.elearn.rwth-aachen.de/definitions/lms/activities/my-random-object-id",
},
"timestamp": datetime.now().strftime("%Y-%m-%dT%H:%M:%SZ"),
"timestamp": (datetime.now() + timedelta(0,30)).strftime("%Y-%m-%dT%H:%M:%SZ"), # timedelta is used to ensure the statement is "after" the consent
},
format="json",
)
......@@ -887,7 +887,7 @@ class TestxAPIObjectMatchingDefinitiondId(BaseTestCase):
},
"id": "https://xapi.elearn.rwth-aachen.de/definitions/lms/activities/invalid-id",
},
"timestamp": datetime.now().strftime("%Y-%m-%dT%H:%M:%SZ"),
"timestamp": (datetime.now() + timedelta(0,30)).strftime("%Y-%m-%dT%H:%M:%SZ"), # timedelta is used to ensure the statement is "after" the consent
},
format="json",
)
......@@ -897,6 +897,147 @@ class TestxAPIObjectMatchingDefinitiondId(BaseTestCase):
)
class TestxAPITimestampAfterConsent(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": [],
}
@patch("xapi.views.store_in_db", mock_store_in_lrs)
def test_xapi_with_timestamp_validation(self):
"""
Ensure xAPI statement "timestamp" field is a point in time after the user has given consent.
"""
# 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)
# Send xAPI statement with timestamp after consent
response = client.post(
"/xapi/statements",
{
"actor": {"mbox": f"mailto:{self.test_user_email}"},
"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() + timedelta(0,30)).strftime("%Y-%m-%dT%H:%M:%SZ"), # timedelta is used to ensure the statement is "after" the consent
},
format="json",
)
self.assertEqual(response.status_code, 200)
self.assertEqual(
response.json()["message"], "xAPI statements successfully stored in LRS"
)
# Send xAPI statement with timestamp before consent
response = client.post(
"/xapi/statements",
{
"actor": {"mbox": f"mailto:{self.test_user_email}"},
"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/invalid-id",
},
"timestamp": (datetime.now() - timedelta(1,0)).strftime("%Y-%m-%dT%H:%M:%SZ"), # timedelta is used to ensure the statement is "before" the consent
},
format="json",
)
self.assertEqual(response.status_code, 400)
self.assertEqual(
response.json()["message"], "xAPI statements couldn't be stored in LRS"
)
class TextxAPIAdditionalLrs(BaseTestCase):
provider_schema = {
"id": "h5p-0",
......@@ -955,7 +1096,7 @@ class TextxAPIAdditionalLrs(BaseTestCase):
},
"id": "https://xapi.elearn.rwth-aachen.de/definitions/lms/activities/my-random-object-id",
},
"timestamp": datetime.now().strftime("%Y-%m-%dT%H:%M:%SZ"),
"timestamp": (datetime.now() + timedelta(0,30)).strftime("%Y-%m-%dT%H:%M:%SZ"), # timedelta is used to ensure the statement is "after" the consent
}
additional_lrs_auth_headers = {'Authorization': 'Bearer token_to_check'}
......
src/xapi/views.py
+ 9
− 2
View file @ a2ad7937
......@@ -3,6 +3,7 @@ import json
import os
from venv import logger
import datetime
from dateutil import parser
import requests
from django.conf import settings
......@@ -36,7 +37,7 @@ with open(os.path.join(PROJECT_PATH, "static/xapi_statement.schema.json")) as f:
def store_in_db(x_api_statement):
collection = lrs_db["statements"]
try:
x_api_statement.set("stored", datetime.datetime.now().isoformat()) # append "stored"-field - see https://github.com/adlnet/xAPI-Spec/blob/master/xAPI-Data.md#248-stored
x_api_statement["stored"] = datetime.datetime.now().isoformat() # append "stored"-field - see https://github.com/adlnet/xAPI-Spec/blob/master/xAPI-Data.md#248-stored
result = collection.insert_one(x_api_statement)
return str(result.inserted_id)
except Exception as e:
......@@ -185,10 +186,16 @@ def process_statement(x_api_statement, provider, latest_schema):
"reason": "User has paused data collection",
}
# validate timestamp, if given
if "timestamp" in x_api_statement.keys():
timestamp = parser.parse(x_api_statement["timestamp"])
else:
timestamp = datetime.datetime.now() # if the statement has no timestamp, use the current date s.t. validation does not fail as long as consent exists
# has the user given consent to this verb?
# maybe TODO: load correct provider schema pertaining to this user consent to validate the verb and objects fully
user_consent = UserConsents.objects.filter(
user=user, provider=provider, verb=verb, consented=True
user=user, provider=provider, verb=verb, consented=True, created__lte=timestamp
).first()
if not user_consent:
......
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Please register or to comment