Skip to content
Snippets Groups Projects
Unverified Commit 5fc16b38 authored by Patrick Cockwell's avatar Patrick Cockwell Committed by GitHub
Browse files

[BD-24] Implement LTI AGS Score Publish Service and Results (#108)

* BD-24 Implement LTI AGS Score Publish Service and Results Service

* Address PR comments and add more validation

* Address PR comments

* Add tests; Fix error with scoreMaximum; Fix quality issues; Adjust user_id results url slightly

* Add permissions tests and address other PR comments

* Fix quality test

* Address PR comments
parent 38a8a0d6
No related branches found
No related tags found
No related merge requests found
Showing with 1127 additions and 47 deletions
......@@ -19,3 +19,6 @@ var/
# virtualenvironment
venv/
# pyenv
.python-version
......@@ -47,6 +47,8 @@ LTI_1P3_ACCESS_TOKEN_SCOPES = [
# LTI-AGS Scopes
'https://purl.imsglobal.org/spec/lti-ags/scope/lineitem.readonly',
'https://purl.imsglobal.org/spec/lti-ags/scope/lineitem',
'https://purl.imsglobal.org/spec/lti-ags/scope/result.readonly',
'https://purl.imsglobal.org/spec/lti-ags/scope/score',
]
......
......@@ -14,3 +14,12 @@ class LineItemParser(parsers.JSONParser):
It's the same as JSON parser, but uses a custom media_type.
"""
media_type = 'application/vnd.ims.lis.v2.lineitem+json'
class LineItemScoreParser(parsers.JSONParser):
"""
Line Item Parser.
It's the same as JSON parser, but uses a custom media_type.
"""
media_type = 'application/vnd.ims.lis.v1.score+json'
......@@ -23,25 +23,32 @@ class LtiAgsPermissions(permissions.BasePermission):
"""
Check if LTI AGS permissions are set in auth token.
"""
has_perm = False
# Retrieves token from request, which was already checked by
# the Authentication class, so we assume it's a sane value.
auth_token = request.headers['Authorization'].split()[1]
scopes = []
if view.action in ['list', 'retrieve']:
# We don't need to wrap this around a try-catch because
# the token was already tested by the Authentication class.
has_perm = request.lti_consumer.check_token(
auth_token,
[
'https://purl.imsglobal.org/spec/lti-ags/scope/lineitem.readonly',
'https://purl.imsglobal.org/spec/lti-ags/scope/lineitem',
],
)
scopes = [
'https://purl.imsglobal.org/spec/lti-ags/scope/lineitem.readonly',
'https://purl.imsglobal.org/spec/lti-ags/scope/lineitem',
]
elif view.action in ['create', 'update', 'partial_update', 'delete']:
has_perm = request.lti_consumer.check_token(
auth_token,
['https://purl.imsglobal.org/spec/lti-ags/scope/lineitem']
)
return has_perm
scopes = [
'https://purl.imsglobal.org/spec/lti-ags/scope/lineitem',
]
elif view.action in ['results']:
scopes = [
'https://purl.imsglobal.org/spec/lti-ags/scope/result.readonly'
]
elif view.action in ['scores']:
scopes = [
'https://purl.imsglobal.org/spec/lti-ags/scope/score',
]
if scopes:
return request.lti_consumer.check_token(auth_token, scopes)
return False
......@@ -26,3 +26,25 @@ class LineItemRenderer(renderers.JSONRenderer):
"""
media_type = 'application/vnd.ims.lis.v2.lineitem+json'
format = 'json'
class LineItemScoreRenderer(renderers.JSONRenderer):
"""
Score Renderer.
It's a JSON renderer, but uses a custom media_type.
Reference: https://www.imsglobal.org/spec/lti-ags/v2p0#media-types-and-schemas
"""
media_type = 'application/vnd.ims.lis.v1.score+json'
format = 'json'
class LineItemResultsRenderer(renderers.JSONRenderer):
"""
Results Renderer.
It's a JSON renderer, but uses a custom media_type.
Reference: https://www.imsglobal.org/spec/lti-ags/v2p0#media-types-and-schemas
"""
media_type = 'application/vnd.ims.lis.v2.resultcontainer+json'
format = 'json'
"""
Serializers for LTI-related endpoints
"""
from rest_framework import serializers
from django.utils import timezone
from rest_framework import serializers, ISO_8601
from rest_framework.reverse import reverse
from opaque_keys import InvalidKeyError
from opaque_keys.edx.keys import UsageKey
from lti_consumer.models import LtiAgsLineItem
from lti_consumer.models import LtiAgsLineItem, LtiAgsScore
class UsageKeyField(serializers.Field):
......@@ -89,3 +90,140 @@ class LtiAgsLineItemSerializer(serializers.ModelSerializer):
'startDateTime',
'endDateTime',
)
class LtiAgsScoreSerializer(serializers.ModelSerializer):
"""
LTI AGS LineItemScore Serializer.
This maps out the internally stored LtiAgsScore to
the LTI-AGS API Specification, as shown in the example
response below:
{
"timestamp": "2017-04-16T18:54:36.736+00:00",
"scoreGiven" : 83,
"scoreMaximum" : 100,
"comment" : "This is exceptional work.",
"activityProgress" : "Completed",
"gradingProgress": "FullyGraded",
"userId" : "5323497"
}
Reference:
https://www.imsglobal.org/spec/lti-ags/v2p0#example-application-vnd-ims-lis-v1-score-json-representation
"""
# NOTE: `serializers.DateTimeField` always outputs the value in the local timezone of the server running the code
# This is because Django is time aware (see settings.USE_TZ) and because Django is unable to determine the timezone
# of the person making the API request, thus falling back on the local timezone. As such, since all outputs will
# necessarily be in a singular timezone, that timezone should be `utc`
timestamp = serializers.DateTimeField(input_formats=[ISO_8601], format=ISO_8601, default_timezone=timezone.utc)
scoreGiven = serializers.FloatField(source='score_given', required=False, allow_null=True, default=None)
scoreMaximum = serializers.FloatField(source='score_maximum', required=False, allow_null=True, default=None)
comment = serializers.CharField(required=False, allow_null=True)
activityProgress = serializers.CharField(source='activity_progress')
gradingProgress = serializers.CharField(source='grading_progress')
userId = serializers.CharField(source='user_id')
def validate_timestamp(self, value):
"""
Ensure that if an existing record is being updated, that the timestamp is in the after the existing one
"""
if self.instance:
if self.instance.timestamp > value:
raise serializers.ValidationError('Score timestamp can only be updated to a later point in time')
if self.instance.timestamp == value:
raise serializers.ValidationError('Score already exists for the provided timestamp')
return value
def validate_scoreMaximum(self, value):
"""
Ensure that scoreMaximum is set when scoreGiven is provided and not None
"""
if not value and self.initial_data.get('scoreGiven', None) is not None:
raise serializers.ValidationError('scoreMaximum is a required field when providing a scoreGiven value.')
return value
class Meta:
model = LtiAgsScore
fields = (
'timestamp',
'scoreGiven',
'scoreMaximum',
'comment',
'activityProgress',
'gradingProgress',
'userId',
)
class LtiAgsResultSerializer(serializers.ModelSerializer):
"""
LTI AGS LineItemResult Serializer.
This maps out the internally stored LtiAgsScpre to
the LTI-AGS API Specification, as shown in the example
response below:
{
"id": "https://lms.example.com/context/2923/lineitems/1/results/5323497",
"scoreOf": "https://lms.example.com/context/2923/lineitems/1",
"userId": "5323497",
"resultScore": 0.83,
"resultMaximum": 1,
"comment": "This is exceptional work."
}
Reference:
https://www.imsglobal.org/spec/lti-ags/v2p0#example-application-vnd-ims-lis-v1-score-json-representation
"""
id = serializers.SerializerMethodField()
scoreOf = serializers.SerializerMethodField()
userId = serializers.CharField(source='user_id')
resultScore = serializers.FloatField(source='score_given')
resultMaximum = serializers.SerializerMethodField()
comment = serializers.CharField()
def get_id(self, obj):
request = self.context.get('request')
return reverse(
'lti_consumer:lti-ags-view-results',
kwargs={
'lti_config_id': obj.line_item.lti_configuration.id,
'pk': obj.line_item.pk,
'user_id': obj.user_id,
},
request=request,
)
def get_scoreOf(self, obj):
request = self.context.get('request')
return reverse(
'lti_consumer:lti-ags-view-detail',
kwargs={
'lti_config_id': obj.line_item.lti_configuration.id,
'pk': obj.line_item.pk
},
request=request,
)
def get_resultMaximum(self, obj):
if obj.score_maximum <= 0:
return 1
return obj.score_maximum
class Meta:
model = LtiAgsScore
fields = (
'id',
'scoreOf',
'userId',
'resultScore',
'resultMaximum',
'comment',
)
......@@ -72,14 +72,20 @@ class TestLtiAuthentication(TestCase):
)
@ddt.data(
["https://purl.imsglobal.org/spec/lti-ags/scope/lineitem.readonly"],
["https://purl.imsglobal.org/spec/lti-ags/scope/lineitem"],
[
"https://purl.imsglobal.org/spec/lti-ags/scope/lineitem.readonly",
"https://purl.imsglobal.org/spec/lti-ags/scope/lineitem",
]
(["https://purl.imsglobal.org/spec/lti-ags/scope/lineitem.readonly"], True),
(["https://purl.imsglobal.org/spec/lti-ags/scope/lineitem"], True),
(["https://purl.imsglobal.org/spec/lti-ags/scope/result.readonly"], False),
(["https://purl.imsglobal.org/spec/lti-ags/scope/score"], False),
(
[
"https://purl.imsglobal.org/spec/lti-ags/scope/lineitem.readonly",
"https://purl.imsglobal.org/spec/lti-ags/scope/lineitem",
],
True
),
)
def test_read_only_lineitem_list(self, token_scopes):
@ddt.unpack
def test_read_only_lineitem_list(self, token_scopes, is_allowed):
"""
Test if LineItem is readable when any of the allowed scopes is
included in the token.
......@@ -95,14 +101,16 @@ class TestLtiAuthentication(TestCase):
# Test list view
mock_view.action = 'list'
self.assertTrue(
self.assertEqual(
perm_class.has_permission(self.mock_request, mock_view),
is_allowed,
)
# Test retrieve view
mock_view.action = 'retrieve'
self.assertTrue(
self.assertEqual(
perm_class.has_permission(self.mock_request, mock_view),
is_allowed,
)
def test_lineitem_no_permissions(self):
......@@ -134,13 +142,15 @@ class TestLtiAuthentication(TestCase):
@ddt.data(
(["https://purl.imsglobal.org/spec/lti-ags/scope/lineitem.readonly"], False),
(["https://purl.imsglobal.org/spec/lti-ags/scope/lineitem"], True),
(["https://purl.imsglobal.org/spec/lti-ags/scope/result.readonly"], False),
(["https://purl.imsglobal.org/spec/lti-ags/scope/score"], False),
(
[
"https://purl.imsglobal.org/spec/lti-ags/scope/lineitem.readonly",
"https://purl.imsglobal.org/spec/lti-ags/scope/lineitem",
],
True
)
),
)
@ddt.unpack
def test_lineitem_write_permissions(self, token_scopes, is_allowed):
......@@ -182,3 +192,57 @@ class TestLtiAuthentication(TestCase):
self.assertFalse(
perm_class.has_permission(self.mock_request, mock_view),
)
@ddt.data(
(["https://purl.imsglobal.org/spec/lti-ags/scope/lineitem.readonly"], False),
(["https://purl.imsglobal.org/spec/lti-ags/scope/lineitem"], False),
(["https://purl.imsglobal.org/spec/lti-ags/scope/result.readonly"], True),
(["https://purl.imsglobal.org/spec/lti-ags/scope/score"], False),
)
@ddt.unpack
def test_results_action_permissions(self, token_scopes, is_allowed):
"""
Test if write operations on LineItem are allowed with the correct token.
"""
perm_class = LtiAgsPermissions()
mock_view = MagicMock()
# Make token and include it in the mock request
token = self._make_token(token_scopes)
self.mock_request.headers = {
"Authorization": "Bearer {}".format(token)
}
# Test results view
mock_view.action = 'results'
self.assertEqual(
perm_class.has_permission(self.mock_request, mock_view),
is_allowed,
)
@ddt.data(
(["https://purl.imsglobal.org/spec/lti-ags/scope/lineitem.readonly"], False),
(["https://purl.imsglobal.org/spec/lti-ags/scope/lineitem"], False),
(["https://purl.imsglobal.org/spec/lti-ags/scope/result.readonly"], False),
(["https://purl.imsglobal.org/spec/lti-ags/scope/score"], True),
)
@ddt.unpack
def test_scores_action_permissions(self, token_scopes, is_allowed):
"""
Test if write operations on LineItem are allowed with the correct token.
"""
perm_class = LtiAgsPermissions()
mock_view = MagicMock()
# Make token and include it in the mock request
token = self._make_token(token_scopes)
self.mock_request.headers = {
"Authorization": "Bearer {}".format(token)
}
# Test scores view
mock_view.action = 'scores'
self.assertEqual(
perm_class.has_permission(self.mock_request, mock_view),
is_allowed,
)
# Generated by Django 2.2.16 on 2020-10-11 14:39
import django.core.validators
from django.db import migrations, models
import django.db.models.deletion
class Migration(migrations.Migration):
dependencies = [
('lti_consumer', '0002_ltiagslineitem'),
]
operations = [
migrations.CreateModel(
name='LtiAgsScore',
fields=[
('id', models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
('timestamp', models.DateTimeField()),
('score_given', models.FloatField(blank=True, null=True, validators=[django.core.validators.MinValueValidator(0)])),
('score_maximum', models.FloatField(blank=True, null=True, validators=[django.core.validators.MinValueValidator(0)])),
('comment', models.TextField(blank=True, null=True)),
('activity_progress', models.CharField(choices=[('Initialized', 'Initialized'), ('Started', 'Started'), ('InProgress', 'InProgress'), ('Submitted', 'Submitted'), ('Completed', 'Completed')], max_length=20)),
('grading_progress', models.CharField(choices=[('FullyGraded', 'FullyGraded'), ('Pending', 'Pending'), ('PendingManual', 'PendingManual'), ('Failed', 'Failed'), ('NotReady', 'NotReady')], max_length=20)),
('user_id', models.CharField(max_length=255)),
('line_item', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='scores', to='lti_consumer.LtiAgsLineItem')),
],
options={
'unique_together': {('line_item', 'user_id')},
},
),
]
......@@ -2,6 +2,8 @@
LTI configuration and linking models.
"""
from django.db import models
from django.core.validators import MinValueValidator
from django.core.exceptions import ValidationError
from opaque_keys.edx.django.models import UsageKeyField
......@@ -204,3 +206,93 @@ class LtiAgsLineItem(models.Model):
self.resource_link_id,
self.label,
)
class LtiAgsScore(models.Model):
"""
Model to store LineItem Score data for LTI Assignments and Grades service.
LTI-AGS Specification: https://www.imsglobal.org/spec/lti-ags/v2p0
Note: When implementing multi-tenancy support, this needs to be changed
and be tied to a deployment ID, because each deployment should isolate
it's resources.
.. no_pii:
"""
# LTI LineItem
# This links the score to a specific line item
line_item = models.ForeignKey(
LtiAgsLineItem,
on_delete=models.CASCADE,
related_name='scores',
)
timestamp = models.DateTimeField()
# All 'scoreGiven' and 'scoreMaximum' values MUST be positive numbers (including 0).
score_given = models.FloatField(null=True, blank=True, validators=[MinValueValidator(0)])
score_maximum = models.FloatField(null=True, blank=True, validators=[MinValueValidator(0)])
comment = models.TextField(null=True, blank=True)
# Activity Progress Choices
INITIALIZED = 'Initialized'
STARTED = 'Started'
IN_PROGRESS = 'InProgress'
SUBMITTED = 'Submitted'
COMPLETED = 'Completed'
ACTIVITY_PROGRESS_CHOICES = [
(INITIALIZED, INITIALIZED),
(STARTED, STARTED),
(IN_PROGRESS, IN_PROGRESS),
(SUBMITTED, SUBMITTED),
(COMPLETED, COMPLETED),
]
activity_progress = models.CharField(
max_length=20,
choices=ACTIVITY_PROGRESS_CHOICES
)
# Grading Progress Choices
FULLY_GRADED = 'FullyGraded'
PENDING = 'Pending'
PENDING_MANUAL = 'PendingManual'
FAILED = 'Failed'
NOT_READY = 'NotReady'
GRADING_PROGRESS_CHOICES = [
(FULLY_GRADED, FULLY_GRADED),
(PENDING, PENDING),
(PENDING_MANUAL, PENDING_MANUAL),
(FAILED, FAILED),
(NOT_READY, NOT_READY),
]
grading_progress = models.CharField(
max_length=20,
choices=GRADING_PROGRESS_CHOICES
)
user_id = models.CharField(max_length=255)
def clean(self):
super().clean()
# 'scoreMaximum' represents the denominator and MUST be present when 'scoreGiven' is present
if self.score_given and self.score_maximum is None:
raise ValidationError({'score_maximum': 'cannot be unset when score_given is set'})
def save(self, *args, **kwargs): # pylint: disable=arguments-differ
self.full_clean()
super().save(*args, **kwargs)
def __str__(self):
return "LineItem {line_item_id}: score {score_given} out of {score_maximum} - {grading_progress}".format(
line_item_id=self.line_item.id,
score_given=self.score_given,
score_maximum=self.score_maximum,
grading_progress=self.grading_progress
)
class Meta:
unique_together = (('line_item', 'user_id'),)
......@@ -6,14 +6,28 @@ from django.views.decorators.csrf import csrf_exempt
from django.views.decorators.http import require_http_methods
from django_filters.rest_framework import DjangoFilterBackend
from opaque_keys.edx.keys import UsageKey
from rest_framework import viewsets
from rest_framework import viewsets, status
from rest_framework.decorators import action
from rest_framework.response import Response
from lti_consumer.models import LtiAgsLineItem
from lti_consumer.lti_1p3.extensions.rest_framework.serializers import LtiAgsLineItemSerializer
from lti_consumer.lti_1p3.extensions.rest_framework.serializers import (
LtiAgsLineItemSerializer,
LtiAgsScoreSerializer,
LtiAgsResultSerializer,
)
from lti_consumer.lti_1p3.extensions.rest_framework.permissions import LtiAgsPermissions
from lti_consumer.lti_1p3.extensions.rest_framework.authentication import Lti1p3ApiAuthentication
from lti_consumer.lti_1p3.extensions.rest_framework.renderers import LineItemsRenderer, LineItemRenderer
from lti_consumer.lti_1p3.extensions.rest_framework.parsers import LineItemParser
from lti_consumer.lti_1p3.extensions.rest_framework.renderers import (
LineItemsRenderer,
LineItemRenderer,
LineItemScoreRenderer,
LineItemResultsRenderer
)
from lti_consumer.lti_1p3.extensions.rest_framework.parsers import (
LineItemParser,
LineItemScoreParser,
)
from lti_consumer.plugin.compat import (
run_xblock_handler,
run_xblock_handler_noauth,
......@@ -130,3 +144,80 @@ class LtiAgsLineItemViewset(viewsets.ModelViewSet):
def perform_create(self, serializer):
lti_configuration = self.request.lti_configuration
serializer.save(lti_configuration=lti_configuration)
@action(
detail=True,
methods=['GET'],
url_path='results/(?P<user_id>[^/.]+)?',
renderer_classes=[LineItemResultsRenderer]
)
def results(self, request, user_id=None, **kwargs):
"""
Return a Result list for an LtiAgsLineItem
URL Parameters:
* user_id (string): String external user id representation.
Query Parameters:
* limit (integer): The maximum number of records to return. Records are
sorted with most recent timestamp first
Returns:
* An array of Result records, formatted by LtiAgsResultSerializer
and returned with the media-type for LineItemResultsRenderer
"""
line_item = self.get_object()
scores = line_item.scores.filter(score_given__isnull=False).order_by('-timestamp')
if user_id:
scores = scores.filter(user_id=user_id)
if request.query_params.get('limit'):
scores = scores[:int(request.query_params.get('limit'))]
serializer = LtiAgsResultSerializer(
list(scores),
context={'request': self.request},
many=True,
)
return Response(serializer.data)
@action(
detail=True,
methods=['POST'],
parser_classes=[LineItemScoreParser],
renderer_classes=[LineItemScoreRenderer]
)
def scores(self, request, *args, **kwargs):
"""
Create a Score record for an LtiAgsLineItem
Data:
* A JSON object capable of being serialized by LtiAgsScoreSerializer
Returns:
* An copy of the saved record, formatted by LtiAgsScoreSerializer
and returned with the media-type for LineItemScoreRenderer
"""
line_item = self.get_object()
user_id = request.data.get('userId')
# Using `filter` and `first` so that when a score does not exist,
# `existing_score` is set to `None`. Using `get` will raise `DoesNotExist`
existing_score = line_item.scores.filter(user_id=user_id).first()
serializer = LtiAgsScoreSerializer(
instance=existing_score,
data=request.data,
context={'request': self.request},
)
serializer.is_valid(raise_exception=True)
serializer.save(line_item=line_item)
headers = self.get_success_headers(serializer.data)
return Response(
serializer.data,
status=status.HTTP_201_CREATED,
headers=headers
)
......@@ -12,17 +12,16 @@ from rest_framework.test import APITransactionTestCase
from lti_consumer.lti_xblock import LtiConsumerXBlock
from lti_consumer.models import LtiConfiguration, LtiAgsLineItem
from lti_consumer.models import LtiConfiguration, LtiAgsLineItem, LtiAgsScore
from lti_consumer.tests.unit.test_utils import make_xblock
@ddt.ddt
class TestLtiAgsLineItemViewSet(APITransactionTestCase):
class LtiAgsLineItemViewSetTestCase(APITransactionTestCase):
"""
Test `LtiAgsLineItemViewset` method.
Test `LtiAgsLineItemViewset` Class.
"""
def setUp(self):
super(TestLtiAgsLineItemViewSet, self).setUp()
super(LtiAgsLineItemViewSetTestCase, self).setUp()
# Create custom LTI Block
self.rsa_key_id = "1"
......@@ -68,14 +67,6 @@ class TestLtiAgsLineItemViewSet(APITransactionTestCase):
self.addCleanup(patcher.stop)
self._lti_block_patch = patcher.start()
# LineItem endpoint
self.lineitem_endpoint = reverse(
'lti_consumer:lti-ags-view-list',
kwargs={
"lti_config_id": self.lti_config.id
}
)
def _set_lti_token(self, scopes=None):
"""
Generates and sets a LTI Auth token in the request client.
......@@ -93,6 +84,24 @@ class TestLtiAgsLineItemViewSet(APITransactionTestCase):
HTTP_AUTHORIZATION="Bearer {}".format(token)
)
@ddt.ddt
class LtiAgsViewSetTokenTests(LtiAgsLineItemViewSetTestCase):
"""
Test `LtiAgsLineItemViewset` token based requests/responses.
"""
def setUp(self):
super().setUp()
# LineItem endpoint
self.lineitem_endpoint = reverse(
'lti_consumer:lti-ags-view-list',
kwargs={
"lti_config_id": self.lti_config.id
}
)
def test_lti_ags_view_no_token(self):
"""
Test the LTI AGS list view when there's no token.
......@@ -118,6 +127,24 @@ class TestLtiAgsLineItemViewSet(APITransactionTestCase):
response = self.client.get(self.lineitem_endpoint)
self.assertEqual(response.status_code, 403)
@ddt.ddt
class LtiAgsViewSetLineItemTests(LtiAgsLineItemViewSetTestCase):
"""
Test `LtiAgsLineItemViewset` LineItem based requests/responses.
"""
def setUp(self):
super().setUp()
# LineItem endpoint
self.lineitem_endpoint = reverse(
'lti_consumer:lti-ags-view-list',
kwargs={
"lti_config_id": self.lti_config.id
}
)
@ddt.data(
'https://purl.imsglobal.org/spec/lti-ags/scope/lineitem.readonly',
'https://purl.imsglobal.org/spec/lti-ags/scope/lineitem'
......@@ -274,3 +301,560 @@ class TestLtiAgsLineItemViewSet(APITransactionTestCase):
)
self.assertEqual(response.status_code, 400)
class LtiAgsViewSetScoresTests(LtiAgsLineItemViewSetTestCase):
"""
Test `LtiAgsLineItemViewset` Score Publishing requests/responses.
"""
def setUp(self):
super().setUp()
# Create LineItem
self.line_item = LtiAgsLineItem.objects.create(
lti_configuration=self.lti_config,
resource_id="test",
resource_link_id=self.xblock.location,
label="test label",
score_maximum=100
)
self.primary_user_id = "primary"
self.secondary_user_id = "secondary"
self.early_timestamp = "2020-01-01T18:54:36.736000+00:00"
self.middle_timestamp = "2021-01-01T18:54:36.736000+00:00"
self.late_timestamp = "2022-01-01T18:54:36.736000+00:00"
# Scores endpoint
self.scores_endpoint = reverse(
'lti_consumer:lti-ags-view-scores',
kwargs={
"lti_config_id": self.lti_config.id,
"pk": self.line_item.id
}
)
def test_create_score(self):
"""
Test the LTI AGS LineItem Score Creation.
"""
self._set_lti_token('https://purl.imsglobal.org/spec/lti-ags/scope/score')
# Create Score
response = self.client.post(
self.scores_endpoint,
data=json.dumps({
"timestamp": self.early_timestamp,
"scoreGiven": 83,
"scoreMaximum": 100,
"comment": "This is exceptional work.",
"activityProgress": LtiAgsScore.COMPLETED,
"gradingProgress": LtiAgsScore.FULLY_GRADED,
"userId": self.primary_user_id
}),
content_type="application/vnd.ims.lis.v1.score+json",
)
self.assertEqual(LtiAgsScore.objects.all().count(), 1)
self.assertEqual(response.status_code, 201)
# The serializer replaces `+00:00` with `Z`
response_timestamp = self.early_timestamp.replace('+00:00', 'Z')
self.assertEqual(
response.data,
{
"timestamp": response_timestamp,
"scoreGiven": 83.0,
"scoreMaximum": 100.0,
"comment": "This is exceptional work.",
"activityProgress": LtiAgsScore.COMPLETED,
"gradingProgress": LtiAgsScore.FULLY_GRADED,
"userId": self.primary_user_id
}
)
score = LtiAgsScore.objects.get(line_item=self.line_item, user_id=self.primary_user_id)
self.assertEqual(score.line_item.id, self.line_item.id)
self.assertEqual(score.timestamp.isoformat(), self.early_timestamp)
self.assertEqual(score.score_given, 83.0)
self.assertEqual(score.score_maximum, 100.0)
self.assertEqual(score.activity_progress, LtiAgsScore.COMPLETED)
self.assertEqual(score.grading_progress, LtiAgsScore.FULLY_GRADED)
self.assertEqual(score.user_id, self.primary_user_id)
def test_create_multiple_scores_with_multiple_users(self):
"""
Test the LTI AGS LineItem Score Creation on the same LineItem for different users.
"""
self._set_lti_token('https://purl.imsglobal.org/spec/lti-ags/scope/score')
# Create Score
response = self.client.post(
self.scores_endpoint,
data=json.dumps({
"timestamp": self.early_timestamp,
"scoreGiven": 21,
"scoreMaximum": 100,
"comment": "This is exceptional work.",
"activityProgress": LtiAgsScore.INITIALIZED,
"gradingProgress": LtiAgsScore.NOT_READY,
"userId": self.primary_user_id
}),
content_type="application/vnd.ims.lis.v1.score+json",
)
self.assertEqual(LtiAgsScore.objects.all().count(), 1)
self.assertEqual(response.status_code, 201)
# Create 2nd Score with same timestamp, but different data
response = self.client.post(
self.scores_endpoint,
data=json.dumps({
"timestamp": self.early_timestamp,
"scoreGiven": 83,
"scoreMaximum": 100,
"comment": "This is exceptional work.",
"activityProgress": LtiAgsScore.COMPLETED,
"gradingProgress": LtiAgsScore.FULLY_GRADED,
"userId": self.secondary_user_id
}),
content_type="application/vnd.ims.lis.v1.score+json",
)
self.assertEqual(LtiAgsScore.objects.all().count(), 2)
self.assertEqual(response.status_code, 201)
# Check db record contents
# Score for primary user
primary_user_score = LtiAgsScore.objects.get(line_item=self.line_item, user_id=self.primary_user_id)
self.assertEqual(primary_user_score.line_item.id, self.line_item.id)
self.assertEqual(primary_user_score.timestamp.isoformat(), self.early_timestamp)
self.assertEqual(primary_user_score.score_given, 21.0)
self.assertEqual(primary_user_score.score_maximum, 100.0)
self.assertEqual(primary_user_score.activity_progress, LtiAgsScore.INITIALIZED)
self.assertEqual(primary_user_score.grading_progress, LtiAgsScore.NOT_READY)
self.assertEqual(primary_user_score.user_id, self.primary_user_id)
# Score for secondary user
secondary_user_score = LtiAgsScore.objects.get(line_item=self.line_item, user_id=self.secondary_user_id)
self.assertEqual(secondary_user_score.line_item.id, self.line_item.id)
self.assertEqual(secondary_user_score.timestamp.isoformat(), self.early_timestamp)
self.assertEqual(secondary_user_score.score_given, 83.0)
self.assertEqual(secondary_user_score.score_maximum, 100.0)
self.assertEqual(secondary_user_score.activity_progress, LtiAgsScore.COMPLETED)
self.assertEqual(secondary_user_score.grading_progress, LtiAgsScore.FULLY_GRADED)
self.assertEqual(secondary_user_score.user_id, self.secondary_user_id)
def test_create_multiple_scores_with_later_timestamp(self):
"""
Test the LTI AGS LineItem Score updating with a later timestamp updates the record.
"""
self._set_lti_token('https://purl.imsglobal.org/spec/lti-ags/scope/score')
# Create Score
response = self.client.post(
self.scores_endpoint,
data=json.dumps({
"timestamp": self.early_timestamp,
"scoreGiven": 21,
"scoreMaximum": 100,
"comment": "This is exceptional work.",
"activityProgress": LtiAgsScore.INITIALIZED,
"gradingProgress": LtiAgsScore.NOT_READY,
"userId": self.primary_user_id
}),
content_type="application/vnd.ims.lis.v1.score+json",
)
self.assertEqual(LtiAgsScore.objects.all().count(), 1)
self.assertEqual(response.status_code, 201)
# Create 2nd Score with same timestamp, but different data
response = self.client.post(
self.scores_endpoint,
data=json.dumps({
"timestamp": self.late_timestamp,
"scoreGiven": 83,
"scoreMaximum": 100,
"comment": "This is exceptional work.",
"activityProgress": LtiAgsScore.COMPLETED,
"gradingProgress": LtiAgsScore.FULLY_GRADED,
"userId": self.primary_user_id
}),
content_type="application/vnd.ims.lis.v1.score+json",
)
self.assertEqual(LtiAgsScore.objects.all().count(), 1)
self.assertEqual(response.status_code, 201)
# Check db record contents
score = LtiAgsScore.objects.get(line_item=self.line_item, user_id=self.primary_user_id)
self.assertEqual(score.line_item.id, self.line_item.id)
self.assertEqual(score.timestamp.isoformat(), self.late_timestamp)
self.assertEqual(score.score_given, 83.0)
self.assertEqual(score.score_maximum, 100.0)
self.assertEqual(score.activity_progress, LtiAgsScore.COMPLETED)
self.assertEqual(score.grading_progress, LtiAgsScore.FULLY_GRADED)
self.assertEqual(score.user_id, self.primary_user_id)
def test_create_multiple_scores_with_same_timestamp(self):
"""
Test the LTI AGS LineItem Score updating with an existing timestamp fails to update the record.
"""
self._set_lti_token('https://purl.imsglobal.org/spec/lti-ags/scope/score')
# Create Score
response = self.client.post(
self.scores_endpoint,
data=json.dumps({
"timestamp": self.early_timestamp,
"scoreGiven": 21,
"scoreMaximum": 100,
"comment": "This is exceptional work.",
"activityProgress": LtiAgsScore.INITIALIZED,
"gradingProgress": LtiAgsScore.NOT_READY,
"userId": self.primary_user_id
}),
content_type="application/vnd.ims.lis.v1.score+json",
)
self.assertEqual(LtiAgsScore.objects.all().count(), 1)
self.assertEqual(response.status_code, 201)
# Create 2nd Score with same timestamp, but different data
response = self.client.post(
self.scores_endpoint,
data=json.dumps({
"timestamp": self.early_timestamp,
"scoreGiven": 83,
"scoreMaximum": 100,
"comment": "This is exceptional work.",
"activityProgress": LtiAgsScore.COMPLETED,
"gradingProgress": LtiAgsScore.FULLY_GRADED,
"userId": self.primary_user_id
}),
content_type="application/vnd.ims.lis.v1.score+json",
)
self.assertEqual(LtiAgsScore.objects.all().count(), 1)
self.assertEqual(response.status_code, 400)
# Check db record contents are the original data
score = LtiAgsScore.objects.get(line_item=self.line_item, user_id=self.primary_user_id)
self.assertEqual(score.line_item.id, self.line_item.id)
self.assertEqual(score.timestamp.isoformat(), self.early_timestamp)
self.assertEqual(score.score_given, 21.0)
self.assertEqual(score.score_maximum, 100.0)
self.assertEqual(score.activity_progress, LtiAgsScore.INITIALIZED)
self.assertEqual(score.grading_progress, LtiAgsScore.NOT_READY)
self.assertEqual(score.user_id, self.primary_user_id)
def test_create_second_score_with_earlier_timestamp(self):
"""
Test the LTI AGS LineItem Score updating with an earlier timestamp fails to update the record.
"""
self._set_lti_token('https://purl.imsglobal.org/spec/lti-ags/scope/score')
# Create Score
response = self.client.post(
self.scores_endpoint,
data=json.dumps({
"timestamp": self.late_timestamp,
"scoreGiven": 21,
"scoreMaximum": 100,
"comment": "This is exceptional work.",
"activityProgress": LtiAgsScore.INITIALIZED,
"gradingProgress": LtiAgsScore.NOT_READY,
"userId": self.primary_user_id
}),
content_type="application/vnd.ims.lis.v1.score+json",
)
self.assertEqual(LtiAgsScore.objects.all().count(), 1)
self.assertEqual(response.status_code, 201)
# Create 2nd Score with earlier timestamp, and different data
response = self.client.post(
self.scores_endpoint,
data=json.dumps({
"timestamp": self.early_timestamp,
"scoreGiven": 83,
"scoreMaximum": 100,
"comment": "This is exceptional work.",
"activityProgress": LtiAgsScore.COMPLETED,
"gradingProgress": LtiAgsScore.FULLY_GRADED,
"userId": self.primary_user_id
}),
content_type="application/vnd.ims.lis.v1.score+json",
)
self.assertEqual(LtiAgsScore.objects.all().count(), 1)
self.assertEqual(response.status_code, 400)
# Check db record contents are the original data
score = LtiAgsScore.objects.get(line_item=self.line_item, user_id=self.primary_user_id)
self.assertEqual(score.line_item.id, self.line_item.id)
self.assertEqual(score.timestamp.isoformat(), self.late_timestamp)
self.assertEqual(score.score_given, 21.0)
self.assertEqual(score.score_maximum, 100.0)
self.assertEqual(score.activity_progress, LtiAgsScore.INITIALIZED)
self.assertEqual(score.grading_progress, LtiAgsScore.NOT_READY)
self.assertEqual(score.user_id, self.primary_user_id)
def test_create_score_with_missing_score_maximum(self):
"""
Test invalid request with missing scoreMaximum.
"""
self._set_lti_token('https://purl.imsglobal.org/spec/lti-ags/scope/score')
# Create invalid Score
response = self.client.post(
self.scores_endpoint,
data=json.dumps({
"timestamp": self.late_timestamp,
"scoreGiven": 21,
"comment": "This is exceptional work.",
"activityProgress": LtiAgsScore.INITIALIZED,
"gradingProgress": LtiAgsScore.NOT_READY,
"userId": self.primary_user_id
}),
content_type="application/vnd.ims.lis.v1.score+json",
)
self.assertEqual(LtiAgsScore.objects.all().count(), 0)
self.assertEqual(response.status_code, 400)
assert 'scoreMaximum' in response.data.keys()
def test_erase_score(self):
"""
Test erasing LTI AGS Scores by omitting scoreGiven and scoreMaximum.
"""
# Have a score already existing
LtiAgsScore.objects.create(
line_item=self.line_item,
timestamp=self.early_timestamp,
score_given=25,
score_maximum=100,
comment="This is exceptional work.",
activity_progress=LtiAgsScore.COMPLETED,
grading_progress=LtiAgsScore.FULLY_GRADED,
user_id=self.primary_user_id
)
self._set_lti_token('https://purl.imsglobal.org/spec/lti-ags/scope/score')
# Erase Score
response = self.client.post(
self.scores_endpoint,
data=json.dumps({
"timestamp": self.late_timestamp,
"comment": None,
"activityProgress": LtiAgsScore.INITIALIZED,
"gradingProgress": LtiAgsScore.NOT_READY,
"userId": self.primary_user_id
}),
content_type="application/vnd.ims.lis.v1.score+json",
)
self.assertEqual(LtiAgsScore.objects.all().count(), 1)
self.assertEqual(response.status_code, 201)
# The serializer replaces `+00:00` with `Z`
response_timestamp = self.late_timestamp.replace('+00:00', 'Z')
self.assertEqual(
response.data,
{
"timestamp": response_timestamp,
"scoreGiven": None,
"scoreMaximum": None,
"comment": None,
"activityProgress": LtiAgsScore.INITIALIZED,
"gradingProgress": LtiAgsScore.NOT_READY,
"userId": self.primary_user_id
}
)
# Check db record contents (because we don't delete the record, just blank it out)
score = LtiAgsScore.objects.get(line_item=self.line_item, user_id=self.primary_user_id)
self.assertEqual(score.line_item.id, self.line_item.id)
self.assertEqual(score.timestamp.isoformat(), self.late_timestamp)
self.assertEqual(score.score_given, None)
self.assertEqual(score.score_maximum, None)
self.assertEqual(score.activity_progress, LtiAgsScore.INITIALIZED)
self.assertEqual(score.grading_progress, LtiAgsScore.NOT_READY)
self.assertEqual(score.user_id, self.primary_user_id)
class LtiAgsViewSetResultsTests(LtiAgsLineItemViewSetTestCase):
"""
Test `LtiAgsLineItemViewset` Results retrieval requests/responses.
"""
def setUp(self):
super().setUp()
# Create LineItem
self.line_item = LtiAgsLineItem.objects.create(
lti_configuration=self.lti_config,
resource_id="test",
resource_link_id=self.xblock.location,
label="test label",
score_maximum=100
)
self.early_timestamp = "2020-01-01T18:54:36.736000+00:00"
self.middle_timestamp = "2021-01-01T18:54:36.736000+00:00"
self.late_timestamp = "2022-01-01T18:54:36.736000+00:00"
# Create Scores
self.primary_user_id = "primary"
LtiAgsScore.objects.create(
line_item=self.line_item,
timestamp=self.late_timestamp,
score_given=83,
score_maximum=100,
comment="This is exceptional work.",
activity_progress=LtiAgsScore.COMPLETED,
grading_progress=LtiAgsScore.FULLY_GRADED,
user_id=self.primary_user_id
)
self.secondary_user_id = "secondary"
LtiAgsScore.objects.create(
line_item=self.line_item,
timestamp=self.middle_timestamp,
score_given=25,
score_maximum=100,
comment="This is not great work.",
activity_progress=LtiAgsScore.COMPLETED,
grading_progress=LtiAgsScore.FULLY_GRADED,
user_id=self.secondary_user_id
)
self.empty_score_user_id = "empty_score_user"
LtiAgsScore.objects.create(
line_item=self.line_item,
timestamp=self.early_timestamp,
score_given=None,
score_maximum=None,
comment=None,
activity_progress=LtiAgsScore.INITIALIZED,
grading_progress=LtiAgsScore.NOT_READY,
user_id=self.empty_score_user_id
)
# LineItem endpoint
self.lineitem_endpoint = reverse(
'lti_consumer:lti-ags-view-detail',
kwargs={
"lti_config_id": self.lti_config.id,
"pk": self.line_item.id
}
)
# Results endpoint
self.results_endpoint = reverse(
'lti_consumer:lti-ags-view-results',
kwargs={
"lti_config_id": self.lti_config.id,
"pk": self.line_item.id
}
)
def test_retrieve_results(self):
"""
Test the LTI AGS LineItem Result Retrieval.
"""
self._set_lti_token('https://purl.imsglobal.org/spec/lti-ags/scope/result.readonly')
# Create Score
response = self.client.get(self.results_endpoint)
self.assertEqual(response.status_code, 200)
# There should be 2 results (not include the empty score user's result)
self.assertEqual(len(response.data), 2)
# Check the data
primary_user_results_endpoint = reverse(
'lti_consumer:lti-ags-view-results',
kwargs={
"lti_config_id": self.lti_config.id,
"pk": self.line_item.id,
"user_id": self.primary_user_id
}
)
secondary_user_results_endpoint = reverse(
'lti_consumer:lti-ags-view-results',
kwargs={
"lti_config_id": self.lti_config.id,
"pk": self.line_item.id,
"user_id": self.secondary_user_id
}
)
self.assertEqual(
[dict(d) for d in response.data],
[
{
"id": "http://testserver" + primary_user_results_endpoint,
"scoreOf": "http://testserver" + self.lineitem_endpoint,
"userId": self.primary_user_id,
"resultScore": 83.0,
"resultMaximum": 100.0,
"comment": "This is exceptional work."
},
{
"id": "http://testserver" + secondary_user_results_endpoint,
"scoreOf": "http://testserver" + self.lineitem_endpoint,
"userId": self.secondary_user_id,
"resultScore": 25.0,
"resultMaximum": 100.0,
"comment": "This is not great work."
}
]
)
def test_retrieve_results_for_user_id(self):
"""
Test the LTI AGS LineItem Resul Retrieval for a single user.
"""
self._set_lti_token('https://purl.imsglobal.org/spec/lti-ags/scope/result.readonly')
results_user_endpoint = reverse(
'lti_consumer:lti-ags-view-results',
kwargs={
"lti_config_id": self.lti_config.id,
"pk": self.line_item.id,
"user_id": self.secondary_user_id
}
)
# Request results with userId
response = self.client.get(results_user_endpoint, data={"userId": self.secondary_user_id})
self.assertEqual(response.status_code, 200)
# There should be 1 result for that user
self.assertEqual(len(response.data), 1)
self.assertEqual(response.data[0]['userId'], self.secondary_user_id)
def test_retrieve_results_with_limit(self):
"""
Test the LTI AGS LineItem Result Retrieval with record limit.
"""
self._set_lti_token('https://purl.imsglobal.org/spec/lti-ags/scope/result.readonly')
# Request results with limit
response = self.client.get(self.results_endpoint, data={"limit": 1})
self.assertEqual(response.status_code, 200)
# There should be 1 results, and it should be the one with the latest timestamp
# Since the AGS Result Service format does not return a timestamp, the only
# way to know which record should be returned is to compare against a known
# value from the records created in the `setUp`. In this case, the
# `primary_user_id` was assigned to the record with the `late_timestamp`
self.assertEqual(len(response.data), 1)
self.assertEqual(response.data[0]['userId'], self.primary_user_id)
......@@ -8,16 +8,16 @@ from jwkest.jwk import RSAKey
from mock import patch
from lti_consumer.lti_xblock import LtiConsumerXBlock
from lti_consumer.models import LtiAgsLineItem, LtiConfiguration
from lti_consumer.models import LtiAgsLineItem, LtiConfiguration, LtiAgsScore
from lti_consumer.tests.unit.test_utils import make_xblock
class TestLtiCofigurationModel(TestCase):
class TestLtiConfigurationModel(TestCase):
"""
Unit tests for LtiConfiguration model methods.
"""
def setUp(self):
super(TestLtiCofigurationModel, self).setUp()
super().setUp()
self.rsa_key_id = "1"
# Generate RSA and save exports
......@@ -114,7 +114,7 @@ class TestLtiAgsLineItemModel(TestCase):
Unit tests for LtiAgsLineItem model methods.
"""
def setUp(self):
super(TestLtiAgsLineItemModel, self).setUp()
super().setUp()
self.dummy_location = 'block-v1:course+test+2020+type@problem+block@test'
self.lti_ags_model = LtiAgsLineItem.objects.create(
......@@ -133,3 +133,39 @@ class TestLtiAgsLineItemModel(TestCase):
str(self.lti_ags_model),
"block-v1:course+test+2020+type@problem+block@test - this-is-a-test"
)
class TestLtiAgsScoreModel(TestCase):
"""
Unit tests for LtiAgsScore model methods.
"""
def setUp(self):
super().setUp()
self.dummy_location = 'block-v1:course+test+2020+type@problem+block@test'
self.line_item = LtiAgsLineItem.objects.create(
lti_configuration=None,
resource_id="test-id",
label="this-is-a-test",
resource_link_id=self.dummy_location,
score_maximum=100,
)
self.score = LtiAgsScore.objects.create(
line_item=self.line_item,
timestamp='2020-10-04T18:54:46.736+00:00',
score_given=10,
score_maximum=100,
comment='Better luck next time',
grading_progress=LtiAgsScore.FULLY_GRADED,
activity_progress=LtiAgsScore.COMPLETED,
user_id='test-user'
)
def test_repr(self):
"""
Test String representation of model.
"""
self.assertEqual(
str(self.score),
"LineItem 1: score 10.0 out of 100.0 - FullyGraded"
)
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment