Skip to content

E
edx-platform
Commit d887c0fe by Sarina Canelake
Options

Remove 'open_ended_grading' djangoapp & URLs (ORA1)

parent cd9fe577
Showing with 0 additions and 1921 deletions
cms/envs/common.py
......@@ -1016,7 +1016,6 @@ ADVANCED_COMPONENT_TYPES = [
'rate', # Allows up-down voting of course content. See https://github.com/pmitros/RateXBlock
'split_test',
'combinedopenended',
'peergrading',
'notes',
'schoolyourself_review',
......
lms/djangoapps/open_ended_grading/open_ended_notifications.py deleted 100644 → 0
import datetime
import json
import logging
from django.conf import settings
from xmodule.open_ended_grading_classes import peer_grading_service
from xmodule.open_ended_grading_classes.controller_query_service import ControllerQueryService
from courseware.access import has_access
from edxmako.shortcuts import render_to_string
from student.models import unique_id_for_user
from util.cache import cache
from .staff_grading_service import StaffGradingService
log = logging.getLogger(__name__)
NOTIFICATION_CACHE_TIME = 300
KEY_PREFIX = "open_ended_"
NOTIFICATION_TYPES = (
('student_needs_to_peer_grade', 'peer_grading', 'Peer Grading'),
('staff_needs_to_grade', 'staff_grading', 'Staff Grading'),
('new_student_grading_to_view', 'open_ended_problems', 'Problems you have submitted'),
('flagged_submissions_exist', 'open_ended_flagged_problems', 'Flagged Submissions')
)
def staff_grading_notifications(course, user):
staff_gs = StaffGradingService(settings.OPEN_ENDED_GRADING_INTERFACE)
pending_grading = False
img_path = ""
course_id = course.id
student_id = unique_id_for_user(user)
notification_type = "staff"
success, notification_dict = get_value_from_cache(student_id, course_id, notification_type)
if success:
return notification_dict
try:
notifications = json.loads(staff_gs.get_notifications(course_id))
if notifications['success']:
if notifications['staff_needs_to_grade']:
pending_grading = True
except:
#Non catastrophic error, so no real action
notifications = {}
#This is a dev_facing_error
log.info(
"Problem with getting notifications from staff grading service for course {0} user {1}.".format(course_id,
student_id))
if pending_grading:
img_path = "/static/images/grading_notification.png"
notification_dict = {'pending_grading': pending_grading, 'img_path': img_path, 'response': notifications}
set_value_in_cache(student_id, course_id, notification_type, notification_dict)
return notification_dict
def peer_grading_notifications(course, user):
peer_gs = peer_grading_service.PeerGradingService(settings.OPEN_ENDED_GRADING_INTERFACE, render_to_string)
pending_grading = False
img_path = ""
course_id = course.id
student_id = unique_id_for_user(user)
notification_type = "peer"
success, notification_dict = get_value_from_cache(student_id, course_id, notification_type)
if success:
return notification_dict
try:
notifications = json.loads(peer_gs.get_notifications(course_id, student_id))
if notifications['success']:
if notifications['student_needs_to_peer_grade']:
pending_grading = True
except:
#Non catastrophic error, so no real action
notifications = {}
#This is a dev_facing_error
log.info(
"Problem with getting notifications from peer grading service for course {0} user {1}.".format(course_id,
student_id))
if pending_grading:
img_path = "/static/images/grading_notification.png"
notification_dict = {'pending_grading': pending_grading, 'img_path': img_path, 'response': notifications}
set_value_in_cache(student_id, course_id, notification_type, notification_dict)
return notification_dict
def combined_notifications(course, user):
"""
Show notifications to a given user for a given course. Get notifications from the cache if possible,
or from the grading controller server if not.
@param course: The course object for which we are getting notifications
@param user: The user object for which we are getting notifications
@return: A dictionary with boolean pending_grading (true if there is pending grading), img_path (for notification
image), and response (actual response from grading controller server).
"""
#Set up return values so that we can return them for error cases
pending_grading = False
img_path = ""
notifications = {}
notification_dict = {'pending_grading': pending_grading, 'img_path': img_path, 'response': notifications}
#We don't want to show anonymous users anything.
if not user.is_authenticated():
return notification_dict
#Initialize controller query service using our mock system
controller_qs = ControllerQueryService(settings.OPEN_ENDED_GRADING_INTERFACE, render_to_string)
student_id = unique_id_for_user(user)
user_is_staff = bool(has_access(user, 'staff', course))
course_id = course.id
notification_type = "combined"
#See if we have a stored value in the cache
success, notification_dict = get_value_from_cache(student_id, course_id, notification_type)
if success:
return notification_dict
#Get the time of the last login of the user
last_login = user.last_login
last_time_viewed = last_login - datetime.timedelta(seconds=(NOTIFICATION_CACHE_TIME + 60))
try:
#Get the notifications from the grading controller
notifications = controller_qs.check_combined_notifications(
course.id,
student_id,
user_is_staff,
last_time_viewed,
)
if notifications.get('success'):
if (notifications.get('staff_needs_to_grade') or
notifications.get('student_needs_to_peer_grade')):
pending_grading = True
except:
#Non catastrophic error, so no real action
#This is a dev_facing_error
log.exception(
u"Problem with getting notifications from controller query service for course {0} user {1}.".format(
course_id, student_id))
if pending_grading:
img_path = "/static/images/grading_notification.png"
notification_dict = {'pending_grading': pending_grading, 'img_path': img_path, 'response': notifications}
#Store the notifications in the cache
set_value_in_cache(student_id, course_id, notification_type, notification_dict)
return notification_dict
def get_value_from_cache(student_id, course_id, notification_type):
key_name = create_key_name(student_id, course_id, notification_type)
success, value = _get_value_from_cache(key_name)
return success, value
def set_value_in_cache(student_id, course_id, notification_type, value):
key_name = create_key_name(student_id, course_id, notification_type)
_set_value_in_cache(key_name, value)
def create_key_name(student_id, course_id, notification_type):
key_name = u"{prefix}{type}_{course}_{student}".format(
prefix=KEY_PREFIX,
type=notification_type,
course=course_id,
student=student_id,
)
return key_name
def _get_value_from_cache(key_name):
value = cache.get(key_name)
success = False
if value is None:
return success, value
try:
value = json.loads(value)
success = True
except:
pass
return success, value
def _set_value_in_cache(key_name, value):
cache.set(key_name, json.dumps(value), NOTIFICATION_CACHE_TIME)
lms/djangoapps/open_ended_grading/staff_grading.py deleted 100644 → 0
"""
LMS part of instructor grading:
- views + ajax handling
- calls the instructor grading service
"""
import logging
log = logging.getLogger(__name__)
class StaffGrading(object):
"""
Wrap up functionality for staff grading of submissions--interface exposes get_html, ajax views.
"""
def __init__(self, course):
self.course = course
def get_html(self):
return "<b>Instructor grading!</b>"
# context = {}
# return render_to_string('courseware/instructor_grading_view.html', context)
lms/djangoapps/open_ended_grading/staff_grading_service.py deleted 100644 → 0
"""
This module provides views that proxy to the staff grading backend service.
"""
import json
import logging
from django.conf import settings
from django.http import HttpResponse, Http404
from django.utils.translation import ugettext as _
from opaque_keys.edx.locations import SlashSeparatedCourseKey
from xmodule.open_ended_grading_classes.grading_service_module import GradingService, GradingServiceError
from courseware.access import has_access
from edxmako.shortcuts import render_to_string
from student.models import unique_id_for_user
from open_ended_grading.utils import does_location_exist
import dogstats_wrapper as dog_stats_api
log = logging.getLogger(__name__)
STAFF_ERROR_MESSAGE = _(
u'Could not contact the external grading server. Please contact the '
u'development team at {email}.'
).format(
email=u'<a href="mailto:{tech_support_email}>{tech_support_email}</a>'.format(
tech_support_email=settings.TECH_SUPPORT_EMAIL
)
)
MAX_ALLOWED_FEEDBACK_LENGTH = 5000
class MockStaffGradingService(object):
"""
A simple mockup of a staff grading service, testing.
"""
def __init__(self):
self.cnt = 0
def get_next(self, course_id, location, grader_id):
self.cnt += 1
return {'success': True,
'submission_id': self.cnt,
'submission': 'Test submission {cnt}'.format(cnt=self.cnt),
'num_graded': 3,
'min_for_ml': 5,
'num_pending': 4,
'prompt': 'This is a fake prompt',
'ml_error_info': 'ML info',
'max_score': 2 + self.cnt % 3,
'rubric': 'A rubric'}
def get_problem_list(self, course_id, grader_id):
self.cnt += 1
return {
'success': True,
'problem_list': [
json.dumps({
'location': 'i4x://MITx/3.091x/problem/open_ended_demo1',
'problem_name': "Problem 1",
'num_graded': 3,
'num_pending': 5,
'min_for_ml': 10,
}),
json.dumps({
'location': 'i4x://MITx/3.091x/problem/open_ended_demo2',
'problem_name': "Problem 2",
'num_graded': 1,
'num_pending': 5,
'min_for_ml': 10,
}),
],
}
def save_grade(self, course_id, grader_id, submission_id, score, feedback, skipped, rubric_scores,
submission_flagged):
return self.get_next(course_id, 'fake location', grader_id)
class StaffGradingService(GradingService):
"""
Interface to staff grading backend.
"""
METRIC_NAME = 'edxapp.open_ended_grading.staff_grading_service'
def __init__(self, config):
config['render_template'] = render_to_string
super(StaffGradingService, self).__init__(config)
self.url = config['url'] + config['staff_grading']
self.login_url = self.url + '/login/'
self.get_next_url = self.url + '/get_next_submission/'
self.save_grade_url = self.url + '/save_grade/'
self.get_problem_list_url = self.url + '/get_problem_list/'
self.get_notifications_url = self.url + "/get_notifications/"
def get_problem_list(self, course_id, grader_id):
"""
Get the list of problems for a given course.
Args:
course_id: course id that we want the problems of
grader_id: who is grading this? The anonymous user_id of the grader.
Returns:
dict with the response from the service. (Deliberately not
writing out the fields here--see the docs on the staff_grading view
in the grading_controller repo)
Raises:
GradingServiceError: something went wrong with the connection.
"""
params = {'course_id': course_id.to_deprecated_string(), 'grader_id': grader_id}
result = self.get(self.get_problem_list_url, params)
tags = [u'course_id:{}'.format(course_id)]
self._record_result('get_problem_list', result, tags)
dog_stats_api.histogram(
self._metric_name('get_problem_list.result.length'),
len(result.get('problem_list', []))
)
return result
def get_next(self, course_id, location, grader_id):
"""
Get the next thing to grade.
Args:
course_id: the course that this problem belongs to
location: location of the problem that we are grading and would like the
next submission for
grader_id: who is grading this? The anonymous user_id of the grader.
Returns:
dict with the response from the service. (Deliberately not
writing out the fields here--see the docs on the staff_grading view
in the grading_controller repo)
Raises:
GradingServiceError: something went wrong with the connection.
"""
result = self._render_rubric(
self.get(
self.get_next_url,
params={
'location': location.to_deprecated_string(),
'grader_id': grader_id
}
)
)
tags = [u'course_id:{}'.format(course_id)]
self._record_result('get_next', result, tags)
return result
def save_grade(self, course_id, grader_id, submission_id, score, feedback, skipped, rubric_scores,
submission_flagged):
"""
Save a score and feedback for a submission.
Returns:
dict with keys
'success': bool
'error': error msg, if something went wrong.
Raises:
GradingServiceError if there's a problem connecting.
"""
data = {'course_id': course_id.to_deprecated_string(),
'submission_id': submission_id,
'score': score,
'feedback': feedback,
'grader_id': grader_id,
'skipped': skipped,
'rubric_scores': rubric_scores,
'rubric_scores_complete': True,
'submission_flagged': submission_flagged}
result = self._render_rubric(self.post(self.save_grade_url, data=data))
tags = [u'course_id:{}'.format(course_id)]
self._record_result('save_grade', result, tags)
return result
def get_notifications(self, course_id):
params = {'course_id': course_id.to_deprecated_string()}
result = self.get(self.get_notifications_url, params)
tags = [
u'course_id:{}'.format(course_id),
u'staff_needs_to_grade:{}'.format(result.get('staff_needs_to_grade'))
]
self._record_result('get_notifications', result, tags)
return result
# don't initialize until staff_grading_service() is called--means that just
# importing this file doesn't create objects that may not have the right config
_service = None
def staff_grading_service():
"""
Return a staff grading service instance--if settings.MOCK_STAFF_GRADING is True,
returns a mock one, otherwise a real one.
Caches the result, so changing the setting after the first call to this
function will have no effect.
"""
global _service
if _service is not None:
return _service
if settings.MOCK_STAFF_GRADING:
_service = MockStaffGradingService()
else:
_service = StaffGradingService(settings.OPEN_ENDED_GRADING_INTERFACE)
return _service
def _err_response(msg):
"""
Return a HttpResponse with a json dump with success=False, and the given error message.
"""
return HttpResponse(json.dumps({'success': False, 'error': msg}),
content_type="application/json")
def _check_access(user, course_id):
"""
Raise 404 if user doesn't have staff access to course_id
"""
if not has_access(user, 'staff', course_id):
raise Http404
return
def get_next(request, course_id):
"""
Get the next thing to grade for course_id and with the location specified
in the request.
Returns a json dict with the following keys:
'success': bool
'submission_id': a unique identifier for the submission, to be passed back
with the grade.
'submission': the submission, rendered as read-only html for grading
'rubric': the rubric, also rendered as html.
'message': if there was no submission available, but nothing went wrong,
there will be a message field.
'error': if success is False, will have an error message with more info.
"""
assert isinstance(course_id, basestring)
course_key = SlashSeparatedCourseKey.from_deprecated_string(course_id)
_check_access(request.user, course_key)
required = set(['location'])
if request.method != 'POST':
raise Http404
actual = set(request.POST.keys())
missing = required - actual
if len(missing) > 0:
return _err_response('Missing required keys {0}'.format(
', '.join(missing)))
grader_id = unique_id_for_user(request.user)
p = request.POST
location = course_key.make_usage_key_from_deprecated_string(p['location'])
return HttpResponse(json.dumps(_get_next(course_key, grader_id, location)),
content_type="application/json")
def get_problem_list(request, course_id):
"""
Get all the problems for the given course id
Returns a json dict with the following keys:
success: bool
problem_list: a list containing json dicts with the following keys:
each dict represents a different problem in the course
location: the location of the problem
problem_name: the name of the problem
num_graded: the number of responses that have been graded
num_pending: the number of responses that are sitting in the queue
min_for_ml: the number of responses that need to be graded before
the ml can be run
'error': if success is False, will have an error message with more info.
"""
assert isinstance(course_id, basestring)
course_key = SlashSeparatedCourseKey.from_deprecated_string(course_id)
_check_access(request.user, course_key)
try:
response = staff_grading_service().get_problem_list(course_key, unique_id_for_user(request.user))
# If 'problem_list' is in the response, then we got a list of problems from the ORA server.
# If it is not, then ORA could not find any problems.
if 'problem_list' in response:
problem_list = response['problem_list']
else:
problem_list = []
# Make an error messages to reflect that we could not find anything to grade.
response['error'] = _(
u'Cannot find any open response problems in this course. '
u'Have you submitted answers to any open response assessment questions? '
u'If not, please do so and return to this page.'
)
valid_problem_list = []
for i in xrange(len(problem_list)):
# Needed to ensure that the 'location' key can be accessed.
try:
problem_list[i] = json.loads(problem_list[i])
except Exception:
pass
if does_location_exist(course_key.make_usage_key_from_deprecated_string(problem_list[i]['location'])):
valid_problem_list.append(problem_list[i])
response['problem_list'] = valid_problem_list
response = json.dumps(response)
return HttpResponse(response, content_type="application/json")
except GradingServiceError:
#This is a dev_facing_error
log.exception(
"Error from staff grading service in open "
"ended grading. server url: {0}".format(staff_grading_service().url)
)
#This is a staff_facing_error
return HttpResponse(json.dumps({'success': False,
'error': STAFF_ERROR_MESSAGE}))
def _get_next(course_id, grader_id, location):
"""
Implementation of get_next (also called from save_grade) -- returns a json string
"""
try:
return staff_grading_service().get_next(course_id, location, grader_id)
except GradingServiceError:
#This is a dev facing error
log.exception(
"Error from staff grading service in open "
"ended grading. server url: {0}".format(staff_grading_service().url)
)
#This is a staff_facing_error
return json.dumps({'success': False,
'error': STAFF_ERROR_MESSAGE})
def save_grade(request, course_id):
"""
Save the grade and feedback for a submission, and, if all goes well, return
the next thing to grade.
Expects the following POST parameters:
'score': int
'feedback': string
'submission_id': int
Returns the same thing as get_next, except that additional error messages
are possible if something goes wrong with saving the grade.
"""
course_key = SlashSeparatedCourseKey.from_deprecated_string(course_id)
_check_access(request.user, course_key)
if request.method != 'POST':
raise Http404
p = request.POST
required = set(['score', 'feedback', 'submission_id', 'location', 'submission_flagged'])
skipped = 'skipped' in p
#If the instructor has skipped grading the submission, then there will not be any rubric scores.
#Only add in the rubric scores if the instructor has not skipped.
if not skipped:
required.add('rubric_scores[]')
actual = set(p.keys())
missing = required - actual
if len(missing) > 0:
return _err_response('Missing required keys {0}'.format(
', '.join(missing)))
success, message = check_feedback_length(p)
if not success:
return _err_response(message)
grader_id = unique_id_for_user(request.user)
location = course_key.make_usage_key_from_deprecated_string(p['location'])
try:
result = staff_grading_service().save_grade(course_key,
grader_id,
p['submission_id'],
p['score'],
p['feedback'],
skipped,
p.getlist('rubric_scores[]'),
p['submission_flagged'])
except GradingServiceError:
#This is a dev_facing_error
log.exception(
"Error saving grade in the staff grading interface in open ended grading. Request: {0} Course ID: {1}".format(
request, course_id))
#This is a staff_facing_error
return _err_response(STAFF_ERROR_MESSAGE)
except ValueError:
#This is a dev_facing_error
log.exception(
"save_grade returned broken json in the staff grading interface in open ended grading: {0}".format(
result_json))
#This is a staff_facing_error
return _err_response(STAFF_ERROR_MESSAGE)
if not result.get('success', False):
#This is a dev_facing_error
log.warning(
'Got success=False from staff grading service in open ended grading. Response: {0}'.format(result_json))
return _err_response(STAFF_ERROR_MESSAGE)
# Ok, save_grade seemed to work. Get the next submission to grade.
return HttpResponse(json.dumps(_get_next(course_id, grader_id, location)),
content_type="application/json")
def check_feedback_length(data):
feedback = data.get("feedback")
if feedback and len(feedback) > MAX_ALLOWED_FEEDBACK_LENGTH:
return False, "Feedback is too long, Max length is {0} characters.".format(
MAX_ALLOWED_FEEDBACK_LENGTH
)
else:
return True, ""
lms/djangoapps/open_ended_grading/tests.py deleted 100644 → 0
"""
Tests for open ended grading interfaces
./manage.py lms --settings test test lms/djangoapps/open_ended_grading
"""
import ddt
import json
import logging
from django.conf import settings
from django.contrib.auth.models import User
from django.core.urlresolvers import reverse
from django.test import RequestFactory
from edxmako.shortcuts import render_to_string
from edxmako.tests import mako_middleware_process_request
from mock import MagicMock, patch, Mock
from opaque_keys.edx.locations import SlashSeparatedCourseKey
from xblock.field_data import DictFieldData
from xblock.fields import ScopeIds
from config_models.models import cache
from courseware.tests import factories
from courseware.tests.helpers import LoginEnrollmentTestCase
from lms.djangoapps.lms_xblock.runtime import LmsModuleSystem
from student.roles import CourseStaffRole
from student.models import unique_id_for_user
from xblock_django.models import XBlockDisableConfig
from xmodule import peer_grading_module
from xmodule.error_module import ErrorDescriptor
from xmodule.modulestore.django import modulestore
from xmodule.modulestore.tests.django_utils import TEST_DATA_MIXED_TOY_MODULESTORE, ModuleStoreTestCase
from xmodule.modulestore.tests.factories import CourseFactory
from xmodule.modulestore.xml_importer import import_course_from_xml
from xmodule.open_ended_grading_classes import peer_grading_service, controller_query_service
from xmodule.tests import test_util_open_ended
from open_ended_grading import staff_grading_service, views, utils
TEST_DATA_DIR = settings.COMMON_TEST_DATA_ROOT
log = logging.getLogger(__name__)
class EmptyStaffGradingService(object):
"""
A staff grading service that does not return a problem list from get_problem_list.
Used for testing to see if error message for empty problem list is correctly displayed.
"""
def get_problem_list(self, course_id, user_id):
"""
Return a staff grading response that is missing a problem list key.
"""
return {'success': True, 'error': 'No problems found.'}
def make_instructor(course, user_email):
"""
Makes a given user an instructor in a course.
"""
CourseStaffRole(course.id).add_users(User.objects.get(email=user_email))
class StudentProblemListMockQuery(object):
"""
Mock controller query service for testing student problem list functionality.
"""
def get_grading_status_list(self, *args, **kwargs):
"""
Get a mock grading status list with locations from the open_ended test course.
@returns: grading status message dictionary.
"""
return {
"version": 1,
"problem_list": [
{
"problem_name": "Test1",
"grader_type": "IN",
"eta_available": True,
"state": "Finished",
"eta": 259200,
"location": "i4x://edX/open_ended/combinedopenended/SampleQuestion1Attempt"
},
{
"problem_name": "Test2",
"grader_type": "NA",
"eta_available": True,
"state": "Waiting to be Graded",
"eta": 259200,
"location": "i4x://edX/open_ended/combinedopenended/SampleQuestion"
},
{
"problem_name": "Test3",
"grader_type": "PE",
"eta_available": True,
"state": "Waiting to be Graded",
"eta": 259200,
"location": "i4x://edX/open_ended/combinedopenended/SampleQuestion454"
},
],
"success": True
}
class TestStaffGradingService(ModuleStoreTestCase, LoginEnrollmentTestCase):
'''
Check that staff grading service proxy works. Basically just checking the
access control and error handling logic -- all the actual work is on the
backend.
'''
MODULESTORE = TEST_DATA_MIXED_TOY_MODULESTORE
def setUp(self):
super(TestStaffGradingService, self).setUp()
self.student = 'view@test.com'
self.instructor = 'view2@test.com'
self.password = 'foo'
self.create_account('u1', self.student, self.password)
self.create_account('u2', self.instructor, self.password)
self.activate_user(self.student)
self.activate_user(self.instructor)
self.course_id = SlashSeparatedCourseKey("edX", "toy", "2012_Fall")
self.location_string = self.course_id.make_usage_key('html', 'TestLocation').to_deprecated_string()
self.toy = modulestore().get_course(self.course_id)
make_instructor(self.toy, self.instructor)
self.mock_service = staff_grading_service.staff_grading_service()
self.logout()
def test_access(self):
"""
Make sure only staff have access.
"""
self.login(self.student, self.password)
# both get and post should return 404
for view_name in ('staff_grading_get_next', 'staff_grading_save_grade'):
url = reverse(view_name, kwargs={'course_id': self.course_id.to_deprecated_string()})
self.assert_request_status_code(404, url, method="GET")
self.assert_request_status_code(404, url, method="POST")
def test_get_next(self):
self.login(self.instructor, self.password)
url = reverse('staff_grading_get_next', kwargs={'course_id': self.course_id.to_deprecated_string()})
data = {'location': self.location_string}
response = self.assert_request_status_code(200, url, method="POST", data=data)
content = json.loads(response.content)
self.assertTrue(content['success'])
self.assertEquals(content['submission_id'], self.mock_service.cnt)
self.assertIsNotNone(content['submission'])
self.assertIsNotNone(content['num_graded'])
self.assertIsNotNone(content['min_for_ml'])
self.assertIsNotNone(content['num_pending'])
self.assertIsNotNone(content['prompt'])
self.assertIsNotNone(content['ml_error_info'])
self.assertIsNotNone(content['max_score'])
self.assertIsNotNone(content['rubric'])
def save_grade_base(self, skip=False):
self.login(self.instructor, self.password)
url = reverse('staff_grading_save_grade', kwargs={'course_id': self.course_id.to_deprecated_string()})
data = {'score': '12',
'feedback': 'great!',
'submission_id': '123',
'location': self.location_string,
'submission_flagged': "true",
'rubric_scores[]': ['1', '2']}
if skip:
data.update({'skipped': True})
response = self.assert_request_status_code(200, url, method="POST", data=data)
content = json.loads(response.content)
self.assertTrue(content['success'], str(content))
self.assertEquals(content['submission_id'], self.mock_service.cnt)
def test_save_grade(self):
self.save_grade_base(skip=False)
def test_save_grade_skip(self):
self.save_grade_base(skip=True)
def test_get_problem_list(self):
self.login(self.instructor, self.password)
url = reverse('staff_grading_get_problem_list', kwargs={'course_id': self.course_id.to_deprecated_string()})
data = {}
response = self.assert_request_status_code(200, url, method="POST", data=data)
content = json.loads(response.content)
self.assertTrue(content['success'])
self.assertEqual(content['problem_list'], [])
@patch('open_ended_grading.staff_grading_service._service', EmptyStaffGradingService())
def test_get_problem_list_missing(self):
"""
Test to see if a staff grading response missing a problem list is given the appropriate error.
Mock the staff grading service to enable the key to be missing.
"""
# Get a valid user object.
instructor = User.objects.get(email=self.instructor)
# Mock a request object.
request = Mock(
user=instructor,
)
# Get the response and load its content.
response = json.loads(staff_grading_service.get_problem_list(request, self.course_id.to_deprecated_string()).content)
# A valid response will have an "error" key.
self.assertTrue('error' in response)
# Check that the error text is correct.
self.assertIn("Cannot find", response['error'])
def test_save_grade_with_long_feedback(self):
"""
Test if feedback is too long save_grade() should return error message.
"""
self.login(self.instructor, self.password)
url = reverse('staff_grading_save_grade', kwargs={'course_id': self.course_id.to_deprecated_string()})
data = {
'score': '12',
'feedback': '',
'submission_id': '123',
'location': self.location_string,
'submission_flagged': "false",
'rubric_scores[]': ['1', '2']
}
feedback_fragment = "This is very long feedback."
data["feedback"] = feedback_fragment * (
(staff_grading_service.MAX_ALLOWED_FEEDBACK_LENGTH / len(feedback_fragment) + 1)
)
response = self.assert_request_status_code(200, url, method="POST", data=data)
content = json.loads(response.content)
# Should not succeed.
self.assertEquals(content['success'], False)
self.assertEquals(
content['error'],
"Feedback is too long, Max length is {0} characters.".format(
staff_grading_service.MAX_ALLOWED_FEEDBACK_LENGTH
)
)
class TestPeerGradingService(ModuleStoreTestCase, LoginEnrollmentTestCase):
'''
Check that staff grading service proxy works. Basically just checking the
access control and error handling logic -- all the actual work is on the
backend.
'''
def setUp(self):
super(TestPeerGradingService, self).setUp()
self.student = 'view@test.com'
self.instructor = 'view2@test.com'
self.password = 'foo'
self.create_account('u1', self.student, self.password)
self.create_account('u2', self.instructor, self.password)
self.activate_user(self.student)
self.activate_user(self.instructor)
self.course_id = SlashSeparatedCourseKey("edX", "toy", "2012_Fall")
self.location_string = self.course_id.make_usage_key('html', 'TestLocation').to_deprecated_string()
self.toy = modulestore().get_course(self.course_id)
location = "i4x://edX/toy/peergrading/init"
field_data = DictFieldData({'data': "<peergrading/>", 'location': location, 'category': 'peergrading'})
self.mock_service = peer_grading_service.MockPeerGradingService()
self.system = LmsModuleSystem(
static_url=settings.STATIC_URL,
track_function=None,
get_module=None,
render_template=render_to_string,
replace_urls=None,
s3_interface=test_util_open_ended.S3_INTERFACE,
open_ended_grading_interface=test_util_open_ended.OPEN_ENDED_GRADING_INTERFACE,
mixins=settings.XBLOCK_MIXINS,
error_descriptor_class=ErrorDescriptor,
descriptor_runtime=None,
)
self.descriptor = peer_grading_module.PeerGradingDescriptor(self.system, field_data, ScopeIds(None, None, None, None))
self.descriptor.xmodule_runtime = self.system
self.peer_module = self.descriptor
self.peer_module.peer_gs = self.mock_service
self.logout()
def test_get_next_submission_success(self):
data = {'location': self.location_string}
response = self.peer_module.get_next_submission(data)
content = response
self.assertTrue(content['success'])
self.assertIsNotNone(content['submission_id'])
self.assertIsNotNone(content['prompt'])
self.assertIsNotNone(content['submission_key'])
self.assertIsNotNone(content['max_score'])
def test_get_next_submission_missing_location(self):
data = {}
d = self.peer_module.get_next_submission(data)
self.assertFalse(d['success'])
self.assertEqual(d['error'], "Missing required keys: location")
def test_save_grade_success(self):
data = {
'rubric_scores[]': [0, 0],
'location': self.location_string,
'submission_id': 1,
'submission_key': 'fake key',
'score': 2,
'feedback': 'feedback',
'submission_flagged': 'false',
'answer_unknown': 'false',
'rubric_scores_complete': 'true'
}
qdict = MagicMock()
def fake_get_item(key):
return data[key]
qdict.__getitem__.side_effect = fake_get_item
qdict.getlist = fake_get_item
qdict.keys = data.keys
response = self.peer_module.save_grade(qdict)
self.assertTrue(response['success'])
def test_save_grade_missing_keys(self):
data = {}
d = self.peer_module.save_grade(data)
self.assertFalse(d['success'])
self.assertTrue(d['error'].find('Missing required keys:') > -1)
def test_is_calibrated_success(self):
data = {'location': self.location_string}
response = self.peer_module.is_student_calibrated(data)
self.assertTrue(response['success'])
self.assertTrue('calibrated' in response)
def test_is_calibrated_failure(self):
data = {}
response = self.peer_module.is_student_calibrated(data)
self.assertFalse(response['success'])
self.assertFalse('calibrated' in response)
def test_show_calibration_essay_success(self):
data = {'location': self.location_string}
response = self.peer_module.show_calibration_essay(data)
self.assertTrue(response['success'])
self.assertIsNotNone(response['submission_id'])
self.assertIsNotNone(response['prompt'])
self.assertIsNotNone(response['submission_key'])
self.assertIsNotNone(response['max_score'])
def test_show_calibration_essay_missing_key(self):
data = {}
response = self.peer_module.show_calibration_essay(data)
self.assertFalse(response['success'])
self.assertEqual(response['error'], "Missing required keys: location")
def test_save_calibration_essay_success(self):
data = {
'rubric_scores[]': [0, 0],
'location': self.location_string,
'submission_id': 1,
'submission_key': 'fake key',
'score': 2,
'feedback': 'feedback',
'submission_flagged': 'false'
}
qdict = MagicMock()
def fake_get_item(key):
return data[key]
qdict.__getitem__.side_effect = fake_get_item
qdict.getlist = fake_get_item
qdict.keys = data.keys
response = self.peer_module.save_calibration_essay(qdict)
self.assertTrue(response['success'])
self.assertTrue('actual_score' in response)
def test_save_calibration_essay_missing_keys(self):
data = {}
response = self.peer_module.save_calibration_essay(data)
self.assertFalse(response['success'])
self.assertTrue(response['error'].find('Missing required keys:') > -1)
self.assertFalse('actual_score' in response)
def test_save_grade_with_long_feedback(self):
"""
Test if feedback is too long save_grade() should return error message.
"""
data = {
'rubric_scores[]': [0, 0],
'location': self.location_string,
'submission_id': 1,
'submission_key': 'fake key',
'score': 2,
'feedback': '',
'submission_flagged': 'false',
'answer_unknown': 'false',
'rubric_scores_complete': 'true'
}
feedback_fragment = "This is very long feedback."
data["feedback"] = feedback_fragment * (
(staff_grading_service.MAX_ALLOWED_FEEDBACK_LENGTH / len(feedback_fragment) + 1)
)
response_dict = self.peer_module.save_grade(data)
# Should not succeed.
self.assertEquals(response_dict['success'], False)
self.assertEquals(
response_dict['error'],
"Feedback is too long, Max length is {0} characters.".format(
staff_grading_service.MAX_ALLOWED_FEEDBACK_LENGTH
)
)
class TestPanel(ModuleStoreTestCase):
"""
Run tests on the open ended panel
"""
def setUp(self):
super(TestPanel, self).setUp()
self.user = factories.UserFactory()
store = modulestore()
course_items = import_course_from_xml(store, self.user.id, TEST_DATA_DIR, ['open_ended']) # pylint: disable=maybe-no-member
self.course = course_items[0]
self.course_key = self.course.id
def test_open_ended_panel(self):
"""
Test to see if the peer grading module in the demo course is found
@return:
"""
found_module, peer_grading_module = views.find_peer_grading_module(self.course)
self.assertTrue(found_module)
@patch(
'open_ended_grading.utils.create_controller_query_service',
Mock(
return_value=controller_query_service.MockControllerQueryService(
settings.OPEN_ENDED_GRADING_INTERFACE,
utils.render_to_string
)
)
)
def test_problem_list(self):
"""
Ensure that the problem list from the grading controller server can be rendered properly locally
@return:
"""
request = RequestFactory().get(
reverse("open_ended_problems", kwargs={'course_id': self.course_key})
)
request.user = self.user
mako_middleware_process_request(request)
response = views.student_problem_list(request, self.course.id.to_deprecated_string())
self.assertRegexpMatches(response.content, "Here is a list of open ended problems for this course.")
class TestPeerGradingFound(ModuleStoreTestCase):
"""
Test to see if peer grading modules can be found properly.
"""
def setUp(self):
super(TestPeerGradingFound, self).setUp()
self.user = factories.UserFactory()
store = modulestore()
course_items = import_course_from_xml(store, self.user.id, TEST_DATA_DIR, ['open_ended_nopath']) # pylint: disable=maybe-no-member
self.course = course_items[0]
self.course_key = self.course.id
def test_peer_grading_nopath(self):
"""
The open_ended_nopath course contains a peer grading module with no path to it.
Ensure that the exception is caught.
"""
found, url = views.find_peer_grading_module(self.course)
self.assertEqual(found, False)
class TestStudentProblemList(ModuleStoreTestCase):
"""
Test if the student problem list correctly fetches and parses problems.
"""
def setUp(self):
super(TestStudentProblemList, self).setUp()
# Load an open ended course with several problems.
self.user = factories.UserFactory()
store = modulestore()
course_items = import_course_from_xml(store, self.user.id, TEST_DATA_DIR, ['open_ended']) # pylint: disable=maybe-no-member
self.course = course_items[0]
self.course_key = self.course.id
# Enroll our user in our course and make them an instructor.
make_instructor(self.course, self.user.email)
@patch(
'open_ended_grading.utils.create_controller_query_service',
Mock(return_value=StudentProblemListMockQuery())
)
def test_get_problem_list(self):
"""
Test to see if the StudentProblemList class can get and parse a problem list from ORA.
Mock the get_grading_status_list function using StudentProblemListMockQuery.
"""
# Initialize a StudentProblemList object.
student_problem_list = utils.StudentProblemList(self.course.id, unique_id_for_user(self.user))
# Get the initial problem list from ORA.
success = student_problem_list.fetch_from_grading_service()
# Should be successful, and we should have three problems. See mock class for details.
self.assertTrue(success)
self.assertEqual(len(student_problem_list.problem_list), 3)
# See if the problem locations are valid.
valid_problems = student_problem_list.add_problem_data(reverse('courses'))
# One location is invalid, so we should now have two.
self.assertEqual(len(valid_problems), 2)
# Ensure that human names are being set properly.
self.assertEqual(valid_problems[0]['grader_type_display_name'], "Instructor Assessment")
@ddt.ddt
class TestTabs(ModuleStoreTestCase):
"""
Test tabs.
"""
def setUp(self):
super(TestTabs, self).setUp()
self.course = CourseFactory(advanced_modules=('combinedopenended'))
self.addCleanup(lambda: self._enable_xblock_disable_config(False))
def _enable_xblock_disable_config(self, enabled):
""" Enable or disable xblocks disable. """
config = XBlockDisableConfig.current()
config.enabled = enabled
config.disabled_blocks = "\n".join(('combinedopenended', 'peergrading'))
config.save()
cache.clear()
@ddt.data(
views.StaffGradingTab,
views.PeerGradingTab,
views.OpenEndedGradingTab,
)
def test_tabs_enabled(self, tab):
self.assertTrue(tab.is_enabled(self.course))
@ddt.data(
views.StaffGradingTab,
views.PeerGradingTab,
views.OpenEndedGradingTab,
)
def test_tabs_disabled(self, tab):
self._enable_xblock_disable_config(True)
self.assertFalse(tab.is_enabled(self.course))
lms/djangoapps/open_ended_grading/utils.py deleted 100644 → 0
import logging
from urllib import urlencode
from xmodule.modulestore import search
from xmodule.modulestore.django import modulestore
from xmodule.modulestore.exceptions import ItemNotFoundError, NoPathToItem
from xmodule.open_ended_grading_classes.controller_query_service import ControllerQueryService
from xmodule.open_ended_grading_classes.grading_service_module import GradingServiceError
from django.utils.translation import ugettext as _
from django.conf import settings
from edxmako.shortcuts import render_to_string
log = logging.getLogger(__name__)
GRADER_DISPLAY_NAMES = {
'ML': _("AI Assessment"),
'PE': _("Peer Assessment"),
'NA': _("Not yet available"),
'BC': _("Automatic Checker"),
'IN': _("Instructor Assessment"),
}
STUDENT_ERROR_MESSAGE = _("Error occurred while contacting the grading service. Please notify course staff.")
STAFF_ERROR_MESSAGE = _("Error occurred while contacting the grading service. Please notify your edX point of contact.")
def generate_problem_url(problem_url_parts, base_course_url):
"""
From a list of problem url parts generated by search.path_to_location and a base course url, generates a url to a problem
@param problem_url_parts: Output of search.path_to_location
@param base_course_url: Base url of a given course
@return: A path to the problem
"""
activate_block_id = problem_url_parts[-1]
problem_url_parts = problem_url_parts[0:-1]
problem_url = base_course_url + "/"
for i, part in enumerate(problem_url_parts):
if part is not None:
# This is the course_key. We need to turn it into its deprecated
# form.
if i == 0:
part = part.to_deprecated_string()
# This is placed between the course id and the rest of the url.
if i == 1:
problem_url += "courseware/"
problem_url += part + "/"
problem_url += '?{}'.format(urlencode({'activate_block_id': unicode(activate_block_id)}))
return problem_url
def does_location_exist(usage_key):
"""
Checks to see if a valid module exists at a given location (ie has not been deleted)
course_id - string course id
location - string location
"""
try:
search.path_to_location(modulestore(), usage_key)
return True
except ItemNotFoundError:
# If the problem cannot be found at the location received from the grading controller server,
# it has been deleted by the course author.
return False
except NoPathToItem:
# If the problem can be found, but there is no path to it, then we assume it is a draft.
# Log a warning in any case.
log.warn("Got an unexpected NoPathToItem error in staff grading with location %s. "
"This is ok if it is a draft; ensure that the location is valid.", usage_key)
return False
def create_controller_query_service():
"""
Return an instance of a service that can query edX ORA.
"""
return ControllerQueryService(settings.OPEN_ENDED_GRADING_INTERFACE, render_to_string)
class StudentProblemList(object):
"""
Get a list of problems that the student has attempted from ORA.
Add in metadata as needed.
"""
def __init__(self, course_id, user_id):
"""
@param course_id: The id of a course object. Get using course.id.
@param user_id: The anonymous id of the user, from the unique_id_for_user function.
"""
self.course_id = course_id
self.user_id = user_id
# We want to append this string to all of our error messages.
self.course_error_ending = _("for course {0} and student {1}.").format(self.course_id, user_id)
# This is our generic error message.
self.error_text = STUDENT_ERROR_MESSAGE
self.success = False
# Create a service to query edX ORA.
self.controller_qs = create_controller_query_service()
def fetch_from_grading_service(self):
"""
Fetch a list of problems that the student has answered from ORA.
Handle various error conditions.
@return: A boolean success indicator.
"""
# In the case of multiple calls, ensure that success is false initially.
self.success = False
try:
#Get list of all open ended problems that the grading server knows about
problem_list_dict = self.controller_qs.get_grading_status_list(self.course_id, self.user_id)
except GradingServiceError:
log.error("Problem contacting open ended grading service " + self.course_error_ending)
return self.success
except ValueError:
log.error("Problem with results from external grading service for open ended" + self.course_error_ending)
return self.success
success = problem_list_dict['success']
if 'error' in problem_list_dict:
self.error_text = problem_list_dict['error']
return success
if 'problem_list' not in problem_list_dict:
log.error("Did not receive a problem list in ORA response" + self.course_error_ending)
return success
self.problem_list = problem_list_dict['problem_list']
self.success = True
return self.success
def add_problem_data(self, base_course_url):
"""
Add metadata to problems.
@param base_course_url: the base url for any course. Can get with reverse('course')
@return: A list of valid problems in the course and their appended data.
"""
# Our list of valid problems.
valid_problems = []
if not self.success or not isinstance(self.problem_list, list):
log.error("Called add_problem_data without a valid problem list" + self.course_error_ending)
return valid_problems
# Iterate through all of our problems and add data.
for problem in self.problem_list:
try:
# Try to load the problem.
usage_key = self.course_id.make_usage_key_from_deprecated_string(problem['location'])
problem_url_parts = search.path_to_location(modulestore(), usage_key)
except (ItemNotFoundError, NoPathToItem):
# If the problem cannot be found at the location received from the grading controller server,
# it has been deleted by the course author. We should not display it.
error_message = "Could not find module for course {0} at location {1}".format(self.course_id,
problem['location'])
log.error(error_message)
continue
# Get the problem url in the courseware.
problem_url = generate_problem_url(problem_url_parts, base_course_url)
# Map the grader name from ORA to a human readable version.
grader_type_display_name = GRADER_DISPLAY_NAMES.get(problem['grader_type'], "edX Assessment")
problem['actual_url'] = problem_url
problem['grader_type_display_name'] = grader_type_display_name
valid_problems.append(problem)
return valid_problems
lms/djangoapps/open_ended_grading/views.py deleted 100644 → 0
import logging
from django.views.decorators.cache import cache_control
from edxmako.shortcuts import render_to_response
from django.core.urlresolvers import reverse
from courseware.courses import get_course_with_access
from courseware.access import has_access
from courseware.tabs import EnrolledTab
from xmodule.open_ended_grading_classes.grading_service_module import GradingServiceError
import json
from student.models import unique_id_for_user
from open_ended_grading import open_ended_notifications
from xmodule.modulestore.django import modulestore
from xmodule.modulestore import search
from opaque_keys.edx.locations import SlashSeparatedCourseKey
from xmodule.modulestore.exceptions import NoPathToItem
from django.http import HttpResponse, Http404, HttpResponseRedirect
from django.utils.translation import ugettext as _
from open_ended_grading.utils import (
STAFF_ERROR_MESSAGE, StudentProblemList, generate_problem_url, create_controller_query_service
)
from xblock_django.models import XBlockDisableConfig
log = logging.getLogger(__name__)
def _reverse_with_slash(url_name, course_key):
"""
Reverses the URL given the name and the course id, and then adds a trailing slash if
it does not exist yet.
@param url_name: The name of the url (eg 'staff_grading').
@param course_id: The id of the course object (eg course.id).
@returns: The reversed url with a trailing slash.
"""
ajax_url = _reverse_without_slash(url_name, course_key)
if not ajax_url.endswith('/'):
ajax_url += '/'
return ajax_url
def _reverse_without_slash(url_name, course_key):
course_id = course_key.to_deprecated_string()
ajax_url = reverse(url_name, kwargs={'course_id': course_id})
return ajax_url
DESCRIPTION_DICT = {
'Peer Grading': _("View all problems that require peer assessment in this particular course."),
'Staff Grading': _("View ungraded submissions submitted by students for the open ended problems in the course."),
'Problems you have submitted': _("View open ended problems that you have previously submitted for grading."),
'Flagged Submissions': _("View submissions that have been flagged by students as inappropriate."),
}
ALERT_DICT = {
'Peer Grading': _("New submissions to grade"),
'Staff Grading': _("New submissions to grade"),
'Problems you have submitted': _("New grades have been returned"),
'Flagged Submissions': _("Submissions have been flagged for review"),
}
class StaffGradingTab(EnrolledTab):
"""
A tab for staff grading.
"""
type = 'staff_grading'
title = _("Staff grading")
view_name = "staff_grading"
@classmethod
def is_enabled(cls, course, user=None):
if XBlockDisableConfig.is_block_type_disabled('combinedopenended'):
return False
if user and not has_access(user, 'staff', course, course.id):
return False
return "combinedopenended" in course.advanced_modules
class PeerGradingTab(EnrolledTab):
"""
A tab for peer grading.
"""
type = 'peer_grading'
# Translators: "Peer grading" appears on a tab that allows
# students to view open-ended problems that require grading
title = _("Peer grading")
view_name = "peer_grading"
@classmethod
def is_enabled(cls, course, user=None):
if XBlockDisableConfig.is_block_type_disabled('combinedopenended'):
return False
if not super(PeerGradingTab, cls).is_enabled(course, user=user):
return False
return "combinedopenended" in course.advanced_modules
class OpenEndedGradingTab(EnrolledTab):
"""
A tab for open ended grading.
"""
type = 'open_ended'
# Translators: "Open Ended Panel" appears on a tab that, when clicked, opens up a panel that
# displays information about open-ended problems that a user has submitted or needs to grade
title = _("Open Ended Panel")
view_name = "open_ended_notifications"
@classmethod
def is_enabled(cls, course, user=None):
if XBlockDisableConfig.is_block_type_disabled('combinedopenended'):
return False
if not super(OpenEndedGradingTab, cls).is_enabled(course, user=user):
return False
return "combinedopenended" in course.advanced_modules
@cache_control(no_cache=True, no_store=True, must_revalidate=True)
def staff_grading(request, course_id):
"""
Show the instructor grading interface.
"""
course_key = SlashSeparatedCourseKey.from_deprecated_string(course_id)
course = get_course_with_access(request.user, 'staff', course_key)
ajax_url = _reverse_with_slash('staff_grading', course_key)
return render_to_response('instructor/staff_grading.html', {
'course': course,
'course_id': course_id,
'ajax_url': ajax_url,
# Checked above
'staff_access': True, })
def find_peer_grading_module(course):
"""
Given a course, finds the first peer grading module in it.
@param course: A course object.
@return: boolean found_module, string problem_url
"""
# Reverse the base course url.
base_course_url = reverse('courses')
found_module = False
problem_url = ""
# Get the peer grading modules currently in the course. Explicitly specify the course id to avoid issues with different runs.
items = modulestore().get_items(course.id, qualifiers={'category': 'peergrading'})
# See if any of the modules are centralized modules (ie display info from multiple problems)
items = [i for i in items if not getattr(i, "use_for_single_location", True)]
# Loop through all potential peer grading modules, and find the first one that has a path to it.
for item in items:
# Generate a url for the first module and redirect the user to it.
try:
problem_url_parts = search.path_to_location(modulestore(), item.location)
except NoPathToItem:
# In the case of nopathtoitem, the peer grading module that was found is in an invalid state, and
# can no longer be accessed. Log an informational message, but this will not impact normal behavior.
log.info(u"Invalid peer grading module location %s in course %s. This module may need to be removed.", item.location, course.id)
continue
problem_url = generate_problem_url(problem_url_parts, base_course_url)
found_module = True
return found_module, problem_url
@cache_control(no_cache=True, no_store=True, must_revalidate=True)
def peer_grading(request, course_id):
'''
When a student clicks on the "peer grading" button in the open ended interface, link them to a peer grading
xmodule in the course.
'''
course_key = SlashSeparatedCourseKey.from_deprecated_string(course_id)
#Get the current course
course = get_course_with_access(request.user, 'load', course_key)
found_module, problem_url = find_peer_grading_module(course)
if not found_module:
error_message = _("""
Error with initializing peer grading.
There has not been a peer grading module created in the courseware that would allow you to grade others.
Please check back later for this.
""")
log.exception(error_message + u"Current course is: {0}".format(course_id))
return HttpResponse(error_message)
return HttpResponseRedirect(problem_url)
@cache_control(no_cache=True, no_store=True, must_revalidate=True)
def student_problem_list(request, course_id):
"""
Show a list of problems they have attempted to a student.
Fetch the list from the grading controller server and append some data.
@param request: The request object for this view.
@param course_id: The id of the course to get the problem list for.
@return: Renders an HTML problem list table.
"""
assert isinstance(course_id, basestring)
course_key = SlashSeparatedCourseKey.from_deprecated_string(course_id)
# Load the course. Don't catch any errors here, as we want them to be loud.
course = get_course_with_access(request.user, 'load', course_key)
# The anonymous student id is needed for communication with ORA.
student_id = unique_id_for_user(request.user)
base_course_url = reverse('courses')
error_text = ""
student_problem_list = StudentProblemList(course_key, student_id)
# Get the problem list from ORA.
success = student_problem_list.fetch_from_grading_service()
# If we fetched the problem list properly, add in additional problem data.
if success:
# Add in links to problems.
valid_problems = student_problem_list.add_problem_data(base_course_url)
else:
# Get an error message to show to the student.
valid_problems = []
error_text = student_problem_list.error_text
ajax_url = _reverse_with_slash('open_ended_problems', course_key)
context = {
'course': course,
'course_id': course_key.to_deprecated_string(),
'ajax_url': ajax_url,
'success': success,
'problem_list': valid_problems,
'error_text': error_text,
# Checked above
'staff_access': False,
}
return render_to_response('open_ended_problems/open_ended_problems.html', context)
@cache_control(no_cache=True, no_store=True, must_revalidate=True)
def flagged_problem_list(request, course_id):
'''
Show a student problem list
'''
course_key = SlashSeparatedCourseKey.from_deprecated_string(course_id)
course = get_course_with_access(request.user, 'staff', course_key)
# call problem list service
success = False
error_text = ""
problem_list = []
# Make a service that can query edX ORA.
controller_qs = create_controller_query_service()
try:
problem_list_dict = controller_qs.get_flagged_problem_list(course_key)
success = problem_list_dict['success']
if 'error' in problem_list_dict:
error_text = problem_list_dict['error']
problem_list = []
else:
problem_list = problem_list_dict['flagged_submissions']
except GradingServiceError:
#This is a staff_facing_error
error_text = STAFF_ERROR_MESSAGE
#This is a dev_facing_error
log.error("Could not get flagged problem list from external grading service for open ended.")
success = False
# catch error if if the json loads fails
except ValueError:
#This is a staff_facing_error
error_text = STAFF_ERROR_MESSAGE
#This is a dev_facing_error
log.error("Could not parse problem list from external grading service response.")
success = False
ajax_url = _reverse_with_slash('open_ended_flagged_problems', course_key)
context = {
'course': course,
'course_id': course_id,
'ajax_url': ajax_url,
'success': success,
'problem_list': problem_list,
'error_text': error_text,
# Checked above
'staff_access': True,
}
return render_to_response('open_ended_problems/open_ended_flagged_problems.html', context)
@cache_control(no_cache=True, no_store=True, must_revalidate=True)
def combined_notifications(request, course_id):
"""
Gets combined notifications from the grading controller and displays them
"""
course_key = SlashSeparatedCourseKey.from_deprecated_string(course_id)
course = get_course_with_access(request.user, 'load', course_key)
user = request.user
notifications = open_ended_notifications.combined_notifications(course, user)
response = notifications['response']
notification_tuples = open_ended_notifications.NOTIFICATION_TYPES
notification_list = []
for response_num in xrange(len(notification_tuples)):
tag = notification_tuples[response_num][0]
if tag in response:
url_name = notification_tuples[response_num][1]
human_name = notification_tuples[response_num][2]
url = _reverse_without_slash(url_name, course_key)
has_img = response[tag]
# check to make sure we have descriptions and alert messages
if human_name in DESCRIPTION_DICT:
description = DESCRIPTION_DICT[human_name]
else:
description = ""
if human_name in ALERT_DICT:
alert_message = ALERT_DICT[human_name]
else:
alert_message = ""
notification_item = {
'url': url,
'name': human_name,
'alert': has_img,
'description': description,
'alert_message': alert_message
}
#The open ended panel will need to link the "peer grading" button in the panel to a peer grading
#xmodule defined in the course. This checks to see if the human name of the server notification
#that we are currently processing is "peer grading". If it is, it looks for a peer grading
#module in the course. If none exists, it removes the peer grading item from the panel.
if human_name == "Peer Grading":
found_module, problem_url = find_peer_grading_module(course)
if found_module:
notification_list.append(notification_item)
else:
notification_list.append(notification_item)
ajax_url = _reverse_with_slash('open_ended_notifications', course_key)
combined_dict = {
'error_text': "",
'notification_list': notification_list,
'course': course,
'success': True,
'ajax_url': ajax_url,
}
return render_to_response('open_ended_problems/combined_notifications.html', combined_dict)
@cache_control(no_cache=True, no_store=True, must_revalidate=True)
def take_action_on_flags(request, course_id):
"""
Takes action on student flagged submissions.
Currently, only support unflag and ban actions.
"""
course_key = SlashSeparatedCourseKey.from_deprecated_string(course_id)
if request.method != 'POST':
raise Http404
required = ['submission_id', 'action_type', 'student_id']
for key in required:
if key not in request.POST:
error_message = u'Missing key {0} from submission. Please reload and try again.'.format(key)
response = {
'success': False,
'error': STAFF_ERROR_MESSAGE + error_message
}
return HttpResponse(json.dumps(response), content_type="application/json")
p = request.POST
submission_id = p['submission_id']
action_type = p['action_type']
student_id = p['student_id']
student_id = student_id.strip(' \t\n\r')
submission_id = submission_id.strip(' \t\n\r')
action_type = action_type.lower().strip(' \t\n\r')
# Make a service that can query edX ORA.
controller_qs = create_controller_query_service()
try:
response = controller_qs.take_action_on_flags(course_key, student_id, submission_id, action_type)
return HttpResponse(json.dumps(response), content_type="application/json")
except GradingServiceError:
log.exception(
u"Error taking action on flagged peer grading submissions, "
u"submission_id: {0}, action_type: {1}, grader_id: {2}"
.format(submission_id, action_type, student_id)
)
response = {
'success': False,
'error': STAFF_ERROR_MESSAGE
}
return HttpResponse(json.dumps(response), content_type="application/json")
lms/envs/common.py
......@@ -1029,26 +1029,6 @@ PAID_COURSE_REGISTRATION_CURRENCY = ['usd', '$']
# Members of this group are allowed to generate payment reports
PAYMENT_REPORT_GENERATOR_GROUP = 'shoppingcart_report_access'
################################# open ended grading config #####################
#By setting up the default settings with an incorrect user name and password,
# will get an error when attempting to connect
OPEN_ENDED_GRADING_INTERFACE = {
'url': 'http://example.com/peer_grading',
'username': 'incorrect_user',
'password': 'incorrect_pass',
'staff_grading': 'staff_grading',
'peer_grading': 'peer_grading',
'grading_controller': 'grading_controller'
}
# Used for testing, debugging peer grading
MOCK_PEER_GRADING = False
# Used for testing, debugging staff grading
MOCK_STAFF_GRADING = False
################################# EdxNotes config #########################
# Configure the LMS to use our stub EdxNotes implementation
......@@ -1828,7 +1808,6 @@ INSTALLED_APPS = (
'dashboard',
'instructor',
'instructor_task',
'open_ended_grading',
'openedx.core.djangoapps.course_groups',
'bulk_email',
'branding',
......
lms/urls.py
......@@ -549,61 +549,6 @@ urlpatterns += (
),
# see ENABLE_INSTRUCTOR_LEGACY_DASHBOARD section for legacy dash urls
# Open Ended grading views
url(
r'^courses/{}/staff_grading$'.format(
settings.COURSE_ID_PATTERN,
),
'open_ended_grading.views.staff_grading',
name='staff_grading',
),
url(
r'^courses/{}/staff_grading/get_next$'.format(
settings.COURSE_ID_PATTERN,
),
'open_ended_grading.staff_grading_service.get_next',
name='staff_grading_get_next',
),
url(
r'^courses/{}/staff_grading/save_grade$'.format(
settings.COURSE_ID_PATTERN,
),
'open_ended_grading.staff_grading_service.save_grade',
name='staff_grading_save_grade',
),
url(
r'^courses/{}/staff_grading/get_problem_list$'.format(
settings.COURSE_ID_PATTERN,
),
'open_ended_grading.staff_grading_service.get_problem_list',
name='staff_grading_get_problem_list',
),
# Open Ended problem list
url(
r'^courses/{}/open_ended_problems$'.format(
settings.COURSE_ID_PATTERN,
),
'open_ended_grading.views.student_problem_list',
name='open_ended_problems',
),
# Open Ended flagged problem list
url(
r'^courses/{}/open_ended_flagged_problems$'.format(
settings.COURSE_ID_PATTERN,
),
'open_ended_grading.views.flagged_problem_list',
name='open_ended_flagged_problems',
),
url(
r'^courses/{}/open_ended_flagged_problems/take_action_on_flags$'.format(
settings.COURSE_ID_PATTERN,
),
'open_ended_grading.views.take_action_on_flags',
name='open_ended_flagged_problems_take_action',
),
# Cohorts management
url(
r'^courses/{}/cohorts/settings$'.format(
......@@ -655,23 +600,6 @@ urlpatterns += (
name='cohort_discussion_topics',
),
# Open Ended Notifications
url(
r'^courses/{}/open_ended_notifications$'.format(
settings.COURSE_ID_PATTERN,
),
'open_ended_grading.views.combined_notifications',
name='open_ended_notifications',
),
url(
r'^courses/{}/peer_grading$'.format(
settings.COURSE_ID_PATTERN,
),
'open_ended_grading.views.peer_grading',
name='peer_grading',
),
url(
r'^courses/{}/notes$'.format(
settings.COURSE_ID_PATTERN,
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment