Andal.LND/edx-platform
6
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Update the access to gradebook to allow course staff and admins

Browse Source
This commit is contained in:
Simon Chen
2018-12-07 16:39:33 -05:00
parent 9866b045c0
commit 02103f7660
7 changed files with 1827 additions and 1709 deletions
Download Patch File Download Diff File Expand all files Collapse all files

546
lms/djangoapps/grades/api/v1/gradebook_views.py
View File

@@ -1,22 +1,99 @@
"""
Defines an endpoint for retrieving assignment type and subsection info for a course.
Defines an endpoint for gradebook data related to a course.
"""
from collections import namedtuple
from contextlib import contextmanager
from functools import wraps
from opaque_keys.edx.keys import CourseKey
from django.urls import reverse
from rest_framework import status
from rest_framework.generics import GenericAPIView
from rest_framework.response import Response
from six import text_type
from util.date_utils import to_timestamp
from courseware.courses import get_course_by_id
from lms.djangoapps.grades.api.serializers import StudentGradebookEntrySerializer
from lms.djangoapps.grades.api.v1.utils import (
USER_MODEL,
CourseEnrollmentPagination,
GradeViewMixin,
PaginatedAPIView,
get_course_key,
verify_course_exists
)
from lms.djangoapps.grades.config.waffle import WRITABLE_GRADEBOOK, waffle_flags
from lms.djangoapps.grades.constants import ScoreDatabaseTableEnum
from lms.djangoapps.grades.course_data import CourseData
from lms.djangoapps.grades.course_grade_factory import CourseGradeFactory
from lms.djangoapps.grades.events import SUBSECTION_GRADE_CALCULATED, subsection_grade_calculated
from lms.djangoapps.grades.models import (
PersistentCourseGrade,
PersistentSubsectionGrade,
PersistentSubsectionGradeOverride,
PersistentSubsectionGradeOverrideHistory
)
from lms.djangoapps.grades.subsection_grade import CreateSubsectionGrade
from lms.djangoapps.grades.tasks import are_grades_frozen, recalculate_subsection_grade_v3
from opaque_keys import InvalidKeyError
from opaque_keys.edx.keys import CourseKey, UsageKey
from openedx.core.djangoapps.course_groups import cohorts
from openedx.core.djangoapps.util.forms import to_bool
from openedx.core.lib.api.view_utils import DeveloperErrorViewMixin, view_auth_classes
from openedx.core.lib.cache_utils import request_cached
from student.auth import has_course_author_access
from student.models import CourseEnrollment
from student.roles import BulkRoleCache
from track.event_transaction_utils import (
create_new_event_transaction_id,
get_event_transaction_id,
get_event_transaction_type,
set_event_transaction_type
)
from xmodule.modulestore.django import modulestore
from xmodule.util.misc import get_default_short_labeler
@contextmanager
def bulk_gradebook_view_context(course_key, users):
"""
Prefetches all course and subsection grades in the given course for the given
list of users, also, fetch all the score relavant data,
storing the result in a RequestCache and deleting grades on context exit.
"""
PersistentSubsectionGrade.prefetch(course_key, users)
PersistentCourseGrade.prefetch(course_key, users)
CourseEnrollment.bulk_fetch_enrollment_states(users, course_key)
cohorts.bulk_cache_cohorts(course_key, users)
BulkRoleCache.prefetch(users)
yield
PersistentSubsectionGrade.clear_prefetched_data(course_key)
PersistentCourseGrade.clear_prefetched_data(course_key)
def verify_writable_gradebook_enabled(view_func):
"""
A decorator to wrap a view function that takes `course_key` as a parameter.
Raises:
A 403 API error if the writable gradebook feature is not enabled for the given course.
"""
@wraps(view_func)
def wrapped_function(self, request, **kwargs):
"""
Wraps the given view function.
"""
course_key = get_course_key(request, kwargs.get('course_id'))
if not waffle_flags()[WRITABLE_GRADEBOOK].is_enabled(course_key):
raise self.api_error(
status_code=status.HTTP_403_FORBIDDEN,
developer_message='The writable gradebook feature is not enabled for this course.',
error_code='feature_not_enabled'
)
return view_func(self, request, **kwargs)
return wrapped_function
@view_auth_classes()
class BaseCourseView(DeveloperErrorViewMixin, GenericAPIView):
"""
@@ -230,3 +307,468 @@ class CourseGradingView(BaseCourseView):
'display_name': subsection.display_name,
})
return subsections
@view_auth_classes()
class GradebookView(GradeViewMixin, PaginatedAPIView):
"""
**Use Case**
* Get course gradebook entries of a single user in a course,
or of all users who are actively enrolled in a course. The currently logged-in user may request
all enrolled user's grades information if they are allowed.
**Example Request**
GET /api/grades/v1/gradebook/{course_id}/ - Get gradebook entries for all users in course
GET /api/grades/v1/gradebook/{course_id}/?username={username} - Get grades for specific user in course
GET /api/grades/v1/gradebook/{course_id}/?username_contains={username_contains}
GET /api/grades/v1/gradebook/{course_id}/?cohort_id={cohort_id}
GET /api/grades/v1/gradebook/{course_id}/?enrollment_mode={enrollment_mode}
**GET Parameters**
A GET request may include the following query parameters.
* username: (optional) A string representation of a user's username.
* username_contains: (optional) A substring against which a case-insensitive substring filter will be performed
on the USER_MODEL.username field.
* cohort_id: (optional) The id of a cohort in this course. If present, will return grades
only for course enrollees who belong to that cohort.
* enrollment_mode: (optional) The slug of an enrollment mode (e.g. "verified"). If present, will return grades
only for course enrollees with the given enrollment mode.
**GET Response Values**
If the request for gradebook data is successful,
an HTTP 200 "OK" response is returned.
The HTTP 200 response for a single has the following values:
* course_id: A string representation of a Course ID.
* email: A string representation of a user's email.
* user_id: The user's integer id.
* username: A string representation of a user's username passed in the request.
* full_name: A string representation of the user's full name.
* passed: Boolean representing whether the course has been
passed according to the course's grading policy.
* percent: A float representing the overall grade for the course
* letter_grade: A letter grade as defined in grading policy (e.g. 'A' 'B' 'C' for 6.002x) or None
* progress_page_url: A link to the user's progress page.
* section_breakdown: A list of subsection grade details, as specified below.
A response for all user's grades in the course is paginated, and contains "count", "next" and "previous"
keys, along with the actual data contained in a "results" list.
An HTTP 404 may be returned for the following reasons:
* The requested course_key is invalid.
* No course corresponding to the requested key exists.
* No user corresponding to the requested username exists.
* The requested user is not enrolled in the requested course.
An HTTP 403 may be returned if the `writable_gradebook` feature is not
enabled for this course.
**Example GET Response**
{
"course_id": "course-v1:edX+DemoX+Demo_Course",
"email": "staff@example.com",
"user_id": 9,
"username": "staff",
"full_name": "",
"passed": false,
"percent": 0.36,
"letter_grade": null,
"progress_page_url": "/courses/course-v1:edX+DemoX+Demo_Course/progress/9/",
"section_breakdown": [
{
"are_grades_published": true,
"auto_grade": false,
"category": null,
"chapter_name": "Introduction",
"comment": "",
"detail": "",
"displayed_value": "0.00",
"is_graded": false,
"grade_description": "(0.00/0.00)",
"is_ag": false,
"is_average": false,
"is_manually_graded": false,
"label": null,
"letter_grade": null,
"module_id": "block-v1:edX+DemoX+Demo_Course+type@sequential+block@edx_introduction",
"percent": 0.0,
"score_earned": 0.0,
"score_possible": 0.0,
"section_block_id": "block-v1:edX+DemoX+Demo_Course+type@chapter+block@abcdefgh123",
"subsection_name": "Demo Course Overview"
},
],
}
**Paginated GET response**
When requesting gradebook entries for all users, the response is paginated and contains the following values:
* count: The total number of user gradebook entries for this course.
* next: The URL containing the next page of data.
* previous: The URL containing the previous page of data.
* results: A list of user gradebook entries, structured as above.
Note: It's important that `GradeViewMixin` is the first inherited class here, so that
self.api_error returns error responses as expected.
"""
pagination_class = CourseEnrollmentPagination
def _section_breakdown(self, course, graded_subsections, course_grade):
"""
Given a course_grade and a list of graded subsections for a given course,
returns a list of grade data broken down by subsection.
Args:
course: A Course Descriptor object
graded_subsections: A list of graded subsection objects in the given course.
course_grade: A CourseGrade object.
"""
breakdown = []
default_labeler = get_default_short_labeler(course)
for subsection in graded_subsections:
subsection_grade = course_grade.subsection_grade(subsection.location)
short_label = default_labeler(subsection_grade.format)
graded_description = 'Not Attempted'
score_earned = 0
score_possible = 0
# For ZeroSubsectionGrades, we don't want to crawl the subsection's
# subtree to find the problem scores specific to this user
# (ZeroSubsectionGrade.attempted_graded is always False).
# We've already fetched the whole course structure in a non-specific way
# when creating `graded_subsections`. Looking at the problem scores
# specific to this user (the user in `course_grade.user`) would require
# us to re-fetch the user-specific course structure from the modulestore,
# which is a costly operation.
if subsection_grade.attempted_graded:
graded_description = '({earned:.2f}/{possible:.2f})'.format(
earned=subsection_grade.graded_total.earned,
possible=subsection_grade.graded_total.possible,
)
score_earned = subsection_grade.graded_total.earned
score_possible = subsection_grade.graded_total.possible
# TODO: https://openedx.atlassian.net/browse/EDUCATOR-3559 -- Some fields should be renamed, others removed:
# 'displayed_value' should maybe be 'description_percent'
# 'grade_description' should be 'description_ratio'
breakdown.append({
'category': subsection_grade.format,
'displayed_value': '{:.2f}'.format(subsection_grade.percent_graded),
'is_graded': subsection_grade.graded,
'grade_description': graded_description,
'label': short_label,
'letter_grade': course_grade.letter_grade,
'module_id': text_type(subsection_grade.location),
'percent': subsection_grade.percent_graded,
'score_earned': score_earned,
'score_possible': score_possible,
'subsection_name': subsection_grade.display_name,
})
return breakdown
def _gradebook_entry(self, user, course, graded_subsections, course_grade):
"""
Returns a dictionary of course- and subsection-level grade data for
a given user in a given course.
Args:
user: A User object.
course: A Course Descriptor object.
graded_subsections: A list of graded subsections in the given course.
course_grade: A CourseGrade object.
"""
user_entry = self._serialize_user_grade(user, course.id, course_grade)
breakdown = self._section_breakdown(course, graded_subsections, course_grade)
user_entry['section_breakdown'] = breakdown
user_entry['progress_page_url'] = reverse(
'student_progress',
kwargs=dict(course_id=text_type(course.id), student_id=user.id)
)
user_entry['user_id'] = user.id
user_entry['full_name'] = user.get_full_name()
return user_entry
@verify_course_exists
@verify_writable_gradebook_enabled
@course_author_access_required
def get(self, request, course_key):
"""
Returns a gradebook entry/entries (i.e. both course and subsection-level grade data)
for all users enrolled in a course, or a single user enrolled in a course
if a `username` parameter is provided.
Args:
request: A Django request object.
course_key: The edx course opaque key of a course object.
"""
course = get_course_by_id(course_key, depth=None)
# We fetch the entire course structure up-front, and use this when iterating
# over users to determine their subsection grades. We purposely avoid fetching
# the user-specific course structure for each user, because that is very expensive.
course_data = CourseData(user=None, course=course)
graded_subsections = list(graded_subsections_for_course(course_data.collected_structure))
if request.GET.get('username'):
with self._get_user_or_raise(request, course_key) as grade_user:
course_grade = CourseGradeFactory().read(grade_user, course)
entry = self._gradebook_entry(grade_user, course, graded_subsections, course_grade)
serializer = StudentGradebookEntrySerializer(entry)
return Response(serializer.data)
else:
filter_kwargs = {}
related_models = []
if request.GET.get('username_contains'):
filter_kwargs['user__username__icontains'] = request.GET.get('username_contains')
related_models.append('user')
if request.GET.get('cohort_id'):
cohort = cohorts.get_cohort_by_id(course_key, request.GET.get('cohort_id'))
if cohort:
filter_kwargs['user__in'] = cohort.users.all()
else:
filter_kwargs['user__in'] = []
if request.GET.get('enrollment_mode'):
filter_kwargs['mode'] = request.GET.get('enrollment_mode')
entries = []
users = self._paginate_users(course_key, filter_kwargs, related_models)
with bulk_gradebook_view_context(course_key, users):
for user, course_grade, exc in CourseGradeFactory().iter(
users, course_key=course_key, collected_block_structure=course_data.collected_structure
):
if not exc:
entries.append(self._gradebook_entry(user, course, graded_subsections, course_grade))
serializer = StudentGradebookEntrySerializer(entries, many=True)
return self.get_paginated_response(serializer.data)
def graded_subsections_for_course(course_structure):
"""
Given a course block structure, yields the subsections of the course that are graded.
Args:
course_structure: A course structure object. Not user-specific.
"""
for chapter_key in course_structure.get_children(course_structure.root_block_usage_key):
for subsection_key in course_structure.get_children(chapter_key):
subsection = course_structure[subsection_key]
if subsection.graded:
yield subsection
GradebookUpdateResponseItem = namedtuple('GradebookUpdateResponseItem', ['user_id', 'usage_id', 'success', 'reason'])
@view_auth_classes()
class GradebookBulkUpdateView(GradeViewMixin, PaginatedAPIView):
"""
**Use Case**
Creates `PersistentSubsectionGradeOverride` objects for multiple (user_id, usage_id)
pairs in a given course, and invokes a Django signal to update subsection grades in
an asynchronous celery task.
**Example Request**
POST /api/grades/v1/gradebook/{course_id}/bulk-update
**POST Parameters**
This endpoint does not accept any URL parameters.
**Example POST Data**
[
{
"user_id": 9,
"usage_id": "block-v1:edX+DemoX+Demo_Course+type@sequential+block@basic_questions",
"grade": {
"earned_all_override": 11,
"possible_all_override": 11,
"earned_graded_override": 11,
"possible_graded_override": 11
}
},
{
"user_id": 9,
"usage_id": "block-v1:edX+DemoX+Demo_Course+type@sequential+block@advanced_questions",
"grade": {
"earned_all_override": 10,
"possible_all_override": 15,
"earned_graded_override": 9,
"possible_graded_override": 12
}
}
]
**POST Response Values**
An HTTP 202 may be returned if a grade override was created for each of the requested (user_id, usage_id)
pairs in the request data.
An HTTP 403 may be returned if the `writable_gradebook` feature is not
enabled for this course.
An HTTP 404 may be returned for the following reasons:
* The requested course_key is invalid.
* No course corresponding to the requested key exists.
* The requesting user is not enrolled in the requested course.
An HTTP 422 may be returned if any of the requested (user_id, usage_id) pairs
did not have a grade override created due to some exception. A `reason` detailing the exception
is provided with each response item.
**Example successful POST Response**
[
{
"user_id": 9,
"usage_id": "some-requested-usage-id",
"success": true,
"reason": null
},
{
"user_id": 9,
"usage_id": "an-invalid-usage-id",
"success": false,
"reason": "<class 'opaque_keys.edx.locator.BlockUsageLocator'>: not-a-valid-usage-key"
},
{
"user_id": 9,
"usage_id": "a-valid-usage-key-that-doesn't-exist",
"success": false,
"reason": "a-valid-usage-key-that-doesn't-exist does not exist in this course"
},
{
"user_id": 1234-I-DO-NOT-EXIST,
"usage_id": "a-valid-usage-key",
"success": false,
"reason": "User matching query does not exist."
}
]
"""
@verify_course_exists
@verify_writable_gradebook_enabled
@course_author_access_required
def post(self, request, course_key):
"""
Creates or updates `PersistentSubsectionGradeOverrides` for the (user_id, usage_key)
specified in the request data. The `SUBSECTION_OVERRIDE_CHANGED` signal is invoked
after the grade override is created, which triggers a celery task to update the
course and subsection grades for the specified user.
"""
if are_grades_frozen(course_key):
raise self.api_error(
status_code=status.HTTP_403_FORBIDDEN,
developer_message='Grades are frozen for this course.',
error_code='grades_frozen'
)
course = get_course_by_id(course_key, depth=None)
result = []
for user_data in request.data:
requested_user_id = user_data['user_id']
requested_usage_id = user_data['usage_id']
try:
user = self._get_single_user(request, course_key, requested_user_id)
usage_key = UsageKey.from_string(requested_usage_id)
except (USER_MODEL.DoesNotExist, InvalidKeyError, CourseEnrollment.DoesNotExist) as exc:
result.append(GradebookUpdateResponseItem(
user_id=requested_user_id,
usage_id=requested_usage_id,
success=False,
reason=text_type(exc)
))
continue
try:
subsection_grade_model = PersistentSubsectionGrade.objects.get(
user_id=user.id,
course_id=course_key,
usage_key=usage_key
)
except PersistentSubsectionGrade.DoesNotExist:
subsection = course.get_child(usage_key)
if subsection:
subsection_grade_model = self._create_subsection_grade(user, course, subsection)
else:
result.append(GradebookUpdateResponseItem(
user_id=requested_user_id,
usage_id=requested_usage_id,
success=False,
reason='usage_key {} does not exist in this course.'.format(usage_key)
))
continue
if subsection_grade_model:
self._create_override(request.user, subsection_grade_model, **user_data['grade'])
result.append(GradebookUpdateResponseItem(
user_id=user.id,
usage_id=text_type(usage_key),
success=True,
reason=None
))
status_code = status.HTTP_422_UNPROCESSABLE_ENTITY
if all((item.success for item in result)):
status_code = status.HTTP_202_ACCEPTED
return Response(
[item._asdict() for item in result],
status=status_code,
content_type='application/json'
)
def _create_subsection_grade(self, user, course, subsection):
course_data = CourseData(user, course=course)
subsection_grade = CreateSubsectionGrade(subsection, course_data.structure, {}, {})
return subsection_grade.update_or_create_model(user, force_update_subsections=True)
def _create_override(self, request_user, subsection_grade_model, **override_data):
"""
Helper method to create a `PersistentSubsectionGradeOverride` object
and send a `SUBSECTION_OVERRIDE_CHANGED` signal.
"""
override, _ = PersistentSubsectionGradeOverride.objects.update_or_create(
grade=subsection_grade_model,
defaults=self._clean_override_data(override_data),
)
_ = PersistentSubsectionGradeOverrideHistory.objects.create(
override_id=override.id,
user=request_user,
feature=PersistentSubsectionGradeOverrideHistory.GRADEBOOK,
action=PersistentSubsectionGradeOverrideHistory.CREATE_OR_UPDATE,
)
set_event_transaction_type(SUBSECTION_GRADE_CALCULATED)
create_new_event_transaction_id()
recalculate_subsection_grade_v3.apply(
kwargs=dict(
user_id=subsection_grade_model.user_id,
anonymous_user_id=None,
course_id=text_type(subsection_grade_model.course_id),
usage_id=text_type(subsection_grade_model.usage_key),
only_if_higher=False,
expected_modified_time=to_timestamp(override.modified),
score_deleted=False,
event_transaction_id=unicode(get_event_transaction_id()),
event_transaction_type=unicode(get_event_transaction_type()),
score_db_table=ScoreDatabaseTableEnum.overrides,
force_update_subsections=True,
)
)
# Emit events to let our tracking system to know we updated subsection grade
subsection_grade_calculated(subsection_grade_model)
def _clean_override_data(self, override_data):
"""
Helper method to strip any grade override field names that won't work
as defaults when calling PersistentSubsectionGradeOverride.update_or_create().
"""
allowed_fields = {
'earned_all_override',
'possible_all_override',
'earned_graded_override',
'possible_graded_override',
}
stripped_data = {}
for field in override_data.keys():
if field in allowed_fields:
stripped_data[field] = override_data[field]
return stripped_data

121
lms/djangoapps/grades/api/v1/tests/mixins.py Normal file
View File

@@ -0,0 +1,121 @@
"""
Mixins classes being used by all test classes within this folder
"""
from datetime import datetime
from pytz import UTC
from lms.djangoapps.courseware.tests.factories import GlobalStaffFactory
from openedx.core.djangoapps.content.course_overviews.tests.factories import CourseOverviewFactory
from student.tests.factories import CourseEnrollmentFactory, UserFactory
from xmodule.modulestore.tests.factories import CourseFactory, ItemFactory
from xmodule.modulestore.tests.django_utils import SharedModuleStoreTestCase, TEST_DATA_SPLIT_MODULESTORE
class GradeViewTestMixin(SharedModuleStoreTestCase):
"""
Mixin class for grades related view tests
The following tests assume that the grading policy is the edX default one:
{
"GRADER": [
{
"drop_count": 2,
"min_count": 12,
"short_label": "HW",
"type": "Homework",
"weight": 0.15
},
{
"drop_count": 2,
"min_count": 12,
"type": "Lab",
"weight": 0.15
},
{
"drop_count": 0,
"min_count": 1,
"short_label": "Midterm",
"type": "Midterm Exam",
"weight": 0.3
},
{
"drop_count": 0,
"min_count": 1,
"short_label": "Final",
"type": "Final Exam",
"weight": 0.4
}
],
"GRADE_CUTOFFS": {
"Pass": 0.5
}
}
"""
MODULESTORE = TEST_DATA_SPLIT_MODULESTORE
@classmethod
def setUpClass(cls):
super(GradeViewTestMixin, cls).setUpClass()
cls.course = cls._create_test_course_with_default_grading_policy(
display_name='test course', run="Testing_course"
)
cls.empty_course = cls._create_test_course_with_default_grading_policy(
display_name='empty test course', run="Empty_testing_course"
)
cls.course_key = cls.course.id
def _create_user_enrollments(self, *users):
date = datetime(2013, 1, 22, tzinfo=UTC)
for user in users:
CourseEnrollmentFactory(
course_id=self.course.id,
user=user,
created=date,
)
def setUp(self):
super(GradeViewTestMixin, self).setUp()
self.password = 'test'
self.global_staff = GlobalStaffFactory.create()
self.student = UserFactory(password=self.password, username='student')
self.other_student = UserFactory(password=self.password, username='other_student')
self._create_user_enrollments(self.student, self.other_student)
@classmethod
def _create_test_course_with_default_grading_policy(cls, display_name, run):
"""
Utility method to create a course with a default grading policy
"""
course = CourseFactory.create(display_name=display_name, run=run)
_ = CourseOverviewFactory.create(id=course.id)
chapter = ItemFactory.create(
category='chapter',
parent_location=course.location,
display_name="Chapter 1",
)
# create a problem for each type and minimum count needed by the grading policy
# A section is not considered if the student answers less than "min_count" problems
for grading_type, min_count in (("Homework", 12), ("Lab", 12), ("Midterm Exam", 1), ("Final Exam", 1)):
for num in xrange(min_count):
section = ItemFactory.create(
category='sequential',
parent_location=chapter.location,
due=datetime(2017, 12, 18, 11, 30, 00),
display_name='Sequential {} {}'.format(grading_type, num),
format=grading_type,
graded=True,
)
vertical = ItemFactory.create(
category='vertical',
parent_location=section.location,
display_name='Vertical {} {}'.format(grading_type, num),
)
ItemFactory.create(
category='problem',
parent_location=vertical.location,
display_name='Problem {} {}'.format(grading_type, num),
)
return course

916
lms/djangoapps/grades/api/v1/tests/test_gradebook_views.py
View File

@@ -1,13 +1,36 @@
"""
Tests for the course grading API view
"""
from __future__ import unicode_literals
import json
from collections import OrderedDict, namedtuple
from datetime import datetime
import ddt
from django.core.urlresolvers import reverse
from mock import MagicMock, patch
from pytz import UTC
from rest_framework import status
from rest_framework.test import APITestCase
from six import text_type
from lms.djangoapps.courseware.tests.factories import StaffFactory
from student.tests.factories import UserFactory
from course_modes.models import CourseMode
from lms.djangoapps.courseware.tests.factories import InstructorFactory, StaffFactory
from lms.djangoapps.grades.api.v1.tests.mixins import GradeViewTestMixin
from lms.djangoapps.grades.api.v1.views import CourseEnrollmentPagination
from lms.djangoapps.grades.config.waffle import WRITABLE_GRADEBOOK, waffle_flags
from lms.djangoapps.grades.course_data import CourseData
from lms.djangoapps.grades.course_grade import CourseGrade
from lms.djangoapps.grades.models import (
PersistentSubsectionGrade,
PersistentSubsectionGradeOverrideHistory
)
from lms.djangoapps.grades.subsection_grade import ReadSubsectionGrade
from openedx.core.djangoapps.content.course_overviews.tests.factories import CourseOverviewFactory
from openedx.core.djangoapps.course_groups.tests.helpers import CohortFactory
from openedx.core.djangoapps.waffle_utils.testutils import override_waffle_flag
from student.tests.factories import CourseEnrollmentFactory, UserFactory
from xmodule.modulestore.tests.django_utils import TEST_DATA_SPLIT_MODULESTORE, SharedModuleStoreTestCase
from xmodule.modulestore.tests.factories import CourseFactory, ItemFactory
@@ -231,3 +254,892 @@ class CourseGradingViewTest(SharedModuleStoreTestCase, APITestCase):
]
}
self.assertEqual(expected_data, resp.data)
class GradebookViewTestBase(GradeViewTestMixin, APITestCase):
"""
Base class for the gradebook GET and POST view tests.
"""
@classmethod
def setUpClass(cls):
super(GradebookViewTestBase, cls).setUpClass()
cls.namespaced_url = 'grades_api:v1:course_gradebook'
cls.waffle_flag = waffle_flags()[WRITABLE_GRADEBOOK]
cls.course = CourseFactory.create(display_name='test-course', run='run-1')
cls.course_key = cls.course.id
cls.course_overview = CourseOverviewFactory.create(id=cls.course.id)
cls.chapter_1 = ItemFactory.create(
category='chapter',
parent_location=cls.course.location,
display_name="Chapter 1",
)
cls.chapter_2 = ItemFactory.create(
category='chapter',
parent_location=cls.course.location,
display_name="Chapter 2",
)
cls.subsections = {
cls.chapter_1.location: [
ItemFactory.create(
category='sequential',
parent_location=cls.chapter_1.location,
due=datetime(2017, 12, 18, 11, 30, 00),
display_name='HW 1',
format='Homework',
graded=True,
),
ItemFactory.create(
category='sequential',
parent_location=cls.chapter_1.location,
due=datetime(2017, 12, 18, 11, 30, 00),
display_name='Lab 1',
format='Lab',
graded=True,
),
],
cls.chapter_2.location: [
ItemFactory.create(
category='sequential',
parent_location=cls.chapter_2.location,
due=datetime(2017, 12, 18, 11, 30, 00),
display_name='HW 2',
format='Homework',
graded=True,
),
ItemFactory.create(
category='sequential',
parent_location=cls.chapter_2.location,
due=datetime(2017, 12, 18, 11, 30, 00),
display_name='Lab 2',
format='Lab',
graded=True,
),
],
}
cls.course_data = CourseData(None, course=cls.course)
# we have to force the collection of course data from the block_structure API
# so that CourseGrade.course_data objects can later have a non-null effective_structure
_ = cls.course_data.collected_structure
def get_url(self, course_key=None):
"""
Helper function to create the course gradebook API url.
"""
return reverse(
self.namespaced_url,
kwargs={
'course_id': course_key or self.course_key,
}
)
def login_staff(self):
"""
Helper function to login the global staff user, who has permissions to read from the
Gradebook API.
"""
self.client.login(username=self.global_staff.username, password=self.password)
return self.global_staff
def login_course_staff(self):
"""
Helper function to login the course staff user, who has permissions to read from the
Gradebook API.
"""
course_staff = StaffFactory.create(course_key=self.course_key)
self._create_user_enrollments(course_staff)
self.client.login(username=course_staff.username, password=self.password)
return course_staff
def login_course_admin(self):
"""
Helper function to login the course admin user, who has permissions to read from the
Gradebook API.
"""
course_admin = InstructorFactory.create(course_key=self.course_key)
self._create_user_enrollments(course_admin)
self.client.login(username=course_admin.username, password=self.password)
return course_admin
@ddt.ddt
class GradebookViewTest(GradebookViewTestBase):
"""
Tests for the gradebook view.
"""
@classmethod
def setUpClass(cls):
super(GradebookViewTest, cls).setUpClass()
cls.mock_subsection_grades = {
cls.subsections[cls.chapter_1.location][0].location: cls.mock_subsection_grade(
cls.subsections[cls.chapter_1.location][0],
earned_all=1.0,
possible_all=2.0,
earned_graded=1.0,
possible_graded=2.0,
),
cls.subsections[cls.chapter_1.location][1].location: cls.mock_subsection_grade(
cls.subsections[cls.chapter_1.location][1],
earned_all=1.0,
possible_all=2.0,
earned_graded=1.0,
possible_graded=2.0,
),
cls.subsections[cls.chapter_2.location][0].location: cls.mock_subsection_grade(
cls.subsections[cls.chapter_2.location][0],
earned_all=1.0,
possible_all=2.0,
earned_graded=1.0,
possible_graded=2.0,
),
cls.subsections[cls.chapter_2.location][1].location: cls.mock_subsection_grade(
cls.subsections[cls.chapter_2.location][1],
earned_all=1.0,
possible_all=2.0,
earned_graded=1.0,
possible_graded=2.0,
),
}
def get_url(self, course_key=None, username=None, username_contains=None): # pylint: disable=arguments-differ
"""
Helper function to create the course gradebook API read url.
"""
base_url = super(GradebookViewTest, self).get_url(course_key)
if username:
return "{0}?username={1}".format(base_url, username)
if username_contains:
return "{0}?username_contains={1}".format(base_url, username_contains)
return base_url
@staticmethod
def mock_subsection_grade(subsection, **kwargs):
"""
Helper function to mock a subsection grade.
"""
model = MagicMock(**kwargs)
factory = MagicMock()
return ReadSubsectionGrade(subsection, model, factory)
def mock_course_grade(self, user, **kwargs):
"""
Helper function to return a mock CourseGrade object.
"""
course_grade = CourseGrade(user=user, course_data=self.course_data, **kwargs)
course_grade.subsection_grade = lambda key: self.mock_subsection_grades[key]
return course_grade
def expected_subsection_grades(self, letter_grade=None):
"""
Helper function to generate expected subsection detail results.
"""
return [
OrderedDict([
('category', 'Homework'),
('displayed_value', '0.50'),
('is_graded', True),
('grade_description', '(1.00/2.00)'),
('label', 'HW 01'),
('letter_grade', letter_grade),
('module_id', text_type(self.subsections[self.chapter_1.location][0].location)),
('percent', 0.5),
('score_earned', 1.0),
('score_possible', 2.0),
('subsection_name', 'HW 1')
]),
OrderedDict([
('category', 'Lab'),
('displayed_value', '0.50'),
('is_graded', True),
('grade_description', '(1.00/2.00)'),
('label', 'Lab 01'),
('letter_grade', letter_grade),
('module_id', text_type(self.subsections[self.chapter_1.location][1].location)),
('percent', 0.5),
('score_earned', 1.0),
('score_possible', 2.0),
('subsection_name', 'Lab 1')
]),
OrderedDict([
('category', 'Homework'),
('displayed_value', '0.50'),
('is_graded', True),
('grade_description', '(1.00/2.00)'),
('label', 'HW 02'),
('letter_grade', letter_grade),
('module_id', text_type(self.subsections[self.chapter_2.location][0].location)),
('percent', 0.5),
('score_earned', 1.0),
('score_possible', 2.0),
('subsection_name', 'HW 2')
]),
OrderedDict([
('category', 'Lab'),
('displayed_value', '0.50'),
('is_graded', True),
('grade_description', '(1.00/2.00)'),
('label', 'Lab 02'),
('letter_grade', letter_grade),
('module_id', text_type(self.subsections[self.chapter_2.location][1].location)),
('percent', 0.5),
('score_earned', 1.0),
('score_possible', 2.0),
('subsection_name', 'Lab 2')
]),
]
def _assert_data_all_users(self, response):
"""
Helper method to assert that self.student and self.other_student
have the expected gradebook data.
"""
expected_results = [
OrderedDict([
('course_id', text_type(self.course.id)),
('email', self.student.email),
('user_id', self.student.id),
('username', self.student.username),
('full_name', self.student.get_full_name()),
('passed', True),
('percent', 0.85),
('letter_grade', 'A'),
('progress_page_url', reverse(
'student_progress',
kwargs=dict(course_id=text_type(self.course.id), student_id=self.student.id)
)),
('section_breakdown', self.expected_subsection_grades(letter_grade='A')),
]),
OrderedDict([
('course_id', text_type(self.course.id)),
('email', self.other_student.email),
('user_id', self.other_student.id),
('username', self.other_student.username),
('full_name', self.other_student.get_full_name()),
('passed', False),
('percent', 0.45),
('letter_grade', None),
('progress_page_url', reverse(
'student_progress',
kwargs=dict(course_id=text_type(self.course.id), student_id=self.other_student.id)
)),
('section_breakdown', self.expected_subsection_grades()),
]),
]
self.assertEqual(status.HTTP_200_OK, response.status_code)
actual_data = dict(response.data)
self.assertIsNone(actual_data['next'])
self.assertIsNone(actual_data['previous'])
self.assertEqual(expected_results, actual_data['results'])
def _assert_empty_response(self, response):
"""
Helper method for assertions about OK, empty responses.
"""
self.assertEqual(status.HTTP_200_OK, response.status_code)
actual_data = dict(response.data)
self.assertIsNone(actual_data['next'])
self.assertIsNone(actual_data['previous'])
self.assertEqual([], actual_data['results'])
def test_feature_not_enabled(self):
self.client.login(username=self.global_staff.username, password=self.password)
with override_waffle_flag(self.waffle_flag, active=False):
resp = self.client.get(
self.get_url(course_key=self.empty_course.id)
)
self.assertEqual(status.HTTP_403_FORBIDDEN, resp.status_code)
def test_anonymous(self):
with override_waffle_flag(self.waffle_flag, active=True):
resp = self.client.get(self.get_url())
self.assertEqual(status.HTTP_401_UNAUTHORIZED, resp.status_code)
def test_student(self):
self.client.login(username=self.student.username, password=self.password)
with override_waffle_flag(self.waffle_flag, active=True):
resp = self.client.get(self.get_url())
self.assertEqual(status.HTTP_403_FORBIDDEN, resp.status_code)
def test_course_does_not_exist(self):
with override_waffle_flag(self.waffle_flag, active=True):
self.login_staff()
resp = self.client.get(
self.get_url(course_key='course-v1:MITx+8.MechCX+2014_T1')
)
self.assertEqual(status.HTTP_404_NOT_FOUND, resp.status_code)
def test_user_does_not_exist(self):
with override_waffle_flag(self.waffle_flag, active=True):
self.login_staff()
resp = self.client.get(
self.get_url(course_key=self.course.id, username='not-a-real-user')
)
self.assertEqual(status.HTTP_404_NOT_FOUND, resp.status_code)
def test_user_not_enrolled(self):
with override_waffle_flag(self.waffle_flag, active=True):
self.login_staff()
resp = self.client.get(
self.get_url(course_key=self.empty_course.id, username=self.student.username)
)
self.assertEqual(status.HTTP_404_NOT_FOUND, resp.status_code)
def test_course_no_enrollments(self):
with override_waffle_flag(self.waffle_flag, active=True):
self.login_staff()
resp = self.client.get(
self.get_url(course_key=self.empty_course.id)
)
self._assert_empty_response(resp)
@ddt.data(
'login_staff',
'login_course_admin',
'login_course_staff',
)
def test_gradebook_data_for_course(self, login_method):
with patch('lms.djangoapps.grades.course_grade_factory.CourseGradeFactory.read') as mock_grade:
mock_grade.side_effect = [
self.mock_course_grade(self.student, passed=True, letter_grade='A', percent=0.85),
self.mock_course_grade(self.other_student, passed=False, letter_grade=None, percent=0.45),
]
with override_waffle_flag(self.waffle_flag, active=True):
getattr(self, login_method)()
resp = self.client.get(
self.get_url(course_key=self.course.id)
)
self._assert_data_all_users(resp)
@ddt.data(
'login_staff',
'login_course_admin',
'login_course_staff',
)
def test_gradebook_data_for_single_learner(self, login_method):
with patch('lms.djangoapps.grades.course_grade_factory.CourseGradeFactory.read') as mock_grade:
mock_grade.return_value = self.mock_course_grade(self.student, passed=True, letter_grade='A', percent=0.85)
with override_waffle_flag(self.waffle_flag, active=True):
getattr(self, login_method)()
resp = self.client.get(
self.get_url(course_key=self.course.id, username=self.student.username)
)
expected_results = OrderedDict([
('course_id', text_type(self.course.id)),
('email', self.student.email),
('user_id', self.student.id),
('username', self.student.username),
('full_name', self.student.get_full_name()),
('passed', True),
('percent', 0.85),
('letter_grade', 'A'),
('progress_page_url', reverse(
'student_progress',
kwargs=dict(course_id=text_type(self.course.id), student_id=self.student.id)
)),
('section_breakdown', self.expected_subsection_grades(letter_grade='A')),
])
self.assertEqual(status.HTTP_200_OK, resp.status_code)
actual_data = dict(resp.data)
self.assertEqual(expected_results, actual_data)
@ddt.data(
'login_staff',
'login_course_admin',
'login_course_staff',
)
def test_gradebook_data_filter_username_contains(self, login_method):
with patch('lms.djangoapps.grades.course_grade_factory.CourseGradeFactory.read') as mock_grade:
mock_grade.return_value = self.mock_course_grade(
self.other_student, passed=True, letter_grade='A', percent=0.85
)
with override_waffle_flag(self.waffle_flag, active=True):
getattr(self, login_method)()
resp = self.client.get(
self.get_url(course_key=self.course.id, username_contains='other')
)
expected_results = [
OrderedDict([
('course_id', text_type(self.course.id)),
('email', self.other_student.email),
('user_id', self.other_student.id),
('username', self.other_student.username),
('full_name', self.other_student.get_full_name()),
('passed', True),
('percent', 0.85),
('letter_grade', 'A'),
('progress_page_url', reverse(
'student_progress',
kwargs=dict(course_id=text_type(self.course.id), student_id=self.other_student.id)
)),
('section_breakdown', self.expected_subsection_grades(letter_grade='A')),
]),
]
self.assertEqual(status.HTTP_200_OK, resp.status_code)
actual_data = dict(resp.data)
self.assertIsNone(actual_data['next'])
self.assertIsNone(actual_data['previous'])
self.assertEqual(expected_results, actual_data['results'])
@ddt.data(
'login_staff',
'login_course_admin',
'login_course_staff',
)
def test_gradebook_data_filter_username_contains_no_match(self, login_method):
with patch('lms.djangoapps.grades.course_grade_factory.CourseGradeFactory.read') as mock_grade:
mock_grade.return_value = self.mock_course_grade(
self.other_student, passed=True, letter_grade='A', percent=0.85
)
with override_waffle_flag(self.waffle_flag, active=True):
getattr(self, login_method)()
resp = self.client.get(
self.get_url(course_key=self.course.id, username_contains='fooooooooooooooooo')
)
self._assert_empty_response(resp)
@ddt.data(
'login_staff',
'login_course_admin',
'login_course_staff',
)
def test_filter_cohort_id_and_enrollment_mode(self, login_method):
with patch('lms.djangoapps.grades.course_grade_factory.CourseGradeFactory.read') as mock_grade:
mock_grade.return_value = self.mock_course_grade(self.student, passed=True, letter_grade='A', percent=0.85)
cohort = CohortFactory(course_id=self.course.id, name="TestCohort", users=[self.student])
with override_waffle_flag(self.waffle_flag, active=True):
getattr(self, login_method)()
# both of our test users are in the audit track, so this is functionally equivalent
# to just `?cohort_id=cohort.id`.
query = '?cohort_id={}&enrollment_mode={}'.format(cohort.id, CourseMode.AUDIT)
resp = self.client.get(
self.get_url(course_key=self.course.id) + query
)
expected_results = [
OrderedDict([
('course_id', text_type(self.course.id)),
('email', self.student.email),
('user_id', self.student.id),
('username', self.student.username),
('full_name', self.student.get_full_name()),
('passed', True),
('percent', 0.85),
('letter_grade', 'A'),
('progress_page_url', reverse(
'student_progress',
kwargs=dict(course_id=text_type(self.course.id), student_id=self.student.id)
)),
('section_breakdown', self.expected_subsection_grades(letter_grade='A')),
]),
]
self.assertEqual(status.HTTP_200_OK, resp.status_code)
actual_data = dict(resp.data)
self.assertIsNone(actual_data['next'])
self.assertIsNone(actual_data['previous'])
self.assertEqual(expected_results, actual_data['results'])
@ddt.data(
'login_staff',
'login_course_admin',
'login_course_staff',
)
def test_filter_cohort_id_does_not_exist(self, login_method):
with patch('lms.djangoapps.grades.course_grade_factory.CourseGradeFactory.read') as mock_grade:
mock_grade.return_value = self.mock_course_grade(self.student, passed=True, letter_grade='A', percent=0.85)
empty_cohort = CohortFactory(course_id=self.course.id, name="TestCohort", users=[])
with override_waffle_flag(self.waffle_flag, active=True):
getattr(self, login_method)()
resp = self.client.get(
self.get_url(course_key=self.course.id) + '?cohort_id={}'.format(empty_cohort.id)
)
self._assert_empty_response(resp)
@ddt.data(
'login_staff',
'login_course_admin',
'login_course_staff',
)
def test_filter_enrollment_mode(self, login_method):
with patch('lms.djangoapps.grades.course_grade_factory.CourseGradeFactory.read') as mock_grade:
mock_grade.side_effect = [
self.mock_course_grade(self.student, passed=True, letter_grade='A', percent=0.85),
self.mock_course_grade(self.other_student, passed=False, letter_grade=None, percent=0.45),
]
# Enroll a verified student, for whom data should not be returned.
verified_student = UserFactory()
_ = CourseEnrollmentFactory(
course_id=self.course.id,
user=verified_student,
created=datetime(2013, 1, 1, tzinfo=UTC),
mode=CourseMode.VERIFIED,
)
with override_waffle_flag(self.waffle_flag, active=True):
getattr(self, login_method)()
resp = self.client.get(
self.get_url(course_key=self.course.id) + '?enrollment_mode={}'.format(CourseMode.AUDIT)
)
self._assert_data_all_users(resp)
@ddt.data(
'login_staff',
'login_course_admin',
'login_course_staff',
)
def test_filter_enrollment_mode_no_students(self, login_method):
with patch('lms.djangoapps.grades.course_grade_factory.CourseGradeFactory.read') as mock_grade:
mock_grade.side_effect = [
self.mock_course_grade(self.student, passed=True, letter_grade='A', percent=0.85),
self.mock_course_grade(self.other_student, passed=False, letter_grade=None, percent=0.45),
]
with override_waffle_flag(self.waffle_flag, active=True):
getattr(self, login_method)()
resp = self.client.get(
self.get_url(course_key=self.course.id) + '?enrollment_mode={}'.format(CourseMode.VERIFIED)
)
self._assert_empty_response(resp)
@ddt.data(None, 2, 3, 10, 60, 80)
def test_page_size_parameter(self, page_size):
user_size = 60
with patch(
'lms.djangoapps.grades.course_grade_factory.CourseGradeFactory.read'
) as mock_grade:
users = UserFactory.create_batch(user_size)
mocked_course_grades = []
for user in users:
self._create_user_enrollments(user)
mocked_course_grades.append(self.mock_course_grade(user, passed=True, letter_grade='A', percent=0.85))
mock_grade.side_effect = mocked_course_grades
with override_waffle_flag(self.waffle_flag, active=True):
self.login_staff()
query = ''
if page_size:
query = '?page_size={}'.format(page_size)
resp = self.client.get(
self.get_url(course_key=self.course.id) + query
)
self.assertEqual(status.HTTP_200_OK, resp.status_code)
actual_data = dict(resp.data)
expected_page_size = page_size or CourseEnrollmentPagination.page_size
if expected_page_size > user_size:
expected_page_size = user_size
self.assertEqual(len(actual_data['results']), expected_page_size)
@ddt.ddt
class GradebookBulkUpdateViewTest(GradebookViewTestBase):
"""
Tests for the gradebook bulk-update view.
"""
@classmethod
def setUpClass(cls):
super(GradebookBulkUpdateViewTest, cls).setUpClass()
cls.namespaced_url = 'grades_api:v1:course_gradebook_bulk_update'
def test_feature_not_enabled(self):
self.client.login(username=self.global_staff.username, password=self.password)
with override_waffle_flag(self.waffle_flag, active=False):
resp = self.client.post(
self.get_url(course_key=self.empty_course.id)
)
self.assertEqual(status.HTTP_403_FORBIDDEN, resp.status_code)
def test_anonymous(self):
with override_waffle_flag(self.waffle_flag, active=True):
resp = self.client.post(self.get_url())
self.assertEqual(status.HTTP_401_UNAUTHORIZED, resp.status_code)
def test_student(self):
self.client.login(username=self.student.username, password=self.password)
with override_waffle_flag(self.waffle_flag, active=True):
resp = self.client.post(self.get_url())
self.assertEqual(status.HTTP_403_FORBIDDEN, resp.status_code)
def test_course_does_not_exist(self):
with override_waffle_flag(self.waffle_flag, active=True):
self.login_staff()
resp = self.client.post(
self.get_url(course_key='course-v1:MITx+8.MechCX+2014_T1')
)
self.assertEqual(status.HTTP_404_NOT_FOUND, resp.status_code)
@ddt.data(
'login_staff',
'login_course_admin',
'login_course_staff',
)
def test_grades_frozen(self, login_method):
"""
Should receive a 403 when grades have been frozen for a course.
"""
with patch('lms.djangoapps.grades.api.v1.gradebook_views.are_grades_frozen', return_value=True):
with override_waffle_flag(self.waffle_flag, active=True):
getattr(self, login_method)()
post_data = [
{
'user_id': self.student.id,
'usage_id': text_type(self.subsections[self.chapter_1.location][0].location),
'grade': {}, # doesn't matter what we put here.
}
]
resp = self.client.post(
self.get_url(),
data=json.dumps(post_data),
content_type='application/json',
)
self.assertEqual(status.HTTP_403_FORBIDDEN, resp.status_code)
@ddt.data(
'login_staff',
'login_course_admin',
'login_course_staff',
)
def test_user_not_enrolled(self, login_method):
with override_waffle_flag(self.waffle_flag, active=True):
getattr(self, login_method)()
unenrolled_student = UserFactory()
post_data = [
{
'user_id': unenrolled_student.id,
'usage_id': text_type(self.subsections[self.chapter_1.location][0].location),
'grade': {}, # doesn't matter what we put here.
}
]
resp = self.client.post(
self.get_url(),
data=json.dumps(post_data),
content_type='application/json',
)
expected_data = [
{
'user_id': unenrolled_student.id,
'usage_id': text_type(self.subsections[self.chapter_1.location][0].location),
'success': False,
'reason': 'CourseEnrollment matching query does not exist.',
},
]
self.assertEqual(status.HTTP_422_UNPROCESSABLE_ENTITY, resp.status_code)
self.assertEqual(expected_data, resp.data)
@ddt.data(
'login_staff',
'login_course_admin',
'login_course_staff',
)
def test_user_does_not_exist(self, login_method):
with override_waffle_flag(self.waffle_flag, active=True):
getattr(self, login_method)()
post_data = [
{
'user_id': -123,
'usage_id': text_type(self.subsections[self.chapter_1.location][0].location),
'grade': {}, # doesn't matter what we put here.
}
]
resp = self.client.post(
self.get_url(),
data=json.dumps(post_data),
content_type='application/json',
)
expected_data = [
{
'user_id': -123,
'usage_id': text_type(self.subsections[self.chapter_1.location][0].location),
'success': False,
'reason': 'User matching query does not exist.',
},
]
self.assertEqual(status.HTTP_422_UNPROCESSABLE_ENTITY, resp.status_code)
self.assertEqual(expected_data, resp.data)
@ddt.data(
'login_staff',
'login_course_admin',
'login_course_staff',
)
def test_invalid_usage_key(self, login_method):
with override_waffle_flag(self.waffle_flag, active=True):
getattr(self, login_method)()
post_data = [
{
'user_id': self.student.id,
'usage_id': 'not-a-valid-usage-key',
'grade': {}, # doesn't matter what we put here.
}
]
resp = self.client.post(
self.get_url(),
data=json.dumps(post_data),
content_type='application/json',
)
expected_data = [
{
'user_id': self.student.id,
'usage_id': 'not-a-valid-usage-key',
'success': False,
'reason': "<class 'opaque_keys.edx.locator.BlockUsageLocator'>: not-a-valid-usage-key",
},
]
self.assertEqual(status.HTTP_422_UNPROCESSABLE_ENTITY, resp.status_code)
self.assertEqual(expected_data, resp.data)
@ddt.data(
'login_staff',
'login_course_admin',
'login_course_staff',
)
def test_subsection_does_not_exist(self, login_method):
"""
When trying to override a grade for a valid usage key that does not exist in the requested course,
we should get an error reason specifying that the key does not exist in the course.
"""
with override_waffle_flag(self.waffle_flag, active=True):
getattr(self, login_method)()
usage_id = 'block-v1:edX+DemoX+Demo_Course+type@sequential+block@workflow'
post_data = [
{
'user_id': self.student.id,
'usage_id': usage_id,
'grade': {}, # doesn't matter what we put here.
}
]
resp = self.client.post(
self.get_url(),
data=json.dumps(post_data),
content_type='application/json',
)
expected_data = [
{
'user_id': self.student.id,
'usage_id': usage_id,
'success': False,
'reason': 'usage_key {} does not exist in this course.'.format(usage_id),
},
]
self.assertEqual(status.HTTP_422_UNPROCESSABLE_ENTITY, resp.status_code)
self.assertEqual(expected_data, resp.data)
@ddt.data('login_staff', 'login_course_staff', 'login_course_admin')
def test_override_is_created(self, login_method):
"""
Test that when we make multiple requests to update grades for the same user/subsection,
the score from the most recent request is recorded.
"""
with override_waffle_flag(self.waffle_flag, active=True):
request_user = getattr(self, login_method)()
post_data = [
{
'user_id': self.student.id,
'usage_id': text_type(self.subsections[self.chapter_1.location][0].location),
'grade': {
'earned_all_override': 3,
'possible_all_override': 3,
'earned_graded_override': 2,
'possible_graded_override': 2,
},
},
{
'user_id': self.student.id,
'usage_id': text_type(self.subsections[self.chapter_1.location][1].location),
'grade': {
'earned_all_override': 1,
'possible_all_override': 4,
'earned_graded_override': 1,
'possible_graded_override': 4,
},
}
]
resp = self.client.post(
self.get_url(),
data=json.dumps(post_data),
content_type='application/json',
)
expected_data = [
{
'user_id': self.student.id,
'usage_id': text_type(self.subsections[self.chapter_1.location][0].location),
'success': True,
'reason': None,
},
{
'user_id': self.student.id,
'usage_id': text_type(self.subsections[self.chapter_1.location][1].location),
'success': True,
'reason': None,
},
]
self.assertEqual(status.HTTP_202_ACCEPTED, resp.status_code)
self.assertEqual(expected_data, resp.data)
second_post_data = [
{
'user_id': self.student.id,
'usage_id': text_type(self.subsections[self.chapter_1.location][1].location),
'grade': {
'earned_all_override': 3,
'possible_all_override': 4,
'earned_graded_override': 3,
'possible_graded_override': 4,
},
},
]
self.client.post(
self.get_url(),
data=json.dumps(second_post_data),
content_type='application/json',
)
GradeFields = namedtuple('GradeFields', ['earned_all', 'possible_all', 'earned_graded', 'possible_graded'])
# We should now have PersistentSubsectionGradeOverride records corresponding to
# our bulk-update request, and PersistentSubsectionGrade records with grade values
# equal to those of the override.
for usage_key, expected_grades in (
(self.subsections[self.chapter_1.location][0].location, GradeFields(3, 3, 2, 2)),
(self.subsections[self.chapter_1.location][1].location, GradeFields(3, 4, 3, 4)),
):
# this selects related PersistentSubsectionGradeOverride objects.
grade = PersistentSubsectionGrade.read_grade(
user_id=self.student.id,
usage_key=usage_key,
)
for field_name in expected_grades._fields:
expected_value = getattr(expected_grades, field_name)
self.assertEqual(expected_value, getattr(grade, field_name))
self.assertEqual(expected_value, getattr(grade.override, field_name + '_override'))
update_records = PersistentSubsectionGradeOverrideHistory.objects.filter(user=request_user)
self.assertEqual(update_records.count(), 3)
for audit_item in update_records:
self.assertEqual(audit_item.user, request_user)
self.assertIsNotNone(audit_item.created)
self.assertEqual(audit_item.feature, PersistentSubsectionGradeOverrideHistory.GRADEBOOK)
self.assertEqual(audit_item.action, PersistentSubsectionGradeOverrideHistory.CREATE_OR_UPDATE)

933
lms/djangoapps/grades/api/v1/tests/test_views.py
View File

@@ -2,143 +2,19 @@
Tests for v1 views
"""
from __future__ import unicode_literals
import json
from collections import OrderedDict, namedtuple
from datetime import datetime
from collections import OrderedDict
import ddt
from django.urls import reverse
from mock import MagicMock, patch
from opaque_keys import InvalidKeyError
from pytz import UTC
from rest_framework import status
from rest_framework.test import APITestCase
from six import text_type
from course_modes.models import CourseMode
from lms.djangoapps.courseware.tests.factories import GlobalStaffFactory
from lms.djangoapps.grades.api.v1.views import CourseGradesView, CourseEnrollmentPagination
from lms.djangoapps.grades.config.waffle import waffle_flags, WRITABLE_GRADEBOOK
from lms.djangoapps.grades.course_data import CourseData
from lms.djangoapps.grades.course_grade import CourseGrade
from lms.djangoapps.grades.models import PersistentSubsectionGrade, PersistentSubsectionGradeOverrideHistory
from lms.djangoapps.grades.subsection_grade import ReadSubsectionGrade
from openedx.core.djangoapps.content.course_overviews.tests.factories import CourseOverviewFactory
from openedx.core.djangoapps.course_groups.tests.helpers import CohortFactory
from lms.djangoapps.grades.api.v1.tests.mixins import GradeViewTestMixin
from lms.djangoapps.grades.api.v1.views import CourseGradesView
from openedx.core.djangoapps.user_authn.tests.utils import AuthAndScopesTestMixin
from openedx.core.djangoapps.waffle_utils.testutils import override_waffle_flag
from student.tests.factories import CourseEnrollmentFactory, UserFactory
from xmodule.modulestore.tests.factories import CourseFactory, ItemFactory
from xmodule.modulestore.tests.django_utils import SharedModuleStoreTestCase, TEST_DATA_SPLIT_MODULESTORE
class GradeViewTestMixin(SharedModuleStoreTestCase):
"""
Mixin class for grades related view tests
The following tests assume that the grading policy is the edX default one:
{
"GRADER": [
{
"drop_count": 2,
"min_count": 12,
"short_label": "HW",
"type": "Homework",
"weight": 0.15
},
{
"drop_count": 2,
"min_count": 12,
"type": "Lab",
"weight": 0.15
},
{
"drop_count": 0,
"min_count": 1,
"short_label": "Midterm",
"type": "Midterm Exam",
"weight": 0.3
},
{
"drop_count": 0,
"min_count": 1,
"short_label": "Final",
"type": "Final Exam",
"weight": 0.4
}
],
"GRADE_CUTOFFS": {
"Pass": 0.5
}
}
"""
MODULESTORE = TEST_DATA_SPLIT_MODULESTORE
@classmethod
def setUpClass(cls):
super(GradeViewTestMixin, cls).setUpClass()
cls.course = cls._create_test_course_with_default_grading_policy(
display_name='test course', run="Testing_course"
)
cls.empty_course = cls._create_test_course_with_default_grading_policy(
display_name='empty test course', run="Empty_testing_course"
)
cls.course_key = cls.course.id
def _create_user_enrollments(self, *users):
date = datetime(2013, 1, 22, tzinfo=UTC)
for user in users:
CourseEnrollmentFactory(
course_id=self.course.id,
user=user,
created=date,
)
def setUp(self):
super(GradeViewTestMixin, self).setUp()
self.password = 'test'
self.global_staff = GlobalStaffFactory.create()
self.student = UserFactory(password=self.password, username='student')
self.other_student = UserFactory(password=self.password, username='other_student')
self._create_user_enrollments(self.student, self.other_student)
@classmethod
def _create_test_course_with_default_grading_policy(cls, display_name, run):
"""
Utility method to create a course with a default grading policy
"""
course = CourseFactory.create(display_name=display_name, run=run)
_ = CourseOverviewFactory.create(id=course.id)
chapter = ItemFactory.create(
category='chapter',
parent_location=course.location,
display_name="Chapter 1",
)
# create a problem for each type and minimum count needed by the grading policy
# A section is not considered if the student answers less than "min_count" problems
for grading_type, min_count in (("Homework", 12), ("Lab", 12), ("Midterm Exam", 1), ("Final Exam", 1)):
for num in xrange(min_count):
section = ItemFactory.create(
category='sequential',
parent_location=chapter.location,
due=datetime(2017, 12, 18, 11, 30, 00),
display_name='Sequential {} {}'.format(grading_type, num),
format=grading_type,
graded=True,
)
vertical = ItemFactory.create(
category='vertical',
parent_location=section.location,
display_name='Vertical {} {}'.format(grading_type, num),
)
ItemFactory.create(
category='problem',
parent_location=vertical.location,
display_name='Problem {} {}'.format(grading_type, num),
)
return course
from student.tests.factories import UserFactory
@ddt.ddt
@@ -373,804 +249,3 @@ class CourseGradesViewTest(GradeViewTestMixin, APITestCase):
])
self.assertEqual(expected_data, resp.data)
class GradebookViewTestBase(GradeViewTestMixin, APITestCase):
"""
Base class for the gradebook GET and POST view tests.
"""
@classmethod
def setUpClass(cls):
super(GradebookViewTestBase, cls).setUpClass()
cls.namespaced_url = 'grades_api:v1:course_gradebook'
cls.waffle_flag = waffle_flags()[WRITABLE_GRADEBOOK]
cls.course = CourseFactory.create(display_name='test-course', run='run-1')
cls.course_key = cls.course.id
cls.course_overview = CourseOverviewFactory.create(id=cls.course.id)
cls.chapter_1 = ItemFactory.create(
category='chapter',
parent_location=cls.course.location,
display_name="Chapter 1",
)
cls.chapter_2 = ItemFactory.create(
category='chapter',
parent_location=cls.course.location,
display_name="Chapter 2",
)
cls.subsections = {
cls.chapter_1.location: [
ItemFactory.create(
category='sequential',
parent_location=cls.chapter_1.location,
due=datetime(2017, 12, 18, 11, 30, 00),
display_name='HW 1',
format='Homework',
graded=True,
),
ItemFactory.create(
category='sequential',
parent_location=cls.chapter_1.location,
due=datetime(2017, 12, 18, 11, 30, 00),
display_name='Lab 1',
format='Lab',
graded=True,
),
],
cls.chapter_2.location: [
ItemFactory.create(
category='sequential',
parent_location=cls.chapter_2.location,
due=datetime(2017, 12, 18, 11, 30, 00),
display_name='HW 2',
format='Homework',
graded=True,
),
ItemFactory.create(
category='sequential',
parent_location=cls.chapter_2.location,
due=datetime(2017, 12, 18, 11, 30, 00),
display_name='Lab 2',
format='Lab',
graded=True,
),
],
}
cls.course_data = CourseData(None, course=cls.course)
# we have to force the collection of course data from the block_structure API
# so that CourseGrade.course_data objects can later have a non-null effective_structure
_ = cls.course_data.collected_structure
def get_url(self, course_key=None):
"""
Helper function to create the course gradebook API url.
"""
return reverse(
self.namespaced_url,
kwargs={
'course_id': course_key or self.course_key,
}
)
def login_staff(self):
"""
Helper function to login the global staff user, who has permissions to read from the
Gradebook API.
"""
self.client.login(username=self.global_staff.username, password=self.password)
@ddt.ddt
class GradebookViewTest(GradebookViewTestBase):
"""
Tests for the gradebook view.
"""
@classmethod
def setUpClass(cls):
super(GradebookViewTest, cls).setUpClass()
cls.mock_subsection_grades = {
cls.subsections[cls.chapter_1.location][0].location: cls.mock_subsection_grade(
cls.subsections[cls.chapter_1.location][0],
earned_all=1.0,
possible_all=2.0,
earned_graded=1.0,
possible_graded=2.0,
),
cls.subsections[cls.chapter_1.location][1].location: cls.mock_subsection_grade(
cls.subsections[cls.chapter_1.location][1],
earned_all=1.0,
possible_all=2.0,
earned_graded=1.0,
possible_graded=2.0,
),
cls.subsections[cls.chapter_2.location][0].location: cls.mock_subsection_grade(
cls.subsections[cls.chapter_2.location][0],
earned_all=1.0,
possible_all=2.0,
earned_graded=1.0,
possible_graded=2.0,
),
cls.subsections[cls.chapter_2.location][1].location: cls.mock_subsection_grade(
cls.subsections[cls.chapter_2.location][1],
earned_all=1.0,
possible_all=2.0,
earned_graded=1.0,
possible_graded=2.0,
),
}
def get_url(self, course_key=None, username=None, username_contains=None): # pylint: disable=arguments-differ
"""
Helper function to create the course gradebook API read url.
"""
base_url = super(GradebookViewTest, self).get_url(course_key)
if username:
return "{0}?username={1}".format(base_url, username)
if username_contains:
return "{0}?username_contains={1}".format(base_url, username_contains)
return base_url
@staticmethod
def mock_subsection_grade(subsection, **kwargs):
"""
Helper function to mock a subsection grade.
"""
model = MagicMock(**kwargs)
factory = MagicMock()
return ReadSubsectionGrade(subsection, model, factory)
def mock_course_grade(self, user, **kwargs):
"""
Helper function to return a mock CourseGrade object.
"""
course_grade = CourseGrade(user=user, course_data=self.course_data, **kwargs)
course_grade.subsection_grade = lambda key: self.mock_subsection_grades[key]
return course_grade
def expected_subsection_grades(self, letter_grade=None):
"""
Helper function to generate expected subsection detail results.
"""
return [
OrderedDict([
('category', 'Homework'),
('displayed_value', '0.50'),
('is_graded', True),
('grade_description', '(1.00/2.00)'),
('label', 'HW 01'),
('letter_grade', letter_grade),
('module_id', text_type(self.subsections[self.chapter_1.location][0].location)),
('percent', 0.5),
('score_earned', 1.0),
('score_possible', 2.0),
('subsection_name', 'HW 1')
]),
OrderedDict([
('category', 'Lab'),
('displayed_value', '0.50'),
('is_graded', True),
('grade_description', '(1.00/2.00)'),
('label', 'Lab 01'),
('letter_grade', letter_grade),
('module_id', text_type(self.subsections[self.chapter_1.location][1].location)),
('percent', 0.5),
('score_earned', 1.0),
('score_possible', 2.0),
('subsection_name', 'Lab 1')
]),
OrderedDict([
('category', 'Homework'),
('displayed_value', '0.50'),
('is_graded', True),
('grade_description', '(1.00/2.00)'),
('label', 'HW 02'),
('letter_grade', letter_grade),
('module_id', text_type(self.subsections[self.chapter_2.location][0].location)),
('percent', 0.5),
('score_earned', 1.0),
('score_possible', 2.0),
('subsection_name', 'HW 2')
]),
OrderedDict([
('category', 'Lab'),
('displayed_value', '0.50'),
('is_graded', True),
('grade_description', '(1.00/2.00)'),
('label', 'Lab 02'),
('letter_grade', letter_grade),
('module_id', text_type(self.subsections[self.chapter_2.location][1].location)),
('percent', 0.5),
('score_earned', 1.0),
('score_possible', 2.0),
('subsection_name', 'Lab 2')
]),
]
def _assert_data_all_users(self, response):
"""
Helper method to assert that self.student and self.other_student
have the expected gradebook data.
"""
expected_results = [
OrderedDict([
('course_id', text_type(self.course.id)),
('email', self.student.email),
('user_id', self.student.id),
('username', self.student.username),
('full_name', self.student.get_full_name()),
('passed', True),
('percent', 0.85),
('letter_grade', 'A'),
('progress_page_url', reverse(
'student_progress',
kwargs=dict(course_id=text_type(self.course.id), student_id=self.student.id)
)),
('section_breakdown', self.expected_subsection_grades(letter_grade='A')),
]),
OrderedDict([
('course_id', text_type(self.course.id)),
('email', self.other_student.email),
('user_id', self.other_student.id),
('username', self.other_student.username),
('full_name', self.other_student.get_full_name()),
('passed', False),
('percent', 0.45),
('letter_grade', None),
('progress_page_url', reverse(
'student_progress',
kwargs=dict(course_id=text_type(self.course.id), student_id=self.other_student.id)
)),
('section_breakdown', self.expected_subsection_grades()),
]),
]
self.assertEqual(status.HTTP_200_OK, response.status_code)
actual_data = dict(response.data)
self.assertIsNone(actual_data['next'])
self.assertIsNone(actual_data['previous'])
self.assertEqual(expected_results, actual_data['results'])
def _assert_empty_response(self, response):
"""
Helper method for assertions about OK, empty responses.
"""
self.assertEqual(status.HTTP_200_OK, response.status_code)
actual_data = dict(response.data)
self.assertIsNone(actual_data['next'])
self.assertIsNone(actual_data['previous'])
self.assertEqual([], actual_data['results'])
def test_feature_not_enabled(self):
self.client.login(username=self.global_staff.username, password=self.password)
with override_waffle_flag(self.waffle_flag, active=False):
resp = self.client.get(
self.get_url(course_key=self.empty_course.id)
)
self.assertEqual(status.HTTP_403_FORBIDDEN, resp.status_code)
def test_anonymous(self):
with override_waffle_flag(self.waffle_flag, active=True):
resp = self.client.get(self.get_url())
self.assertEqual(status.HTTP_401_UNAUTHORIZED, resp.status_code)
def test_student(self):
self.client.login(username=self.student.username, password=self.password)
with override_waffle_flag(self.waffle_flag, active=True):
resp = self.client.get(self.get_url())
self.assertEqual(status.HTTP_403_FORBIDDEN, resp.status_code)
def test_course_does_not_exist(self):
with override_waffle_flag(self.waffle_flag, active=True):
self.login_staff()
resp = self.client.get(
self.get_url(course_key='course-v1:MITx+8.MechCX+2014_T1')
)
self.assertEqual(status.HTTP_404_NOT_FOUND, resp.status_code)
def test_user_does_not_exist(self):
with override_waffle_flag(self.waffle_flag, active=True):
self.login_staff()
resp = self.client.get(
self.get_url(course_key=self.course.id, username='not-a-real-user')
)
self.assertEqual(status.HTTP_404_NOT_FOUND, resp.status_code)
def test_user_not_enrolled(self):
with override_waffle_flag(self.waffle_flag, active=True):
self.login_staff()
resp = self.client.get(
self.get_url(course_key=self.empty_course.id, username=self.student.username)
)
self.assertEqual(status.HTTP_404_NOT_FOUND, resp.status_code)
def test_course_no_enrollments(self):
with override_waffle_flag(self.waffle_flag, active=True):
self.login_staff()
resp = self.client.get(
self.get_url(course_key=self.empty_course.id)
)
self._assert_empty_response(resp)
def test_gradebook_data_for_course(self):
with patch('lms.djangoapps.grades.course_grade_factory.CourseGradeFactory.read') as mock_grade:
mock_grade.side_effect = [
self.mock_course_grade(self.student, passed=True, letter_grade='A', percent=0.85),
self.mock_course_grade(self.other_student, passed=False, letter_grade=None, percent=0.45),
]
with override_waffle_flag(self.waffle_flag, active=True):
self.login_staff()
resp = self.client.get(
self.get_url(course_key=self.course.id)
)
self._assert_data_all_users(resp)
def test_gradebook_data_for_single_learner(self):
with patch('lms.djangoapps.grades.course_grade_factory.CourseGradeFactory.read') as mock_grade:
mock_grade.return_value = self.mock_course_grade(self.student, passed=True, letter_grade='A', percent=0.85)
with override_waffle_flag(self.waffle_flag, active=True):
self.login_staff()
resp = self.client.get(
self.get_url(course_key=self.course.id, username=self.student.username)
)
expected_results = OrderedDict([
('course_id', text_type(self.course.id)),
('email', self.student.email),
('user_id', self.student.id),
('username', self.student.username),
('full_name', self.student.get_full_name()),
('passed', True),
('percent', 0.85),
('letter_grade', 'A'),
('progress_page_url', reverse(
'student_progress',
kwargs=dict(course_id=text_type(self.course.id), student_id=self.student.id)
)),
('section_breakdown', self.expected_subsection_grades(letter_grade='A')),
])
self.assertEqual(status.HTTP_200_OK, resp.status_code)
actual_data = dict(resp.data)
self.assertEqual(expected_results, actual_data)
def test_gradebook_data_filter_username_contains(self):
with patch('lms.djangoapps.grades.course_grade_factory.CourseGradeFactory.read') as mock_grade:
mock_grade.return_value = self.mock_course_grade(
self.other_student, passed=True, letter_grade='A', percent=0.85
)
with override_waffle_flag(self.waffle_flag, active=True):
self.login_staff()
resp = self.client.get(
self.get_url(course_key=self.course.id, username_contains='other')
)
expected_results = [
OrderedDict([
('course_id', text_type(self.course.id)),
('email', self.other_student.email),
('user_id', self.other_student.id),
('username', self.other_student.username),
('full_name', self.other_student.get_full_name()),
('passed', True),
('percent', 0.85),
('letter_grade', 'A'),
('progress_page_url', reverse(
'student_progress',
kwargs=dict(course_id=text_type(self.course.id), student_id=self.other_student.id)
)),
('section_breakdown', self.expected_subsection_grades(letter_grade='A')),
]),
]
self.assertEqual(status.HTTP_200_OK, resp.status_code)
actual_data = dict(resp.data)
self.assertIsNone(actual_data['next'])
self.assertIsNone(actual_data['previous'])
self.assertEqual(expected_results, actual_data['results'])
def test_gradebook_data_filter_username_contains_no_match(self):
with patch('lms.djangoapps.grades.course_grade_factory.CourseGradeFactory.read') as mock_grade:
mock_grade.return_value = self.mock_course_grade(
self.other_student, passed=True, letter_grade='A', percent=0.85
)
with override_waffle_flag(self.waffle_flag, active=True):
self.login_staff()
resp = self.client.get(
self.get_url(course_key=self.course.id, username_contains='fooooooooooooooooo')
)
self._assert_empty_response(resp)
def test_filter_cohort_id_and_enrollment_mode(self):
with patch('lms.djangoapps.grades.course_grade_factory.CourseGradeFactory.read') as mock_grade:
mock_grade.return_value = self.mock_course_grade(self.student, passed=True, letter_grade='A', percent=0.85)
cohort = CohortFactory(course_id=self.course.id, name="TestCohort", users=[self.student])
with override_waffle_flag(self.waffle_flag, active=True):
self.login_staff()
# both of our test users are in the audit track, so this is functionally equivalent
# to just `?cohort_id=cohort.id`.
query = '?cohort_id={}&enrollment_mode={}'.format(cohort.id, CourseMode.AUDIT)
resp = self.client.get(
self.get_url(course_key=self.course.id) + query
)
expected_results = [
OrderedDict([
('course_id', text_type(self.course.id)),
('email', self.student.email),
('user_id', self.student.id),
('username', self.student.username),
('full_name', self.student.get_full_name()),
('passed', True),
('percent', 0.85),
('letter_grade', 'A'),
('progress_page_url', reverse(
'student_progress',
kwargs=dict(course_id=text_type(self.course.id), student_id=self.student.id)
)),
('section_breakdown', self.expected_subsection_grades(letter_grade='A')),
]),
]
self.assertEqual(status.HTTP_200_OK, resp.status_code)
actual_data = dict(resp.data)
self.assertIsNone(actual_data['next'])
self.assertIsNone(actual_data['previous'])
self.assertEqual(expected_results, actual_data['results'])
def test_filter_cohort_id_does_not_exist(self):
with patch('lms.djangoapps.grades.course_grade_factory.CourseGradeFactory.read') as mock_grade:
mock_grade.return_value = self.mock_course_grade(self.student, passed=True, letter_grade='A', percent=0.85)
empty_cohort = CohortFactory(course_id=self.course.id, name="TestCohort", users=[])
with override_waffle_flag(self.waffle_flag, active=True):
self.login_staff()
resp = self.client.get(
self.get_url(course_key=self.course.id) + '?cohort_id={}'.format(empty_cohort.id)
)
self._assert_empty_response(resp)
def test_filter_enrollment_mode(self):
with patch('lms.djangoapps.grades.course_grade_factory.CourseGradeFactory.read') as mock_grade:
mock_grade.side_effect = [
self.mock_course_grade(self.student, passed=True, letter_grade='A', percent=0.85),
self.mock_course_grade(self.other_student, passed=False, letter_grade=None, percent=0.45),
]
# Enroll a verified student, for whom data should not be returned.
verified_student = UserFactory()
_ = CourseEnrollmentFactory(
course_id=self.course.id,
user=verified_student,
created=datetime(2013, 1, 1, tzinfo=UTC),
mode=CourseMode.VERIFIED,
)
with override_waffle_flag(self.waffle_flag, active=True):
self.login_staff()
resp = self.client.get(
self.get_url(course_key=self.course.id) + '?enrollment_mode={}'.format(CourseMode.AUDIT)
)
self._assert_data_all_users(resp)
def test_filter_enrollment_mode_no_students(self):
with patch('lms.djangoapps.grades.course_grade_factory.CourseGradeFactory.read') as mock_grade:
mock_grade.side_effect = [
self.mock_course_grade(self.student, passed=True, letter_grade='A', percent=0.85),
self.mock_course_grade(self.other_student, passed=False, letter_grade=None, percent=0.45),
]
with override_waffle_flag(self.waffle_flag, active=True):
self.login_staff()
resp = self.client.get(
self.get_url(course_key=self.course.id) + '?enrollment_mode={}'.format(CourseMode.VERIFIED)
)
self._assert_empty_response(resp)
@ddt.data(None, 2, 3, 10, 60, 80)
def test_page_size_parameter(self, page_size):
user_size = 60
with patch(
'lms.djangoapps.grades.course_grade_factory.CourseGradeFactory.read'
) as mock_grade:
users = UserFactory.create_batch(user_size)
mocked_course_grades = []
for user in users:
self._create_user_enrollments(user)
mocked_course_grades.append(self.mock_course_grade(user, passed=True, letter_grade='A', percent=0.85))
mock_grade.side_effect = mocked_course_grades
with override_waffle_flag(self.waffle_flag, active=True):
self.login_staff()
query = ''
if page_size:
query = '?page_size={}'.format(page_size)
resp = self.client.get(
self.get_url(course_key=self.course.id) + query
)
self.assertEqual(status.HTTP_200_OK, resp.status_code)
actual_data = dict(resp.data)
expected_page_size = page_size or CourseEnrollmentPagination.page_size
if expected_page_size > user_size:
expected_page_size = user_size
self.assertEqual(len(actual_data['results']), expected_page_size)
class GradebookBulkUpdateViewTest(GradebookViewTestBase):
"""
Tests for the gradebook bulk-update view.
"""
@classmethod
def setUpClass(cls):
super(GradebookBulkUpdateViewTest, cls).setUpClass()
cls.namespaced_url = 'grades_api:v1:course_gradebook_bulk_update'
def test_feature_not_enabled(self):
self.client.login(username=self.global_staff.username, password=self.password)
with override_waffle_flag(self.waffle_flag, active=False):
resp = self.client.post(
self.get_url(course_key=self.empty_course.id)
)
self.assertEqual(status.HTTP_403_FORBIDDEN, resp.status_code)
def test_anonymous(self):
with override_waffle_flag(self.waffle_flag, active=True):
resp = self.client.post(self.get_url())
self.assertEqual(status.HTTP_401_UNAUTHORIZED, resp.status_code)
def test_student(self):
self.client.login(username=self.student.username, password=self.password)
with override_waffle_flag(self.waffle_flag, active=True):
resp = self.client.post(self.get_url())
self.assertEqual(status.HTTP_403_FORBIDDEN, resp.status_code)
def test_course_does_not_exist(self):
with override_waffle_flag(self.waffle_flag, active=True):
self.login_staff()
resp = self.client.post(
self.get_url(course_key='course-v1:MITx+8.MechCX+2014_T1')
)
self.assertEqual(status.HTTP_404_NOT_FOUND, resp.status_code)
def test_grades_frozen(self):
"""
Should receive a 403 when grades have been frozen for a course.
"""
with patch('lms.djangoapps.grades.api.v1.views.are_grades_frozen', return_value=True):
with override_waffle_flag(self.waffle_flag, active=True):
self.login_staff()
post_data = [
{
'user_id': self.student.id,
'usage_id': text_type(self.subsections[self.chapter_1.location][0].location),
'grade': {}, # doesn't matter what we put here.
}
]
resp = self.client.post(
self.get_url(),
data=json.dumps(post_data),
content_type='application/json',
)
self.assertEqual(status.HTTP_403_FORBIDDEN, resp.status_code)
def test_user_not_enrolled(self):
with override_waffle_flag(self.waffle_flag, active=True):
self.login_staff()
unenrolled_student = UserFactory()
post_data = [
{
'user_id': unenrolled_student.id,
'usage_id': text_type(self.subsections[self.chapter_1.location][0].location),
'grade': {}, # doesn't matter what we put here.
}
]
resp = self.client.post(
self.get_url(),
data=json.dumps(post_data),
content_type='application/json',
)
expected_data = [
{
'user_id': unenrolled_student.id,
'usage_id': text_type(self.subsections[self.chapter_1.location][0].location),
'success': False,
'reason': 'CourseEnrollment matching query does not exist.',
},
]
self.assertEqual(status.HTTP_422_UNPROCESSABLE_ENTITY, resp.status_code)
self.assertEqual(expected_data, resp.data)
def test_user_does_not_exist(self):
with override_waffle_flag(self.waffle_flag, active=True):
self.login_staff()
post_data = [
{
'user_id': -123,
'usage_id': text_type(self.subsections[self.chapter_1.location][0].location),
'grade': {}, # doesn't matter what we put here.
}
]
resp = self.client.post(
self.get_url(),
data=json.dumps(post_data),
content_type='application/json',
)
expected_data = [
{
'user_id': -123,
'usage_id': text_type(self.subsections[self.chapter_1.location][0].location),
'success': False,
'reason': 'User matching query does not exist.',
},
]
self.assertEqual(status.HTTP_422_UNPROCESSABLE_ENTITY, resp.status_code)
self.assertEqual(expected_data, resp.data)
def test_invalid_usage_key(self):
with override_waffle_flag(self.waffle_flag, active=True):
self.login_staff()
post_data = [
{
'user_id': self.student.id,
'usage_id': 'not-a-valid-usage-key',
'grade': {}, # doesn't matter what we put here.
}
]
resp = self.client.post(
self.get_url(),
data=json.dumps(post_data),
content_type='application/json',
)
expected_data = [
{
'user_id': self.student.id,
'usage_id': 'not-a-valid-usage-key',
'success': False,
'reason': "<class 'opaque_keys.edx.locator.BlockUsageLocator'>: not-a-valid-usage-key",
},
]
self.assertEqual(status.HTTP_422_UNPROCESSABLE_ENTITY, resp.status_code)
self.assertEqual(expected_data, resp.data)
def test_subsection_does_not_exist(self):
"""
When trying to override a grade for a valid usage key that does not exist in the requested course,
we should get an error reason specifying that the key does not exist in the course.
"""
with override_waffle_flag(self.waffle_flag, active=True):
self.login_staff()
usage_id = 'block-v1:edX+DemoX+Demo_Course+type@sequential+block@workflow'
post_data = [
{
'user_id': self.student.id,
'usage_id': usage_id,
'grade': {}, # doesn't matter what we put here.
}
]
resp = self.client.post(
self.get_url(),
data=json.dumps(post_data),
content_type='application/json',
)
expected_data = [
{
'user_id': self.student.id,
'usage_id': usage_id,
'success': False,
'reason': 'usage_key {} does not exist in this course.'.format(usage_id),
},
]
self.assertEqual(status.HTTP_422_UNPROCESSABLE_ENTITY, resp.status_code)
self.assertEqual(expected_data, resp.data)
def test_override_is_created(self):
"""
Test that when we make multiple requests to update grades for the same user/subsection,
the score from the most recent request is recorded.
"""
with override_waffle_flag(self.waffle_flag, active=True):
self.login_staff()
post_data = [
{
'user_id': self.student.id,
'usage_id': text_type(self.subsections[self.chapter_1.location][0].location),
'grade': {
'earned_all_override': 3,
'possible_all_override': 3,
'earned_graded_override': 2,
'possible_graded_override': 2,
},
},
{
'user_id': self.student.id,
'usage_id': text_type(self.subsections[self.chapter_1.location][1].location),
'grade': {
'earned_all_override': 1,
'possible_all_override': 4,
'earned_graded_override': 1,
'possible_graded_override': 4,
},
}
]
resp = self.client.post(
self.get_url(),
data=json.dumps(post_data),
content_type='application/json',
)
expected_data = [
{
'user_id': self.student.id,
'usage_id': text_type(self.subsections[self.chapter_1.location][0].location),
'success': True,
'reason': None,
},
{
'user_id': self.student.id,
'usage_id': text_type(self.subsections[self.chapter_1.location][1].location),
'success': True,
'reason': None,
},
]
self.assertEqual(status.HTTP_202_ACCEPTED, resp.status_code)
self.assertEqual(expected_data, resp.data)
second_post_data = [
{
'user_id': self.student.id,
'usage_id': text_type(self.subsections[self.chapter_1.location][1].location),
'grade': {
'earned_all_override': 3,
'possible_all_override': 4,
'earned_graded_override': 3,
'possible_graded_override': 4,
},
},
]
self.client.post(
self.get_url(),
data=json.dumps(second_post_data),
content_type='application/json',
)
GradeFields = namedtuple('GradeFields', ['earned_all', 'possible_all', 'earned_graded', 'possible_graded'])
# We should now have PersistentSubsectionGradeOverride records corresponding to
# our bulk-update request, and PersistentSubsectionGrade records with grade values
# equal to those of the override.
for usage_key, expected_grades in (
(self.subsections[self.chapter_1.location][0].location, GradeFields(3, 3, 2, 2)),
(self.subsections[self.chapter_1.location][1].location, GradeFields(3, 4, 3, 4)),
):
# this selects related PersistentSubsectionGradeOverride objects.
grade = PersistentSubsectionGrade.read_grade(
user_id=self.student.id,
usage_key=usage_key,
)
for field_name in expected_grades._fields:
expected_value = getattr(expected_grades, field_name)
self.assertEqual(expected_value, getattr(grade, field_name))
self.assertEqual(expected_value, getattr(grade.override, field_name + '_override'))
update_records = PersistentSubsectionGradeOverrideHistory.objects.filter(user=self.global_staff)
self.assertEqual(update_records.count(), 3)
for audit_item in update_records:
self.assertEqual(audit_item.user, self.global_staff)
self.assertIsNotNone(audit_item.created)
self.assertEqual(audit_item.feature, PersistentSubsectionGradeOverrideHistory.GRADEBOOK)
self.assertEqual(audit_item.action, PersistentSubsectionGradeOverrideHistory.CREATE_OR_UPDATE)

4
lms/djangoapps/grades/api/v1/urls.py
View File

@@ -26,12 +26,12 @@ urlpatterns = [
),
url(
r'^gradebook/{course_id}/$'.format(course_id=settings.COURSE_ID_PATTERN),
views.GradebookView.as_view(),
gradebook_views.GradebookView.as_view(),
name='course_gradebook'
),
url(
r'^gradebook/{course_id}/bulk-update$'.format(course_id=settings.COURSE_ID_PATTERN),
views.GradebookBulkUpdateView.as_view(),
gradebook_views.GradebookBulkUpdateView.as_view(),
name='course_gradebook_bulk_update'
),
url(

234
lms/djangoapps/grades/api/v1/utils.py Normal file
View File

@@ -0,0 +1,234 @@
"""
Define some view level utility functions here that multiple view modules will share
"""
from contextlib import contextmanager
from functools import wraps
from rest_framework import status
from rest_framework.exceptions import AuthenticationFailed
from rest_framework.pagination import CursorPagination
from rest_framework.response import Response
from rest_framework.views import APIView
from six import text_type
from django.contrib.auth import get_user_model
from lms.djangoapps.grades.course_grade_factory import CourseGradeFactory
from opaque_keys import InvalidKeyError
from opaque_keys.edx.keys import CourseKey
from openedx.core.djangoapps.content.course_overviews.models import CourseOverview
from openedx.core.lib.api.view_utils import DeveloperErrorViewMixin
from student.models import CourseEnrollment
USER_MODEL = get_user_model()
def get_course_key(request, course_id=None):
if not course_id:
return CourseKey.from_string(request.GET.get('course_id'))
return CourseKey.from_string(course_id)
def verify_course_exists(view_func):
"""
A decorator to wrap a view function that takes `course_key` as a parameter.
Raises:
An API error if the `course_key` is invalid, or if no `CourseOverview` exists for the given key.
"""
@wraps(view_func)
def wrapped_function(self, request, **kwargs):
"""
Wraps the given view_function.
"""
try:
course_key = get_course_key(request, kwargs.get('course_id'))
except InvalidKeyError:
raise self.api_error(
status_code=status.HTTP_404_NOT_FOUND,
developer_message='The provided course key cannot be parsed.',
error_code='invalid_course_key'
)
if not CourseOverview.get_from_id_if_exists(course_key):
raise self.api_error(
status_code=status.HTTP_404_NOT_FOUND,
developer_message="Requested grade for unknown course {course}".format(course=text_type(course_key)),
error_code='course_does_not_exist'
)
return view_func(self, request, **kwargs)
return wrapped_function
class CourseEnrollmentPagination(CursorPagination):
"""
Paginates over CourseEnrollment objects.
"""
ordering = 'id'
page_size = 50
page_size_query_param = 'page_size'
def get_page_size(self, request):
"""
Get the page size based on the defined page size parameter if defined.
"""
try:
page_size_string = request.query_params[self.page_size_query_param]
return int(page_size_string)
except (KeyError, ValueError):
pass
return self.page_size
class PaginatedAPIView(APIView):
"""
An `APIView` class enhanced with the pagination methods of `GenericAPIView`.
"""
# pylint: disable=attribute-defined-outside-init
@property
def paginator(self):
"""
The paginator instance associated with the view, or `None`.
"""
if not hasattr(self, '_paginator'):
if self.pagination_class is None:
self._paginator = None
else:
self._paginator = self.pagination_class()
return self._paginator
def paginate_queryset(self, queryset):
"""
Return a single page of results, or `None` if pagination is disabled.
"""
if self.paginator is None:
return None
return self.paginator.paginate_queryset(queryset, self.request, view=self)
def get_paginated_response(self, data):
"""
Return a paginated style `Response` object for the given output data.
"""
assert self.paginator is not None
return self.paginator.get_paginated_response(data)
class GradeViewMixin(DeveloperErrorViewMixin):
"""
Mixin class for Grades related views.
"""
def _get_single_user(self, request, course_key, user_id=None):
"""
Returns a single USER_MODEL object corresponding to either the user_id provided, or if no id is provided,
then the request's `username` parameter, or the current `request.user` if no `username` was provided.
Args:
request (Request): django request object to check for username or request.user object
course_key (CourseLocator): The course to retrieve user grades for.
user_id (int): Optional user id to fetch the user object for.
Returns:
A USER_MODEL object.
Raises:
USER_MODEL.DoesNotExist if no such user exists.
CourseEnrollment.DoesNotExist if the user is not enrolled in the given course.
"""
# May raise USER_MODEL.DoesNotExist if no user matching the given query exists.
if user_id:
grade_user = USER_MODEL.objects.get(id=user_id)
elif 'username' in request.GET:
grade_user = USER_MODEL.objects.get(username=request.GET.get('username'))
else:
grade_user = request.user
# May raise CourseEnrollment.DoesNotExist if no enrollment exists for this user/course.
_ = CourseEnrollment.objects.get(user=grade_user, course_id=course_key)
return grade_user
@contextmanager
def _get_user_or_raise(self, request, course_key):
"""
Raises an API error if the username specified by the request does not exist, or if the
user is not enrolled in the given course.
Args:
request (Request): django request object to check for username or request.user object
course_key (CourseLocator): The course to retrieve user grades for.
Yields:
A USER_MODEL object.
"""
try:
yield self._get_single_user(request, course_key)
except USER_MODEL.DoesNotExist:
raise self.api_error(
status_code=status.HTTP_404_NOT_FOUND,
developer_message='The user matching the requested username does not exist.',
error_code='user_does_not_exist'
)
except CourseEnrollment.DoesNotExist:
raise self.api_error(
status_code=status.HTTP_404_NOT_FOUND,
developer_message='The user matching the requested username is not enrolled in this course',
error_code='user_not_enrolled'
)
def _get_single_user_grade(self, grade_user, course_key):
"""
Returns a grade response for the user object corresponding to the request's 'username' parameter,
or the current request.user if no 'username' was provided.
Args:
request (Request): django request object to check for username or request.user object
course_key (CourseLocator): The course to retrieve user grades for.
Returns:
A serializable list of grade responses
"""
course_grade = CourseGradeFactory().read(grade_user, course_key=course_key)
return Response([self._serialize_user_grade(grade_user, course_key, course_grade)])
def _paginate_users(self, course_key, course_enrollment_filter=None, related_models=None):
"""
Args:
course_key (CourseLocator): The course to retrieve grades for.
course_enrollment_filter: Optional dictionary of keyword arguments to pass
to `CourseEnrollment.filter()`.
related_models: Optional list of related models to join to the CourseEnrollment table.
Returns:
A list of users, pulled from a paginated queryset of enrollments, who are enrolled in the given course.
"""
filter_kwargs = {
'course_id': course_key,
'is_active': True,
}
filter_kwargs.update(course_enrollment_filter or {})
enrollments_in_course = CourseEnrollment.objects.filter(**filter_kwargs)
if related_models:
enrollments_in_course = enrollments_in_course.select_related(*related_models)
paged_enrollments = self.paginate_queryset(enrollments_in_course)
return [enrollment.user for enrollment in paged_enrollments]
def _serialize_user_grade(self, user, course_key, course_grade):
"""
Serialize a single grade to dict to use in Responses
"""
return {
'username': user.username,
'email': user.email,
'course_id': str(course_key),
'passed': course_grade.passed,
'percent': course_grade.percent,
'letter_grade': course_grade.letter_grade,
}
def perform_authentication(self, request):
"""
Ensures that the user is authenticated (e.g. not an AnonymousUser).
"""
super(GradeViewMixin, self).perform_authentication(request)
if request.user.is_anonymous:
raise AuthenticationFailed

782
lms/djangoapps/grades/api/v1/views.py
View File

@@ -1,116 +1,22 @@
""" API v0 views. """
import logging
from collections import namedtuple
from contextlib import contextmanager
from functools import wraps
from django.contrib.auth import get_user_model
from django.urls import reverse
from rest_framework import status
from rest_framework.exceptions import AuthenticationFailed
from rest_framework.pagination import CursorPagination
from rest_framework.response import Response
from rest_framework.views import APIView
from six import text_type
from util.date_utils import to_timestamp
from courseware.courses import get_course_with_access
from edx_rest_framework_extensions import permissions
from edx_rest_framework_extensions.auth.jwt.authentication import JwtAuthentication
from edx_rest_framework_extensions.auth.session.authentication import SessionAuthenticationAllowInactiveUser
from lms.djangoapps.grades.api.serializers import StudentGradebookEntrySerializer
from lms.djangoapps.grades.config.waffle import waffle_flags, WRITABLE_GRADEBOOK
from lms.djangoapps.grades.constants import ScoreDatabaseTableEnum
from lms.djangoapps.grades.course_data import CourseData
from lms.djangoapps.grades.api.v1.utils import (
CourseEnrollmentPagination,
GradeViewMixin,
PaginatedAPIView,
get_course_key,
verify_course_exists
)
from lms.djangoapps.grades.course_grade_factory import CourseGradeFactory
from lms.djangoapps.grades.events import SUBSECTION_GRADE_CALCULATED, subsection_grade_calculated
from lms.djangoapps.grades.models import (
PersistentCourseGrade,
PersistentSubsectionGrade,
PersistentSubsectionGradeOverride,
PersistentSubsectionGradeOverrideHistory
)
from lms.djangoapps.grades.subsection_grade import CreateSubsectionGrade
from lms.djangoapps.grades.tasks import recalculate_subsection_grade_v3, are_grades_frozen
from opaque_keys import InvalidKeyError
from opaque_keys.edx.keys import CourseKey, UsageKey
from openedx.core.djangoapps.content.course_overviews.models import CourseOverview
from openedx.core.djangoapps.course_groups import cohorts
from lms.djangoapps.grades.models import PersistentCourseGrade
from openedx.core.lib.api.authentication import OAuth2AuthenticationAllowInactiveUser
from openedx.core.lib.api.view_utils import DeveloperErrorViewMixin
from student.models import CourseEnrollment
from student.roles import BulkRoleCache
from track.event_transaction_utils import (
create_new_event_transaction_id,
get_event_transaction_id,
get_event_transaction_type,
set_event_transaction_type
)
from xmodule.util.misc import get_default_short_labeler
log = logging.getLogger(__name__)
USER_MODEL = get_user_model()
def get_course_key(request, course_id=None):
if not course_id:
return CourseKey.from_string(request.GET.get('course_id'))
return CourseKey.from_string(course_id)
def verify_course_exists(view_func):
"""
A decorator to wrap a view function that takes `course_key` as a parameter.
Raises:
An API error if the `course_key` is invalid, or if no `CourseOverview` exists for the given key.
"""
@wraps(view_func)
def wrapped_function(self, request, **kwargs):
"""
Wraps the given view_function.
"""
try:
course_key = get_course_key(request, kwargs.get('course_id'))
except InvalidKeyError:
raise self.api_error(
status_code=status.HTTP_404_NOT_FOUND,
developer_message='The provided course key cannot be parsed.',
error_code='invalid_course_key'
)
if not CourseOverview.get_from_id_if_exists(course_key):
raise self.api_error(
status_code=status.HTTP_404_NOT_FOUND,
developer_message="Requested grade for unknown course {course}".format(course=text_type(course_key)),
error_code='course_does_not_exist'
)
return view_func(self, request, **kwargs)
return wrapped_function
def verify_writable_gradebook_enabled(view_func):
"""
A decorator to wrap a view function that takes `course_key` as a parameter.
Raises:
A 403 API error if the writable gradebook feature is not enabled for the given course.
"""
@wraps(view_func)
def wrapped_function(self, request, **kwargs):
"""
Wraps the given view function.
"""
course_key = get_course_key(request, kwargs.get('course_id'))
if not waffle_flags()[WRITABLE_GRADEBOOK].is_enabled(course_key):
raise self.api_error(
status_code=status.HTTP_403_FORBIDDEN,
developer_message='The writable gradebook feature is not enabled for this course.',
error_code='feature_not_enabled'
)
return view_func(self, request, **kwargs)
return wrapped_function
@contextmanager
@@ -125,197 +31,6 @@ def bulk_course_grade_context(course_key, users):
PersistentCourseGrade.clear_prefetched_data(course_key)
@contextmanager
def bulk_gradebook_view_context(course_key, users):
"""
Prefetches all course and subsection grades in the given course for the given
list of users, also, fetch all the score relavant data,
storing the result in a RequestCache and deleting grades on context exit.
"""
PersistentSubsectionGrade.prefetch(course_key, users)
PersistentCourseGrade.prefetch(course_key, users)
CourseEnrollment.bulk_fetch_enrollment_states(users, course_key)
cohorts.bulk_cache_cohorts(course_key, users)
BulkRoleCache.prefetch(users)
yield
PersistentSubsectionGrade.clear_prefetched_data(course_key)
PersistentCourseGrade.clear_prefetched_data(course_key)
class CourseEnrollmentPagination(CursorPagination):
"""
Paginates over CourseEnrollment objects.
"""
ordering = 'id'
page_size = 50
page_size_query_param = 'page_size'
def get_page_size(self, request):
"""
Get the page size based on the defined page size parameter if defined.
"""
try:
page_size_string = request.query_params[self.page_size_query_param]
return int(page_size_string)
except (KeyError, ValueError):
pass
return self.page_size
class PaginatedAPIView(APIView):
"""
An `APIView` class enhanced with the pagination methods of `GenericAPIView`.
"""
# pylint: disable=attribute-defined-outside-init
@property
def paginator(self):
"""
The paginator instance associated with the view, or `None`.
"""
if not hasattr(self, '_paginator'):
if self.pagination_class is None:
self._paginator = None
else:
self._paginator = self.pagination_class()
return self._paginator
def paginate_queryset(self, queryset):
"""
Return a single page of results, or `None` if pagination is disabled.
"""
if self.paginator is None:
return None
return self.paginator.paginate_queryset(queryset, self.request, view=self)
def get_paginated_response(self, data):
"""
Return a paginated style `Response` object for the given output data.
"""
assert self.paginator is not None
return self.paginator.get_paginated_response(data)
class GradeViewMixin(DeveloperErrorViewMixin):
"""
Mixin class for Grades related views.
"""
def _get_single_user(self, request, course_key, user_id=None):
"""
Returns a single USER_MODEL object corresponding to either the user_id provided, or if no id is provided,
then the request's `username` parameter, or the current `request.user` if no `username` was provided.
Args:
request (Request): django request object to check for username or request.user object
course_key (CourseLocator): The course to retrieve user grades for.
user_id (int): Optional user id to fetch the user object for.
Returns:
A USER_MODEL object.
Raises:
USER_MODEL.DoesNotExist if no such user exists.
CourseEnrollment.DoesNotExist if the user is not enrolled in the given course.
"""
# May raise USER_MODEL.DoesNotExist if no user matching the given query exists.
if user_id:
grade_user = USER_MODEL.objects.get(id=user_id)
elif 'username' in request.GET:
grade_user = USER_MODEL.objects.get(username=request.GET.get('username'))
else:
grade_user = request.user
# May raise CourseEnrollment.DoesNotExist if no enrollment exists for this user/course.
_ = CourseEnrollment.objects.get(user=grade_user, course_id=course_key)
return grade_user
@contextmanager
def _get_user_or_raise(self, request, course_key):
"""
Raises an API error if the username specified by the request does not exist, or if the
user is not enrolled in the given course.
Args:
request (Request): django request object to check for username or request.user object
course_key (CourseLocator): The course to retrieve user grades for.
Yields:
A USER_MODEL object.
"""
try:
yield self._get_single_user(request, course_key)
except USER_MODEL.DoesNotExist:
raise self.api_error(
status_code=status.HTTP_404_NOT_FOUND,
developer_message='The user matching the requested username does not exist.',
error_code='user_does_not_exist'
)
except CourseEnrollment.DoesNotExist:
raise self.api_error(
status_code=status.HTTP_404_NOT_FOUND,
developer_message='The user matching the requested username is not enrolled in this course',
error_code='user_not_enrolled'
)
def _get_single_user_grade(self, grade_user, course_key):
"""
Returns a grade response for the user object corresponding to the request's 'username' parameter,
or the current request.user if no 'username' was provided.
Args:
request (Request): django request object to check for username or request.user object
course_key (CourseLocator): The course to retrieve user grades for.
Returns:
A serializable list of grade responses
"""
course_grade = CourseGradeFactory().read(grade_user, course_key=course_key)
return Response([self._serialize_user_grade(grade_user, course_key, course_grade)])
def _paginate_users(self, course_key, course_enrollment_filter=None, related_models=None):
"""
Args:
course_key (CourseLocator): The course to retrieve grades for.
course_enrollment_filter: Optional dictionary of keyword arguments to pass
to `CourseEnrollment.filter()`.
related_models: Optional list of related models to join to the CourseEnrollment table.
Returns:
A list of users, pulled from a paginated queryset of enrollments, who are enrolled in the given course.
"""
filter_kwargs = {
'course_id': course_key,
'is_active': True,
}
filter_kwargs.update(course_enrollment_filter or {})
enrollments_in_course = CourseEnrollment.objects.filter(**filter_kwargs)
if related_models:
enrollments_in_course = enrollments_in_course.select_related(*related_models)
paged_enrollments = self.paginate_queryset(enrollments_in_course)
return [enrollment.user for enrollment in paged_enrollments]
def _serialize_user_grade(self, user, course_key, course_grade):
"""
Serialize a single grade to dict to use in Responses
"""
return {
'username': user.username,
'email': user.email,
'course_id': str(course_key),
'passed': course_grade.passed,
'percent': course_grade.percent,
'letter_grade': course_grade.letter_grade,
}
def perform_authentication(self, request):
"""
Ensures that the user is authenticated (e.g. not an AnonymousUser).
"""
super(GradeViewMixin, self).perform_authentication(request)
if request.user.is_anonymous:
raise AuthenticationFailed
class CourseGradesView(GradeViewMixin, PaginatedAPIView):
"""
**Use Case**
@@ -421,484 +136,3 @@ class CourseGradesView(GradeViewMixin, PaginatedAPIView):
user_grades.append(self._serialize_user_grade(user, course_key, course_grade))
return self.get_paginated_response(user_grades)
class GradebookView(GradeViewMixin, PaginatedAPIView):
"""
**Use Case**
* Get course gradebook entries of a single user in a course,
or of all users who are actively enrolled in a course. The currently logged-in user may request
all enrolled user's grades information if they are allowed.
**Example Request**
GET /api/grades/v1/gradebook/{course_id}/ - Get gradebook entries for all users in course
GET /api/grades/v1/gradebook/{course_id}/?username={username} - Get grades for specific user in course
GET /api/grades/v1/gradebook/{course_id}/?username_contains={username_contains}
GET /api/grades/v1/gradebook/{course_id}/?cohort_id={cohort_id}
GET /api/grades/v1/gradebook/{course_id}/?enrollment_mode={enrollment_mode}
**GET Parameters**
A GET request may include the following query parameters.
* username: (optional) A string representation of a user's username.
* username_contains: (optional) A substring against which a case-insensitive substring filter will be performed
on the USER_MODEL.username field.
* cohort_id: (optional) The id of a cohort in this course. If present, will return grades
only for course enrollees who belong to that cohort.
* enrollment_mode: (optional) The slug of an enrollment mode (e.g. "verified"). If present, will return grades
only for course enrollees with the given enrollment mode.
**GET Response Values**
If the request for gradebook data is successful,
an HTTP 200 "OK" response is returned.
The HTTP 200 response for a single has the following values:
* course_id: A string representation of a Course ID.
* email: A string representation of a user's email.
* user_id: The user's integer id.
* username: A string representation of a user's username passed in the request.
* full_name: A string representation of the user's full name.
* passed: Boolean representing whether the course has been
passed according to the course's grading policy.
* percent: A float representing the overall grade for the course
* letter_grade: A letter grade as defined in grading policy (e.g. 'A' 'B' 'C' for 6.002x) or None
* progress_page_url: A link to the user's progress page.
* section_breakdown: A list of subsection grade details, as specified below.
A response for all user's grades in the course is paginated, and contains "count", "next" and "previous"
keys, along with the actual data contained in a "results" list.
An HTTP 404 may be returned for the following reasons:
* The requested course_key is invalid.
* No course corresponding to the requested key exists.
* No user corresponding to the requested username exists.
* The requested user is not enrolled in the requested course.
An HTTP 403 may be returned if the `writable_gradebook` feature is not
enabled for this course.
**Example GET Response**
{
"course_id": "course-v1:edX+DemoX+Demo_Course",
"email": "staff@example.com",
"user_id": 9,
"username": "staff",
"full_name": "",
"passed": false,
"percent": 0.36,
"letter_grade": null,
"progress_page_url": "/courses/course-v1:edX+DemoX+Demo_Course/progress/9/",
"section_breakdown": [
{
"are_grades_published": true,
"auto_grade": false,
"category": null,
"chapter_name": "Introduction",
"comment": "",
"detail": "",
"displayed_value": "0.00",
"is_graded": false,
"grade_description": "(0.00/0.00)",
"is_ag": false,
"is_average": false,
"is_manually_graded": false,
"label": null,
"letter_grade": null,
"module_id": "block-v1:edX+DemoX+Demo_Course+type@sequential+block@edx_introduction",
"percent": 0.0,
"score_earned": 0.0,
"score_possible": 0.0,
"section_block_id": "block-v1:edX+DemoX+Demo_Course+type@chapter+block@abcdefgh123",
"subsection_name": "Demo Course Overview"
},
],
}
**Paginated GET response**
When requesting gradebook entries for all users, the response is paginated and contains the following values:
* count: The total number of user gradebook entries for this course.
* next: The URL containing the next page of data.
* previous: The URL containing the previous page of data.
* results: A list of user gradebook entries, structured as above.
Note: It's important that `GradeViewMixin` is the first inherited class here, so that
self.api_error returns error responses as expected.
"""
authentication_classes = (
JwtAuthentication,
OAuth2AuthenticationAllowInactiveUser,
SessionAuthenticationAllowInactiveUser,
)
permission_classes = (permissions.JWT_RESTRICTED_APPLICATION_OR_USER_ACCESS,)
pagination_class = CourseEnrollmentPagination
required_scopes = ['grades:read']
def _section_breakdown(self, course, graded_subsections, course_grade):
"""
Given a course_grade and a list of graded subsections for a given course,
returns a list of grade data broken down by subsection.
Args:
course: A Course Descriptor object
graded_subsections: A list of graded subsection objects in the given course.
course_grade: A CourseGrade object.
"""
breakdown = []
default_labeler = get_default_short_labeler(course)
for subsection in graded_subsections:
subsection_grade = course_grade.subsection_grade(subsection.location)
short_label = default_labeler(subsection_grade.format)
graded_description = 'Not Attempted'
score_earned = 0
score_possible = 0
# For ZeroSubsectionGrades, we don't want to crawl the subsection's
# subtree to find the problem scores specific to this user
# (ZeroSubsectionGrade.attempted_graded is always False).
# We've already fetched the whole course structure in a non-specific way
# when creating `graded_subsections`. Looking at the problem scores
# specific to this user (the user in `course_grade.user`) would require
# us to re-fetch the user-specific course structure from the modulestore,
# which is a costly operation.
if subsection_grade.attempted_graded:
graded_description = '({earned:.2f}/{possible:.2f})'.format(
earned=subsection_grade.graded_total.earned,
possible=subsection_grade.graded_total.possible,
)
score_earned = subsection_grade.graded_total.earned
score_possible = subsection_grade.graded_total.possible
# TODO: https://openedx.atlassian.net/browse/EDUCATOR-3559 -- Some fields should be renamed, others removed:
# 'displayed_value' should maybe be 'description_percent'
# 'grade_description' should be 'description_ratio'
breakdown.append({
'category': subsection_grade.format,
'displayed_value': '{:.2f}'.format(subsection_grade.percent_graded),
'is_graded': subsection_grade.graded,
'grade_description': graded_description,
'label': short_label,
'letter_grade': course_grade.letter_grade,
'module_id': text_type(subsection_grade.location),
'percent': subsection_grade.percent_graded,
'score_earned': score_earned,
'score_possible': score_possible,
'subsection_name': subsection_grade.display_name,
})
return breakdown
def _gradebook_entry(self, user, course, graded_subsections, course_grade):
"""
Returns a dictionary of course- and subsection-level grade data for
a given user in a given course.
Args:
user: A User object.
course: A Course Descriptor object.
graded_subsections: A list of graded subsections in the given course.
course_grade: A CourseGrade object.
"""
user_entry = self._serialize_user_grade(user, course.id, course_grade)
breakdown = self._section_breakdown(course, graded_subsections, course_grade)
user_entry['section_breakdown'] = breakdown
user_entry['progress_page_url'] = reverse(
'student_progress',
kwargs=dict(course_id=text_type(course.id), student_id=user.id)
)
user_entry['user_id'] = user.id
user_entry['full_name'] = user.get_full_name()
return user_entry
@verify_course_exists
@verify_writable_gradebook_enabled
def get(self, request, course_id):
"""
Returns a gradebook entry/entries (i.e. both course and subsection-level grade data)
for all users enrolled in a course, or a single user enrolled in a course
if a `username` parameter is provided.
Args:
request: A Django request object.
course_id: A string representation of a CourseKey object.
"""
course_key = get_course_key(request, course_id)
course = get_course_with_access(request.user, 'staff', course_key, depth=None)
# We fetch the entire course structure up-front, and use this when iterating
# over users to determine their subsection grades. We purposely avoid fetching
# the user-specific course structure for each user, because that is very expensive.
course_data = CourseData(user=None, course=course)
graded_subsections = list(graded_subsections_for_course(course_data.collected_structure))
if request.GET.get('username'):
with self._get_user_or_raise(request, course_key) as grade_user:
course_grade = CourseGradeFactory().read(grade_user, course)
entry = self._gradebook_entry(grade_user, course, graded_subsections, course_grade)
serializer = StudentGradebookEntrySerializer(entry)
return Response(serializer.data)
else:
filter_kwargs = {}
related_models = []
if request.GET.get('username_contains'):
filter_kwargs['user__username__icontains'] = request.GET.get('username_contains')
related_models.append('user')
if request.GET.get('cohort_id'):
cohort = cohorts.get_cohort_by_id(course_key, request.GET.get('cohort_id'))
if cohort:
filter_kwargs['user__in'] = cohort.users.all()
else:
filter_kwargs['user__in'] = []
if request.GET.get('enrollment_mode'):
filter_kwargs['mode'] = request.GET.get('enrollment_mode')
entries = []
users = self._paginate_users(course_key, filter_kwargs, related_models)
with bulk_gradebook_view_context(course_key, users):
for user, course_grade, exc in CourseGradeFactory().iter(
users, course_key=course_key, collected_block_structure=course_data.collected_structure
):
if not exc:
entries.append(self._gradebook_entry(user, course, graded_subsections, course_grade))
serializer = StudentGradebookEntrySerializer(entries, many=True)
return self.get_paginated_response(serializer.data)
def graded_subsections_for_course(course_structure):
"""
Given a course block structure, yields the subsections of the course that are graded.
Args:
course_structure: A course structure object. Not user-specific.
"""
for chapter_key in course_structure.get_children(course_structure.root_block_usage_key):
for subsection_key in course_structure.get_children(chapter_key):
subsection = course_structure[subsection_key]
if subsection.graded:
yield subsection
GradebookUpdateResponseItem = namedtuple('GradebookUpdateResponseItem', ['user_id', 'usage_id', 'success', 'reason'])
class GradebookBulkUpdateView(GradeViewMixin, PaginatedAPIView):
"""
**Use Case**
Creates `PersistentSubsectionGradeOverride` objects for multiple (user_id, usage_id)
pairs in a given course, and invokes a Django signal to update subsection grades in
an asynchronous celery task.
**Example Request**
POST /api/grades/v1/gradebook/{course_id}/bulk-update
**POST Parameters**
This endpoint does not accept any URL parameters.
**Example POST Data**
[
{
"user_id": 9,
"usage_id": "block-v1:edX+DemoX+Demo_Course+type@sequential+block@basic_questions",
"grade": {
"earned_all_override": 11,
"possible_all_override": 11,
"earned_graded_override": 11,
"possible_graded_override": 11
}
},
{
"user_id": 9,
"usage_id": "block-v1:edX+DemoX+Demo_Course+type@sequential+block@advanced_questions",
"grade": {
"earned_all_override": 10,
"possible_all_override": 15,
"earned_graded_override": 9,
"possible_graded_override": 12
}
}
]
**POST Response Values**
An HTTP 202 may be returned if a grade override was created for each of the requested (user_id, usage_id)
pairs in the request data.
An HTTP 403 may be returned if the `writable_gradebook` feature is not
enabled for this course.
An HTTP 404 may be returned for the following reasons:
* The requested course_key is invalid.
* No course corresponding to the requested key exists.
* The requesting user is not enrolled in the requested course.
An HTTP 422 may be returned if any of the requested (user_id, usage_id) pairs
did not have a grade override created due to some exception. A `reason` detailing the exception
is provided with each response item.
**Example successful POST Response**
[
{
"user_id": 9,
"usage_id": "some-requested-usage-id",
"success": true,
"reason": null
},
{
"user_id": 9,
"usage_id": "an-invalid-usage-id",
"success": false,
"reason": "<class 'opaque_keys.edx.locator.BlockUsageLocator'>: not-a-valid-usage-key"
},
{
"user_id": 9,
"usage_id": "a-valid-usage-key-that-doesn't-exist",
"success": false,
"reason": "a-valid-usage-key-that-doesn't-exist does not exist in this course"
},
{
"user_id": 1234-I-DO-NOT-EXIST,
"usage_id": "a-valid-usage-key",
"success": false,
"reason": "User matching query does not exist."
}
]
"""
authentication_classes = (
JwtAuthentication,
OAuth2AuthenticationAllowInactiveUser,
SessionAuthenticationAllowInactiveUser,
)
permission_classes = (permissions.JWT_RESTRICTED_APPLICATION_OR_USER_ACCESS,)
required_scopes = ['grades:write']
@verify_course_exists
@verify_writable_gradebook_enabled
def post(self, request, course_id):
"""
Creates or updates `PersistentSubsectionGradeOverrides` for the (user_id, usage_key)
specified in the request data. The `SUBSECTION_OVERRIDE_CHANGED` signal is invoked
after the grade override is created, which triggers a celery task to update the
course and subsection grades for the specified user.
"""
course_key = get_course_key(request, course_id)
if are_grades_frozen(course_key):
raise self.api_error(
status_code=status.HTTP_403_FORBIDDEN,
developer_message='Grades are frozen for this course.',
error_code='grades_frozen'
)
course = get_course_with_access(request.user, 'staff', course_key, depth=None)
result = []
for user_data in request.data:
requested_user_id = user_data['user_id']
requested_usage_id = user_data['usage_id']
try:
user = self._get_single_user(request, course_key, requested_user_id)
usage_key = UsageKey.from_string(requested_usage_id)
except (USER_MODEL.DoesNotExist, InvalidKeyError, CourseEnrollment.DoesNotExist) as exc:
result.append(GradebookUpdateResponseItem(
user_id=requested_user_id,
usage_id=requested_usage_id,
success=False,
reason=text_type(exc)
))
continue
try:
subsection_grade_model = PersistentSubsectionGrade.objects.get(
user_id=user.id,
course_id=course_key,
usage_key=usage_key
)
except PersistentSubsectionGrade.DoesNotExist:
subsection = course.get_child(usage_key)
if subsection:
subsection_grade_model = self._create_subsection_grade(user, course, subsection)
else:
result.append(GradebookUpdateResponseItem(
user_id=requested_user_id,
usage_id=requested_usage_id,
success=False,
reason='usage_key {} does not exist in this course.'.format(usage_key)
))
continue
if subsection_grade_model:
self._create_override(request.user, subsection_grade_model, **user_data['grade'])
result.append(GradebookUpdateResponseItem(
user_id=user.id,
usage_id=text_type(usage_key),
success=True,
reason=None
))
status_code = status.HTTP_422_UNPROCESSABLE_ENTITY
if all((item.success for item in result)):
status_code = status.HTTP_202_ACCEPTED
return Response(
[item._asdict() for item in result],
status=status_code,
content_type='application/json'
)
def _create_subsection_grade(self, user, course, subsection):
course_data = CourseData(user, course=course)
subsection_grade = CreateSubsectionGrade(subsection, course_data.structure, {}, {})
return subsection_grade.update_or_create_model(user, force_update_subsections=True)
def _create_override(self, request_user, subsection_grade_model, **override_data):
"""
Helper method to create a `PersistentSubsectionGradeOverride` object
and send a `SUBSECTION_OVERRIDE_CHANGED` signal.
"""
override, _ = PersistentSubsectionGradeOverride.objects.update_or_create(
grade=subsection_grade_model,
defaults=self._clean_override_data(override_data),
)
_ = PersistentSubsectionGradeOverrideHistory.objects.create(
override_id=override.id,
user=request_user,
feature=PersistentSubsectionGradeOverrideHistory.GRADEBOOK,
action=PersistentSubsectionGradeOverrideHistory.CREATE_OR_UPDATE,
)
set_event_transaction_type(SUBSECTION_GRADE_CALCULATED)
create_new_event_transaction_id()
recalculate_subsection_grade_v3.apply(
kwargs=dict(
user_id=subsection_grade_model.user_id,
anonymous_user_id=None,
course_id=text_type(subsection_grade_model.course_id),
usage_id=text_type(subsection_grade_model.usage_key),
only_if_higher=False,
expected_modified_time=to_timestamp(override.modified),
score_deleted=False,
event_transaction_id=unicode(get_event_transaction_id()),
event_transaction_type=unicode(get_event_transaction_type()),
score_db_table=ScoreDatabaseTableEnum.overrides,
force_update_subsections=True,
)
)
# Emit events to let our tracking system to know we updated subsection grade
subsection_grade_calculated(subsection_grade_model)
def _clean_override_data(self, override_data):
"""
Helper method to strip any grade override field names that won't work
as defaults when calling PersistentSubsectionGradeOverride.update_or_create().
"""
allowed_fields = {
'earned_all_override',
'possible_all_override',
'earned_graded_override',
'possible_graded_override',
}
stripped_data = {}
for field in override_data.keys():
if field in allowed_fields:
stripped_data[field] = override_data[field]
return stripped_data