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

Merge pull request #14588 from edx/efischer/kill_stuff

Browse Source 
Kill edx-reverification-block
This commit is contained in:
Eric Fischer
2017-03-06 09:29:36 -05:00
committed by GitHub
parent 01ed17426f 7b85c0ea00
commit 83f2be4929
27 changed files with 82 additions and 3113 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
cms/djangoapps/contentstore/tests/test_utils.py
View File

@@ -535,9 +535,9 @@ class GetUserPartitionInfoTest(ModuleStoreTestCase):
),
UserPartition(
id=1,
name="Verification user partition",
scheme=UserPartition.get_scheme("verification"),
description="Verification user partition",
name="Completely random user partition",
scheme=UserPartition.get_scheme("random"),
description="Random user partition",
groups=[
Group(id=0, name="Group C"),
],
@@ -562,9 +562,9 @@ class GetUserPartitionInfoTest(ModuleStoreTestCase):
),
UserPartition(
id=1,
name="Verification user partition",
scheme=UserPartition.get_scheme("verification"),
description="Verification user partition",
name="Completely random user partition",
scheme=UserPartition.get_scheme("random"),
description="Random user partition",
groups=[
Group(id=0, name="Group C"),
],
@@ -574,7 +574,7 @@ class GetUserPartitionInfoTest(ModuleStoreTestCase):
# Expect that the partition with no groups is excluded from the results
partitions = self._get_partition_info()
self.assertEqual(len(partitions), 1)
self.assertEqual(partitions[0]["scheme"], "verification")
self.assertEqual(partitions[0]["scheme"], "random")
def _set_partitions(self, partitions):
"""Set the user partitions of the course descriptor. """

10
cms/djangoapps/contentstore/views/tests/test_item.py
View File

@@ -342,9 +342,9 @@ class GetItemTest(ItemTest):
self.course.user_partitions = [
UserPartition(
id=0,
name="Verification user partition",
scheme=UserPartition.get_scheme("verification"),
description="Verification user partition",
name="Random user partition",
scheme=UserPartition.get_scheme("random"),
description="Random user partition",
groups=[
Group(id=0, name="Group A"),
Group(id=1, name="Group B"),
@@ -364,8 +364,8 @@ class GetItemTest(ItemTest):
self.assertEqual(result["user_partitions"], [
{
"id": 0,
"name": "Verification user partition",
"scheme": "verification",
"name": "Random user partition",
"scheme": "random",
"groups": [
{
"id": 0,

31
cms/lib/xblock/test/test_authoring_mixin.py
View File

@@ -56,26 +56,6 @@ class AuthoringMixinTestCase(ModuleStoreTestCase):
self.course.user_partitions = [self.content_partition]
self.store.update_item(self.course, self.user.id)
def create_verification_user_partitions(self, checkpoint_names):
"""
Create user partitions for verification checkpoints.
"""
scheme = UserPartition.get_scheme("verification")
self.course.user_partitions = [
UserPartition(
id=0,
name=checkpoint_name,
description="Verification checkpoint",
scheme=scheme,
groups=[
Group(scheme.ALLOW, "Completed verification at {}".format(checkpoint_name)),
Group(scheme.DENY, "Did not complete verification at {}".format(checkpoint_name)),
],
)
for checkpoint_name in checkpoint_names
]
self.store.update_item(self.course, self.user.id)
def set_staff_only(self, item_location):
"""Make an item visible to staff only."""
item = self.store.get_item(item_location)
@@ -149,14 +129,3 @@ class AuthoringMixinTestCase(ModuleStoreTestCase):
'Content group no longer exists.'
]
)
def test_html_verification_checkpoints(self):
self.create_verification_user_partitions(["Midterm A", "Midterm B"])
self.verify_visibility_view_contains(
self.video_location,
[
"Verification Checkpoint",
"Midterm A",
"Midterm B",
]
)

38
cms/templates/visibility_editor.html
View File

@@ -1,12 +1,10 @@
<%
from django.utils.translation import ugettext as _
from openedx.core.djangoapps.credit.partition_schemes import VerificationPartitionScheme
from contentstore.utils import ancestor_has_staff_lock, get_visibility_partition_info
partition_info = get_visibility_partition_info(xblock)
user_partitions = partition_info["user_partitions"]
cohort_partitions = partition_info["cohort_partitions"]
verification_partitions = partition_info["verification_partitions"]
has_selected_groups = partition_info["has_selected_groups"]
selected_verified_partition_id = partition_info["selected_verified_partition_id"]
@@ -92,42 +90,6 @@ is_staff_locked = ancestor_has_staff_lock(xblock)
% endfor
% endfor
## Allow only one verification checkpoint to be selected at a time.
% if verification_partitions:
<div role="group" aria-labelledby="verification-access-title">
<div id="verification-access-title" class="sr">${_('Verification Checkpoint')}</div>
<div class="field field-checkbox field-visibility-verification">
<input type="checkbox"
id="verification-access-checkbox"
name="verification-access-checkbox"
class="input input-checkbox"
value=""
aria-describedby="verification-help-text"
${'checked="checked"' if selected_verified_partition_id is not None else ''}
/>
<label for="verification-access-checkbox" class="label">
${_('Verification Checkpoint')}:
</label>
<label class="sr" for="verification-access-dropdown">
${_('Verification checkpoint to complete')}
</label>
<select id="verification-access-dropdown">
% for partition in verification_partitions:
<option
value="${partition["id"]}"
${ "selected" if partition["id"] == selected_verified_partition_id else ""}
>${partition["name"]}</option>
% endfor
</select>
<div class="note" id="verification-help-text">
${_("Learners who require verification must pass the selected checkpoint to see the content in this component. Learners who do not require verification see this content by default.")}
</div>
</div>
</div>
% endif
</div>
</div>
</section>

3
lms/djangoapps/courseware/module_render.py
View File

@@ -48,7 +48,7 @@ from lms.djangoapps.grades.signals.signals import SCORE_PUBLISHED
from lms.djangoapps.lms_xblock.field_data import LmsFieldData
from lms.djangoapps.lms_xblock.models import XBlockAsidesConfig
from lms.djangoapps.lms_xblock.runtime import LmsModuleSystem
from lms.djangoapps.verify_student.services import VerificationService, ReverificationService
from lms.djangoapps.verify_student.services import VerificationService
from openedx.core.djangoapps.bookmarks.services import BookmarksService
from openedx.core.djangoapps.crawlers.models import CrawlersConfig
from openedx.core.djangoapps.credit.services import CreditService
@@ -679,7 +679,6 @@ def get_module_system_for_user(user, student_data, # TODO # pylint: disable=to
'field-data': field_data,
'user': DjangoXBlockUserService(user, user_is_staff=user_is_staff),
'verification': VerificationService(),
'reverification': ReverificationService(),
'proctoring': ProctoringService(),
'milestones': milestones_helpers.get_service(),
'credit': CreditService(),

6
lms/djangoapps/grades/tests/test_tasks.py
View File

@@ -145,8 +145,8 @@ class RecalculateSubsectionGradeTest(ModuleStoreTestCase):
@ddt.data(
(ModuleStoreEnum.Type.mongo, 1, 24, True),
(ModuleStoreEnum.Type.mongo, 1, 21, False),
(ModuleStoreEnum.Type.split, 3, 23, True),
(ModuleStoreEnum.Type.split, 3, 20, False),
(ModuleStoreEnum.Type.split, 3, 24, True),
(ModuleStoreEnum.Type.split, 3, 21, False),
)
@ddt.unpack
def test_query_counts(self, default_store, num_mongo_calls, num_sql_calls, create_multiple_subsections):
@@ -159,7 +159,7 @@ class RecalculateSubsectionGradeTest(ModuleStoreTestCase):
@ddt.data(
(ModuleStoreEnum.Type.mongo, 1, 24),
(ModuleStoreEnum.Type.split, 3, 23),
(ModuleStoreEnum.Type.split, 3, 24),
)
@ddt.unpack
def test_query_counts_dont_change_with_more_content(self, default_store, num_mongo_calls, num_sql_calls):

46
lms/djangoapps/verify_student/admin.py
View File

@@ -6,12 +6,7 @@ Admin site configurations for verify_student.
from config_models.admin import ConfigurationModelAdmin
from ratelimitbackend import admin
from lms.djangoapps.verify_student.models import (
IcrvStatusEmailsConfiguration,
SkippedReverification,
SoftwareSecurePhotoVerification,
VerificationStatus,
)
from lms.djangoapps.verify_student.models import SoftwareSecurePhotoVerification
@admin.register(SoftwareSecurePhotoVerification)
@@ -22,42 +17,3 @@ class SoftwareSecurePhotoVerificationAdmin(admin.ModelAdmin):
list_display = ('id', 'user', 'status', 'receipt_id', 'submitted_at', 'updated_at',)
raw_id_fields = ('user', 'reviewing_user', 'copy_id_photo_from',)
search_fields = ('receipt_id', 'user__username',)
@admin.register(VerificationStatus)
class VerificationStatusAdmin(admin.ModelAdmin):
"""
Admin for the VerificationStatus table.
"""
list_display = ('timestamp', 'user', 'status', 'checkpoint')
readonly_fields = ()
search_fields = ('checkpoint__checkpoint_location', 'user__username')
raw_id_fields = ('user',)
def get_readonly_fields(self, request, obj=None):
"""When editing an existing record, all fields should be read-only.
VerificationStatus records should be immutable; to change the user's
status, create a new record with the updated status and a more
recent timestamp.
"""
if obj:
return self.readonly_fields + ('status', 'checkpoint', 'user', 'response', 'error')
return self.readonly_fields
@admin.register(SkippedReverification)
class SkippedReverificationAdmin(admin.ModelAdmin):
"""Admin for the SkippedReverification table. """
list_display = ('created_at', 'user', 'course_id', 'checkpoint')
raw_id_fields = ('user',)
readonly_fields = ('user', 'course_id')
search_fields = ('user__username', 'course_id', 'checkpoint__checkpoint_location')
def has_add_permission(self, request):
"""Skipped verifications can't be created in Django admin. """
return False
admin.site.register(IcrvStatusEmailsConfiguration, ConfigurationModelAdmin)

319
lms/djangoapps/verify_student/models.py
View File

@@ -18,17 +18,14 @@ from email.utils import formatdate
import pytz
import requests
import uuid
from lazy import lazy
from opaque_keys.edx.keys import UsageKey
from django.conf import settings
from django.contrib.auth.models import User
from django.core.exceptions import ObjectDoesNotExist
from django.core.urlresolvers import reverse
from django.core.cache import cache
from django.core.files.base import ContentFile
from django.dispatch import receiver
from django.db import models, transaction
from django.db import models
from django.utils.functional import cached_property
from django.utils.translation import ugettext as _, ugettext_lazy
@@ -42,10 +39,9 @@ from lms.djangoapps.verify_student.ssencrypt import (
random_aes_key, encrypt_and_encode,
generate_signed_message, rsa_encrypt
)
from xmodule.modulestore.django import modulestore
from xmodule.modulestore.exceptions import ItemNotFoundError
from openedx.core.djangoapps.xmodule_django.models import CourseKeyField
from openedx.core.djangoapps.site_configuration import helpers as configuration_helpers
from openedx.core.djangolib.model_mixins import DeprecatedModelMixin
log = logging.getLogger(__name__)
@@ -1103,12 +1099,9 @@ def invalidate_deadline_caches(sender, **kwargs): # pylint: disable=unused-argu
cache.delete(VerificationDeadline.ALL_DEADLINES_CACHE_KEY)
class VerificationCheckpoint(models.Model):
"""Represents a point at which a user is asked to re-verify his/her
identity.
Each checkpoint is uniquely identified by a
(course_id, checkpoint_location) tuple.
class VerificationCheckpoint(DeprecatedModelMixin, models.Model): # pylint: disable=model-missing-unicode
"""
DEPRECATED - do not use. To be removed in a future Open edX release (Hawthorn).
"""
course_id = CourseKeyField(max_length=255, db_index=True)
checkpoint_location = models.CharField(max_length=255)
@@ -1118,86 +1111,10 @@ class VerificationCheckpoint(models.Model):
app_label = "verify_student"
unique_together = ('course_id', 'checkpoint_location')
def __unicode__(self):
"""
Unicode representation of the checkpoint.
"""
return u"{checkpoint} in {course}".format(
checkpoint=self.checkpoint_name,
course=self.course_id
)
@lazy
def checkpoint_name(self):
"""Lazy method for getting checkpoint name of reverification block.
Return location of the checkpoint if no related assessment found in
database.
"""
checkpoint_key = UsageKey.from_string(self.checkpoint_location)
try:
checkpoint_name = modulestore().get_item(checkpoint_key).related_assessment
except ItemNotFoundError:
log.warning(
u"Verification checkpoint block with location '%s' and course id '%s' "
u"not found in database.", self.checkpoint_location, unicode(self.course_id)
)
checkpoint_name = self.checkpoint_location
return checkpoint_name
def add_verification_attempt(self, verification_attempt):
"""Add the verification attempt in M2M relation of photo_verification.
Arguments:
verification_attempt(object): SoftwareSecurePhotoVerification object
Returns:
None
"""
self.photo_verification.add(verification_attempt) # pylint: disable=no-member
def get_user_latest_status(self, user_id):
"""Get the status of the latest checkpoint attempt of the given user.
Args:
user_id(str): Id of user
Returns:
VerificationStatus object if found any else None
"""
try:
return self.checkpoint_status.filter(user_id=user_id).latest()
except ObjectDoesNotExist:
return None
@classmethod
def get_or_create_verification_checkpoint(cls, course_id, checkpoint_location):
"""
Get or create the verification checkpoint for given 'course_id' and
checkpoint name.
Arguments:
course_id (CourseKey): CourseKey
checkpoint_location (str): Verification checkpoint location
Raises:
IntegrityError if create fails due to concurrent create.
Returns:
VerificationCheckpoint object if exists otherwise None
"""
with transaction.atomic():
checkpoint, __ = cls.objects.get_or_create(course_id=course_id, checkpoint_location=checkpoint_location)
return checkpoint
class VerificationStatus(models.Model):
"""This model is an append-only table that represents user status changes
during the verification process.
A verification status represents a user’s progress through the verification
process for a particular checkpoint.
class VerificationStatus(DeprecatedModelMixin, models.Model): # pylint: disable=model-missing-unicode
"""
DEPRECATED - do not use. To be removed in a future Open edX release (Hawthorn).
"""
SUBMITTED_STATUS = "submitted"
APPROVED_STATUS = "approved"
@@ -1224,172 +1141,24 @@ class VerificationStatus(models.Model):
verbose_name = "Verification Status"
verbose_name_plural = "Verification Statuses"
@classmethod
def add_verification_status(cls, checkpoint, user, status):
"""Create new verification status object.
Arguments:
checkpoint(VerificationCheckpoint): VerificationCheckpoint object
user(User): user object
status(str): Status from VERIFICATION_STATUS_CHOICES
Returns:
None
"""
cls.objects.create(checkpoint=checkpoint, user=user, status=status)
@classmethod
def add_status_from_checkpoints(cls, checkpoints, user, status):
"""Create new verification status objects for a user against the given
checkpoints.
Arguments:
checkpoints(list): list of VerificationCheckpoint objects
user(User): user object
status(str): Status from VERIFICATION_STATUS_CHOICES
Returns:
None
"""
for checkpoint in checkpoints:
cls.objects.create(checkpoint=checkpoint, user=user, status=status)
@classmethod
def get_user_status_at_checkpoint(cls, user, course_key, location):
"""
Get the user's latest status at the checkpoint.
Arguments:
user (User): The user whose status we are retrieving.
course_key (CourseKey): The identifier for the course.
location (UsageKey): The location of the checkpoint in the course.
Returns:
unicode or None
"""
try:
return cls.objects.filter(
user=user,
checkpoint__course_id=course_key,
checkpoint__checkpoint_location=unicode(location),
).latest().status
except cls.DoesNotExist:
return None
@classmethod
def get_user_attempts(cls, user_id, course_key, checkpoint_location):
"""
Get re-verification attempts against a user for a given 'checkpoint'
and 'course_id'.
Arguments:
user_id (str): User Id string
course_key (str): A CourseKey of a course
checkpoint_location (str): Verification checkpoint location
Returns:
Count of re-verification attempts
"""
return cls.objects.filter(
user_id=user_id,
checkpoint__course_id=course_key,
checkpoint__checkpoint_location=checkpoint_location,
status=cls.SUBMITTED_STATUS
).count()
@classmethod
def get_location_id(cls, photo_verification):
"""Get the location ID of reverification XBlock.
Args:
photo_verification(object): SoftwareSecurePhotoVerification object
Return:
Location Id of XBlock if any else empty string
"""
try:
verification_status = cls.objects.filter(checkpoint__photo_verification=photo_verification).latest()
return verification_status.checkpoint.checkpoint_location
except cls.DoesNotExist:
return ""
@classmethod
def get_all_checkpoints(cls, user_id, course_key):
"""Return dict of all the checkpoints with their status.
Args:
user_id(int): Id of user.
course_key(unicode): Unicode of course key
Returns:
dict: {checkpoint:status}
"""
all_checks_points = cls.objects.filter(
user_id=user_id, checkpoint__course_id=course_key
)
check_points = {}
for check in all_checks_points:
check_points[check.checkpoint.checkpoint_location] = check.status
return check_points
@classmethod
def cache_key_name(cls, user_id, course_key):
"""Return the name of the key to use to cache the current configuration
Args:
user_id(int): Id of user.
course_key(unicode): Unicode of course key
Returns:
Unicode cache key
"""
return u"verification.{}.{}".format(user_id, unicode(course_key))
@receiver(models.signals.post_save, sender=VerificationStatus)
@receiver(models.signals.post_delete, sender=VerificationStatus)
def invalidate_verification_status_cache(sender, instance, **kwargs): # pylint: disable=unused-argument, invalid-name
"""Invalidate the cache of VerificationStatus model. """
cache_key = VerificationStatus.cache_key_name(
instance.user.id,
unicode(instance.checkpoint.course_id)
)
cache.delete(cache_key)
# DEPRECATED: this feature has been permanently enabled.
# Once the application code has been updated in production,
# this table can be safely deleted.
class InCourseReverificationConfiguration(ConfigurationModel):
"""Configure in-course re-verification.
Enable or disable in-course re-verification feature.
When this flag is disabled, the "in-course re-verification" feature
will be disabled.
When the flag is enabled, the "in-course re-verification" feature
will be enabled.
class InCourseReverificationConfiguration(DeprecatedModelMixin, ConfigurationModel): # pylint: disable=model-missing-unicode
"""
DEPRECATED - do not use. To be removed in a future Open edX release (Hawthorn).
"""
pass
class IcrvStatusEmailsConfiguration(ConfigurationModel):
"""Toggle in-course reverification (ICRV) status emails
Disabled by default. When disabled, ICRV status emails will not be sent.
When enabled, ICRV status emails are sent.
class IcrvStatusEmailsConfiguration(DeprecatedModelMixin, ConfigurationModel): # pylint: disable=model-missing-unicode
"""
DEPRECATED - do not use. To be removed in a future Open edX release (Hawthorn).
"""
pass
class SkippedReverification(models.Model):
"""Model for tracking skipped Reverification of a user against a specific
course.
If a user skipped a Reverification checkpoint for a specific course then in
future that user cannot see the reverification link.
class SkippedReverification(DeprecatedModelMixin, models.Model): # pylint: disable=model-missing-unicode
"""
DEPRECATED - do not use. To be removed in a future Open edX release (Hawthorn).
"""
user = models.ForeignKey(User)
course_id = CourseKeyField(max_length=255, db_index=True)
@@ -1399,57 +1168,3 @@ class SkippedReverification(models.Model):
class Meta(object):
app_label = "verify_student"
unique_together = (('user', 'course_id'),)
@classmethod
@transaction.atomic
def add_skipped_reverification_attempt(cls, checkpoint, user_id, course_id):
"""Create skipped reverification object.
Arguments:
checkpoint(VerificationCheckpoint): VerificationCheckpoint object
user_id(str): User Id of currently logged in user
course_id(CourseKey): CourseKey
Returns:
None
"""
cls.objects.create(checkpoint=checkpoint, user_id=user_id, course_id=course_id)
@classmethod
def check_user_skipped_reverification_exists(cls, user_id, course_id):
"""Check existence of a user's skipped re-verification attempt for a
specific course.
Arguments:
user_id(str): user id
course_id(CourseKey): CourseKey
Returns:
Boolean
"""
has_skipped = cls.objects.filter(user_id=user_id, course_id=course_id).exists()
return has_skipped
@classmethod
def cache_key_name(cls, user_id, course_key):
"""Return the name of the key to use to cache the current configuration
Arguments:
user(User): user object
course_key(CourseKey): CourseKey
Returns:
string: cache key name
"""
return u"skipped_reverification.{}.{}".format(user_id, unicode(course_key))
@receiver(models.signals.post_save, sender=SkippedReverification)
@receiver(models.signals.post_delete, sender=SkippedReverification)
def invalidate_skipped_verification_cache(sender, instance, **kwargs): # pylint: disable=unused-argument, invalid-name
"""Invalidate the cache of skipped verification model. """
cache_key = SkippedReverification.cache_key_name(
instance.user.id,
unicode(instance.course_id)
)
cache.delete(cache_key)

122
lms/djangoapps/verify_student/services.py
View File

@@ -11,7 +11,6 @@ from django.db import IntegrityError
from opaque_keys.edx.keys import CourseKey
from student.models import User, CourseEnrollment
from lms.djangoapps.verify_student.models import VerificationCheckpoint, VerificationStatus, SkippedReverification
from .models import SoftwareSecurePhotoVerification
@@ -47,124 +46,3 @@ class VerificationService(object):
Returns the URL for a user to verify themselves.
"""
return reverse('verify_student_reverify')
class ReverificationService(object):
"""
Reverification XBlock service
"""
SKIPPED_STATUS = "skipped"
NON_VERIFIED_TRACK = "not-verified"
def get_status(self, user_id, course_id, related_assessment_location):
"""Get verification attempt status against a user for a given
'checkpoint' and 'course_id'.
Args:
user_id (str): User Id string
course_id (str): A string of course id
related_assessment_location (str): Location of Reverification XBlock
Returns: str or None
"""
user = User.objects.get(id=user_id)
course_key = CourseKey.from_string(course_id)
if not CourseEnrollment.is_enrolled_as_verified(user, course_key):
return self.NON_VERIFIED_TRACK
elif SkippedReverification.check_user_skipped_reverification_exists(user_id, course_key):
return self.SKIPPED_STATUS
try:
checkpoint_status = VerificationStatus.objects.filter(
user_id=user_id,
checkpoint__course_id=course_key,
checkpoint__checkpoint_location=related_assessment_location
).latest()
return checkpoint_status.status
except ObjectDoesNotExist:
return None
def start_verification(self, course_id, related_assessment_location):
"""Create re-verification link against a verification checkpoint.
Args:
course_id(str): A string of course id
related_assessment_location(str): Location of Reverification XBlock
Returns:
Re-verification link
"""
course_key = CourseKey.from_string(course_id)
# Get-or-create the verification checkpoint
VerificationCheckpoint.get_or_create_verification_checkpoint(course_key, related_assessment_location)
re_verification_link = reverse(
'verify_student_incourse_reverify',
args=(
unicode(course_key),
unicode(related_assessment_location)
)
)
return re_verification_link
def skip_verification(self, user_id, course_id, related_assessment_location):
"""Add skipped verification attempt entry for a user against a given
'checkpoint'.
Args:
user_id(str): User Id string
course_id(str): A string of course_id
related_assessment_location(str): Location of Reverification XBlock
Returns:
None
"""
course_key = CourseKey.from_string(course_id)
checkpoint = VerificationCheckpoint.objects.get(
course_id=course_key,
checkpoint_location=related_assessment_location
)
user = User.objects.get(id=user_id)
# user can skip a reverification attempt only if that user has not already
# skipped an attempt
try:
SkippedReverification.add_skipped_reverification_attempt(checkpoint, user_id, course_key)
except IntegrityError:
log.exception("Skipped attempt already exists for user %s: with course %s:", user_id, unicode(course_id))
return
try:
# Avoid circular import
from openedx.core.djangoapps.credit.api import set_credit_requirement_status
# As a user skips the reverification it declines to fulfill the requirement so
# requirement sets to declined.
set_credit_requirement_status(
user,
course_key,
'reverification',
checkpoint.checkpoint_location,
status='declined'
)
except Exception as err: # pylint: disable=broad-except
log.error("Unable to add credit requirement status for user with id %d: %s", user_id, err)
def get_attempts(self, user_id, course_id, related_assessment_location):
"""Get re-verification attempts against a user for a given 'checkpoint'
and 'course_id'.
Args:
user_id(str): User Id string
course_id(str): A string of course id
related_assessment_location(str): Location of Reverification XBlock
Returns:
Number of re-verification attempts of a user
"""
course_key = CourseKey.from_string(course_id)
return VerificationStatus.get_user_attempts(user_id, course_key, related_assessment_location)

307
lms/djangoapps/verify_student/tests/test_models.py
View File

@@ -5,7 +5,6 @@ import json
import boto
import ddt
from django.conf import settings
from django.db import IntegrityError
from freezegun import freeze_time
import mock
from mock import patch
@@ -24,9 +23,7 @@ from openedx.core.djangolib.testing.utils import CacheIsolationTestCase
from lms.djangoapps.verify_student.models import (
SoftwareSecurePhotoVerification,
VerificationException, VerificationCheckpoint,
VerificationStatus, SkippedReverification,
VerificationDeadline
VerificationException, VerificationDeadline
)
@@ -522,308 +519,6 @@ class TestPhotoVerification(MockS3Mixin, ModuleStoreTestCase):
self.assertEqual(fourth_result, first_result)
@ddt.ddt
class VerificationCheckpointTest(ModuleStoreTestCase):
"""Tests for the VerificationCheckpoint model. """
def setUp(self):
super(VerificationCheckpointTest, self).setUp()
self.user = UserFactory.create()
self.course = CourseFactory.create()
self.checkpoint_midterm = u'i4x://{org}/{course}/edx-reverification-block/midterm_uuid'.format(
org=self.course.id.org, course=self.course.id.course
)
self.checkpoint_final = u'i4x://{org}/{course}/edx-reverification-block/final_uuid'.format(
org=self.course.id.org, course=self.course.id.course
)
@ddt.data('midterm', 'final')
def test_get_or_create_verification_checkpoint(self, checkpoint):
"""
Test that a reverification checkpoint is created properly.
"""
checkpoint_location = u'i4x://{org}/{course}/edx-reverification-block/{checkpoint}'.format(
org=self.course.id.org, course=self.course.id.course, checkpoint=checkpoint
)
# create the 'VerificationCheckpoint' checkpoint
verification_checkpoint = VerificationCheckpoint.objects.create(
course_id=self.course.id,
checkpoint_location=checkpoint_location
)
self.assertEqual(
VerificationCheckpoint.get_or_create_verification_checkpoint(self.course.id, checkpoint_location),
verification_checkpoint
)
def test_get_or_create_verification_checkpoint_for_not_existing_values(self):
# Retrieving a checkpoint that doesn't yet exist will create it
location = u'i4x://edX/DemoX/edx-reverification-block/invalid_location'
checkpoint = VerificationCheckpoint.get_or_create_verification_checkpoint(self.course.id, location)
self.assertIsNot(checkpoint, None)
self.assertEqual(checkpoint.course_id, self.course.id)
self.assertEqual(checkpoint.checkpoint_location, location)
def test_get_or_create_integrity_error(self):
# Create the checkpoint
VerificationCheckpoint.objects.create(
course_id=self.course.id,
checkpoint_location=self.checkpoint_midterm,
)
# Simulate that the get-or-create operation raises an IntegrityError.
# This can happen when two processes both try to get-or-create at the same time
# when the database is set to REPEATABLE READ.
# To avoid IntegrityError situations when calling this method, set the view to
# use a READ COMMITTED transaction instead.
with patch.object(VerificationCheckpoint.objects, "get_or_create") as mock_get_or_create:
mock_get_or_create.side_effect = IntegrityError
with self.assertRaises(IntegrityError):
_ = VerificationCheckpoint.get_or_create_verification_checkpoint(
self.course.id,
self.checkpoint_midterm
)
def test_unique_together_constraint(self):
"""
Test the unique together constraint.
"""
# create the VerificationCheckpoint checkpoint
VerificationCheckpoint.objects.create(course_id=self.course.id, checkpoint_location=self.checkpoint_midterm)
# test creating the VerificationCheckpoint checkpoint with same course
# id and checkpoint name
with self.assertRaises(IntegrityError):
VerificationCheckpoint.objects.create(course_id=self.course.id, checkpoint_location=self.checkpoint_midterm)
def test_add_verification_attempt_software_secure(self):
"""
Test adding Software Secure photo verification attempts for the
reverification checkpoints.
"""
# adding two check points.
first_checkpoint = VerificationCheckpoint.objects.create(
course_id=self.course.id, checkpoint_location=self.checkpoint_midterm
)
second_checkpoint = VerificationCheckpoint.objects.create(
course_id=self.course.id, checkpoint_location=self.checkpoint_final
)
# make an attempt for the 'first_checkpoint'
first_checkpoint.add_verification_attempt(SoftwareSecurePhotoVerification.objects.create(user=self.user))
self.assertEqual(first_checkpoint.photo_verification.count(), 1)
# make another attempt for the 'first_checkpoint'
first_checkpoint.add_verification_attempt(SoftwareSecurePhotoVerification.objects.create(user=self.user))
self.assertEqual(first_checkpoint.photo_verification.count(), 2)
# make new attempt for the 'second_checkpoint'
attempt = SoftwareSecurePhotoVerification.objects.create(user=self.user)
second_checkpoint.add_verification_attempt(attempt)
self.assertEqual(second_checkpoint.photo_verification.count(), 1)
# remove the attempt from 'second_checkpoint'
second_checkpoint.photo_verification.remove(attempt)
self.assertEqual(second_checkpoint.photo_verification.count(), 0)
@ddt.ddt
class VerificationStatusTest(ModuleStoreTestCase):
""" Tests for the VerificationStatus model. """
def setUp(self):
super(VerificationStatusTest, self).setUp()
self.user = UserFactory.create()
self.course = CourseFactory.create()
self.first_checkpoint_location = u'i4x://{org}/{course}/edx-reverification-block/first_checkpoint_uuid'.format(
org=self.course.id.org, course=self.course.id.course
)
self.first_checkpoint = VerificationCheckpoint.objects.create(
course_id=self.course.id,
checkpoint_location=self.first_checkpoint_location
)
self.second_checkpoint_location = u'i4x://{org}/{course}/edx-reverification-block/second_checkpoint_uuid'.\
format(org=self.course.id.org, course=self.course.id.course)
self.second_checkpoint = VerificationCheckpoint.objects.create(
course_id=self.course.id,
checkpoint_location=self.second_checkpoint_location
)
@ddt.data('submitted', "approved", "denied", "error")
def test_add_verification_status(self, status):
""" Adding verification status using the class method. """
# adding verification status
VerificationStatus.add_verification_status(
checkpoint=self.first_checkpoint,
user=self.user,
status=status
)
# test the status from database
result = VerificationStatus.objects.filter(checkpoint=self.first_checkpoint)[0]
self.assertEqual(result.status, status)
self.assertEqual(result.user, self.user)
@ddt.data("approved", "denied", "error")
def test_add_status_from_checkpoints(self, status):
"""Test verification status for reverification checkpoints after
submitting software secure photo verification.
"""
# add initial verification status for checkpoints
initial_status = "submitted"
VerificationStatus.add_verification_status(
checkpoint=self.first_checkpoint,
user=self.user,
status=initial_status
)
VerificationStatus.add_verification_status(
checkpoint=self.second_checkpoint,
user=self.user,
status=initial_status
)
# now add verification status for multiple checkpoint points
VerificationStatus.add_status_from_checkpoints(
checkpoints=[self.first_checkpoint, self.second_checkpoint], user=self.user, status=status
)
# test that verification status entries with new status have been added
# for both checkpoints
result = VerificationStatus.objects.filter(user=self.user, checkpoint=self.first_checkpoint)
self.assertEqual(len(result), len(self.first_checkpoint.checkpoint_status.all()))
self.assertEqual(
list(result.values_list('checkpoint__checkpoint_location', flat=True)),
list(self.first_checkpoint.checkpoint_status.values_list('checkpoint__checkpoint_location', flat=True))
)
result = VerificationStatus.objects.filter(user=self.user, checkpoint=self.second_checkpoint)
self.assertEqual(len(result), len(self.second_checkpoint.checkpoint_status.all()))
self.assertEqual(
list(result.values_list('checkpoint__checkpoint_location', flat=True)),
list(self.second_checkpoint.checkpoint_status.values_list('checkpoint__checkpoint_location', flat=True))
)
def test_get_location_id(self):
"""
Getting location id for a specific checkpoint.
"""
# creating software secure attempt against checkpoint
self.first_checkpoint.add_verification_attempt(SoftwareSecurePhotoVerification.objects.create(user=self.user))
# add initial verification status for checkpoint
VerificationStatus.add_verification_status(
checkpoint=self.first_checkpoint,
user=self.user,
status='submitted',
)
attempt = SoftwareSecurePhotoVerification.objects.filter(user=self.user)
self.assertIsNotNone(VerificationStatus.get_location_id(attempt))
self.assertEqual(VerificationStatus.get_location_id(None), '')
def test_get_user_attempts(self):
"""
Test adding verification status.
"""
VerificationStatus.add_verification_status(
checkpoint=self.first_checkpoint,
user=self.user,
status='submitted'
)
actual_attempts = VerificationStatus.get_user_attempts(
self.user.id,
self.course.id,
self.first_checkpoint_location
)
self.assertEqual(actual_attempts, 1)
class SkippedReverificationTest(ModuleStoreTestCase):
"""
Tests for the SkippedReverification model.
"""
def setUp(self):
super(SkippedReverificationTest, self).setUp()
self.user = UserFactory.create()
self.course = CourseFactory.create()
dummy_checkpoint_location = u'i4x://edX/DemoX/edx-reverification-block/midterm_uuid'
self.checkpoint = VerificationCheckpoint.objects.create(
course_id=self.course.id,
checkpoint_location=dummy_checkpoint_location
)
def test_add_skipped_attempts(self):
"""
Test 'add_skipped_reverification_attempt' method.
"""
# add verification status
SkippedReverification.add_skipped_reverification_attempt(
checkpoint=self.checkpoint, user_id=self.user.id, course_id=unicode(self.course.id)
)
# test the status of skipped reverification from database
result = SkippedReverification.objects.filter(course_id=self.course.id)[0]
self.assertEqual(result.checkpoint, self.checkpoint)
self.assertEqual(result.user, self.user)
self.assertEqual(result.course_id, self.course.id)
def test_unique_constraint(self):
"""Test that adding skipped re-verification with same user and course
id will raise 'IntegrityError' exception.
"""
# add verification object
SkippedReverification.add_skipped_reverification_attempt(
checkpoint=self.checkpoint, user_id=self.user.id, course_id=unicode(self.course.id)
)
with self.assertRaises(IntegrityError):
SkippedReverification.add_skipped_reverification_attempt(
checkpoint=self.checkpoint, user_id=self.user.id, course_id=unicode(self.course.id)
)
# create skipped attempt for different user
user2 = UserFactory.create()
SkippedReverification.add_skipped_reverification_attempt(
checkpoint=self.checkpoint, user_id=user2.id, course_id=unicode(self.course.id)
)
# test the status of skipped reverification from database
result = SkippedReverification.objects.filter(user=user2)[0]
self.assertEqual(result.checkpoint, self.checkpoint)
self.assertEqual(result.user, user2)
self.assertEqual(result.course_id, self.course.id)
def test_check_user_skipped_reverification_exists(self):
"""
Test the 'check_user_skipped_reverification_exists' method's response.
"""
# add verification status
SkippedReverification.add_skipped_reverification_attempt(
checkpoint=self.checkpoint, user_id=self.user.id, course_id=unicode(self.course.id)
)
self.assertTrue(
SkippedReverification.check_user_skipped_reverification_exists(
user_id=self.user.id,
course_id=self.course.id
)
)
user2 = UserFactory.create()
self.assertFalse(
SkippedReverification.check_user_skipped_reverification_exists(
user_id=user2.id,
course_id=self.course.id
)
)
class VerificationDeadlineTest(CacheIsolationTestCase):
"""
Tests for the VerificationDeadline model.

194
lms/djangoapps/verify_student/tests/test_services.py
View File

@@ -1,194 +0,0 @@
"""
Tests of re-verification service.
"""
import ddt
from opaque_keys.edx.keys import CourseKey
from course_modes.models import CourseMode
from course_modes.tests.factories import CourseModeFactory
from student.models import CourseEnrollment
from student.tests.factories import UserFactory
from lms.djangoapps.verify_student.models import VerificationCheckpoint, VerificationStatus, SkippedReverification
from lms.djangoapps.verify_student.services import ReverificationService
from openedx.core.djangoapps.credit.api import get_credit_requirement_status, set_credit_requirements
from openedx.core.djangoapps.credit.models import CreditCourse
from xmodule.modulestore.tests.django_utils import ModuleStoreTestCase
from xmodule.modulestore.tests.factories import CourseFactory, ItemFactory
@ddt.ddt
class TestReverificationService(ModuleStoreTestCase):
"""
Tests for the re-verification service.
"""
def setUp(self):
super(TestReverificationService, self).setUp()
self.user = UserFactory.create(username="rusty", password="test")
self.course = CourseFactory.create(org='Robot', number='999', display_name='Test Course')
self.course_id = self.course.id
CourseModeFactory.create(
mode_slug="verified",
course_id=self.course_id,
min_price=100,
)
self.course_key = CourseKey.from_string(unicode(self.course_id))
self.item = ItemFactory.create(parent=self.course, category='chapter', display_name='Test Section')
self.final_checkpoint_location = u'i4x://{org}/{course}/edx-reverification-block/final_uuid'.format(
org=self.course_id.org, course=self.course_id.course
)
# Enroll in a verified mode
self.enrollment = CourseEnrollment.enroll(self.user, self.course_id, mode=CourseMode.VERIFIED)
@ddt.data('final', 'midterm')
def test_start_verification(self, checkpoint_name):
"""Test the 'start_verification' service method.
Check that if a reverification checkpoint exists for a specific course
then 'start_verification' method returns that checkpoint otherwise it
creates that checkpoint.
"""
reverification_service = ReverificationService()
checkpoint_location = u'i4x://{org}/{course}/edx-reverification-block/{checkpoint}'.format(
org=self.course_id.org, course=self.course_id.course, checkpoint=checkpoint_name
)
expected_url = (
'/verify_student/reverify'
'/{course_key}'
'/{checkpoint_location}/'
).format(course_key=unicode(self.course_id), checkpoint_location=checkpoint_location)
self.assertEqual(
reverification_service.start_verification(unicode(self.course_id), checkpoint_location),
expected_url
)
def test_get_status(self):
"""Test the verification statuses of a user for a given 'checkpoint'
and 'course_id'.
"""
reverification_service = ReverificationService()
self.assertIsNone(
reverification_service.get_status(self.user.id, unicode(self.course_id), self.final_checkpoint_location)
)
checkpoint_obj = VerificationCheckpoint.objects.create(
course_id=unicode(self.course_id),
checkpoint_location=self.final_checkpoint_location
)
VerificationStatus.objects.create(checkpoint=checkpoint_obj, user=self.user, status='submitted')
self.assertEqual(
reverification_service.get_status(self.user.id, unicode(self.course_id), self.final_checkpoint_location),
'submitted'
)
VerificationStatus.objects.create(checkpoint=checkpoint_obj, user=self.user, status='approved')
self.assertEqual(
reverification_service.get_status(self.user.id, unicode(self.course_id), self.final_checkpoint_location),
'approved'
)
def test_skip_verification(self):
"""
Test adding skip attempt of a user for a reverification checkpoint.
"""
reverification_service = ReverificationService()
VerificationCheckpoint.objects.create(
course_id=unicode(self.course_id),
checkpoint_location=self.final_checkpoint_location
)
reverification_service.skip_verification(self.user.id, unicode(self.course_id), self.final_checkpoint_location)
self.assertEqual(
SkippedReverification.objects.filter(user=self.user, course_id=self.course_id).count(),
1
)
# now test that a user can have only one entry for a skipped
# reverification for a course
reverification_service.skip_verification(self.user.id, unicode(self.course_id), self.final_checkpoint_location)
self.assertEqual(
SkippedReverification.objects.filter(user=self.user, course_id=self.course_id).count(),
1
)
# testing service for skipped attempt.
self.assertEqual(
reverification_service.get_status(self.user.id, unicode(self.course_id), self.final_checkpoint_location),
'skipped'
)
@ddt.data(
*CourseMode.CREDIT_ELIGIBLE_MODES
)
def test_declined_verification_on_skip(self, mode):
"""Test that status with value 'declined' is added in credit
requirement status model when a user skip's an ICRV.
"""
reverification_service = ReverificationService()
checkpoint = VerificationCheckpoint.objects.create(
course_id=unicode(self.course_id),
checkpoint_location=self.final_checkpoint_location
)
# Create credit course and set credit requirements.
CreditCourse.objects.create(course_key=self.course_key, enabled=True)
self.enrollment.update_enrollment(mode=mode)
set_credit_requirements(
self.course_key,
[
{
"namespace": "reverification",
"name": checkpoint.checkpoint_location,
"display_name": "Assessment 1",
"criteria": {},
}
]
)
reverification_service.skip_verification(self.user.id, unicode(self.course_id), self.final_checkpoint_location)
requirement_status = get_credit_requirement_status(
self.course_key, self.user.username, 'reverification', checkpoint.checkpoint_location
)
self.assertEqual(SkippedReverification.objects.filter(user=self.user, course_id=self.course_id).count(), 1)
self.assertEqual(len(requirement_status), 1)
self.assertEqual(requirement_status[0].get('name'), checkpoint.checkpoint_location)
self.assertEqual(requirement_status[0].get('status'), 'declined')
def test_get_attempts(self):
"""Check verification attempts count against a user for a given
'checkpoint' and 'course_id'.
"""
reverification_service = ReverificationService()
course_id = unicode(self.course_id)
self.assertEqual(
reverification_service.get_attempts(self.user.id, course_id, self.final_checkpoint_location),
0
)
# now create a checkpoint and add user's entry against it then test
# that the 'get_attempts' service method returns correct count
checkpoint_obj = VerificationCheckpoint.objects.create(
course_id=course_id,
checkpoint_location=self.final_checkpoint_location
)
VerificationStatus.objects.create(checkpoint=checkpoint_obj, user=self.user, status='submitted')
self.assertEqual(
reverification_service.get_attempts(self.user.id, course_id, self.final_checkpoint_location),
1
)
def test_not_in_verified_track(self):
# No longer enrolled in a verified track
self.enrollment.update_enrollment(mode=CourseMode.HONOR)
# Should be marked as "skipped" (opted out)
service = ReverificationService()
status = service.get_status(self.user.id, unicode(self.course_id), self.final_checkpoint_location)
self.assertEqual(status, service.NON_VERIFIED_TRACK)

657
lms/djangoapps/verify_student/tests/test_views.py
View File

@@ -16,7 +16,7 @@ import boto
import moto
import pytz
from bs4 import BeautifulSoup
from mock import patch, Mock, ANY
from mock import patch, Mock
import requests
from django.conf import settings
@@ -25,15 +25,12 @@ from django.core import mail
from django.test import TestCase
from django.test.client import Client, RequestFactory
from django.test.utils import override_settings
from django.utils import timezone
from opaque_keys.edx.locations import SlashSeparatedCourseKey
from opaque_keys.edx.locator import CourseLocator
from opaque_keys.edx.keys import UsageKey
from course_modes.models import CourseMode
from course_modes.tests.factories import CourseModeFactory
from courseware.url_helpers import get_redirect_url
from common.test.utils import XssTestMixin
from commerce.models import CommerceConfiguration
from commerce.tests import TEST_PAYMENT_DATA, TEST_API_URL, TEST_PUBLIC_URL_ROOT
@@ -43,22 +40,17 @@ from openedx.core.djangoapps.theming.tests.test_util import with_comprehensive_t
from shoppingcart.models import Order, CertificateItem
from student.tests.factories import UserFactory, CourseEnrollmentFactory
from student.models import CourseEnrollment
from util.date_utils import get_default_time_display
from util.testing import UrlResetMixin
from lms.djangoapps.verify_student.views import (
checkout_with_ecommerce_service, render_to_response, PayAndVerifyView,
_compose_message_reverification_email
)
from lms.djangoapps.verify_student.models import (
VerificationDeadline, SoftwareSecurePhotoVerification,
VerificationCheckpoint, VerificationStatus,
IcrvStatusEmailsConfiguration,
)
from xmodule.modulestore.tests.django_utils import ModuleStoreTestCase
from xmodule.modulestore.tests.factories import CourseFactory, ItemFactory
from xmodule.modulestore.tests.factories import CourseFactory
from xmodule.modulestore.django import modulestore
from xmodule.modulestore import ModuleStoreEnum
from xmodule.modulestore.tests.factories import check_mongo_calls
def mock_render_to_response(*args, **kwargs):
@@ -1840,159 +1832,6 @@ class TestPhotoVerificationResultsCallback(ModuleStoreTestCase):
)
self.assertIn('Result Unknown not understood', response.content)
@mock.patch(
'lms.djangoapps.verify_student.ssencrypt.has_valid_signature',
mock.Mock(side_effect=mocked_has_valid_signature)
)
def test_in_course_reverify_disabled(self):
"""
Test for verification passed.
"""
data = {
"EdX-ID": self.receipt_id,
"Result": "PASS",
"Reason": "",
"MessageType": "You have been verified."
}
json_data = json.dumps(data)
response = self.client.post(
reverse('verify_student_results_callback'), data=json_data,
content_type='application/json',
HTTP_AUTHORIZATION='test BBBBBBBBBBBBBBBBBBBB:testing',
HTTP_DATE='testdate'
)
attempt = SoftwareSecurePhotoVerification.objects.get(receipt_id=self.receipt_id)
self.assertEqual(attempt.status, u'approved')
self.assertEquals(response.content, 'OK!')
# Verify that photo submission confirmation email was sent
self.assertEqual(len(mail.outbox), 0)
user_status = VerificationStatus.objects.filter(user=self.user).count()
self.assertEqual(user_status, 0)
@mock.patch(
'lms.djangoapps.verify_student.ssencrypt.has_valid_signature',
mock.Mock(side_effect=mocked_has_valid_signature)
)
def test_pass_in_course_reverify_result(self):
"""
Test for verification passed.
"""
# Verify that ICRV status email was sent when config is enabled
IcrvStatusEmailsConfiguration.objects.create(enabled=True)
self.create_reverification_xblock()
data = {
"EdX-ID": self.receipt_id,
"Result": "PASS",
"Reason": "",
"MessageType": "You have been verified."
}
json_data = json.dumps(data)
response = self.client.post(
reverse('verify_student_results_callback'), data=json_data,
content_type='application/json',
HTTP_AUTHORIZATION='test BBBBBBBBBBBBBBBBBBBB:testing',
HTTP_DATE='testdate'
)
attempt = SoftwareSecurePhotoVerification.objects.get(receipt_id=self.receipt_id)
self.assertEqual(attempt.status, u'approved')
self.assertEquals(response.content, 'OK!')
self.assertEqual(len(mail.outbox), 1)
self.assertEqual("Re-verification Status", mail.outbox[0].subject)
@mock.patch('verify_student.ssencrypt.has_valid_signature', mock.Mock(side_effect=mocked_has_valid_signature))
def test_icrv_status_email_with_disable_config(self):
"""
Verify that photo re-verification status email was not sent when config is disable
"""
IcrvStatusEmailsConfiguration.objects.create(enabled=False)
self.create_reverification_xblock()
data = {
"EdX-ID": self.receipt_id,
"Result": "PASS",
"Reason": "",
"MessageType": "You have been verified."
}
json_data = json.dumps(data)
response = self.client.post(
reverse('verify_student_results_callback'), data=json_data,
content_type='application/json',
HTTP_AUTHORIZATION='test BBBBBBBBBBBBBBBBBBBB:testing',
HTTP_DATE='testdate'
)
attempt = SoftwareSecurePhotoVerification.objects.get(receipt_id=self.receipt_id)
self.assertEqual(attempt.status, u'approved')
self.assertEquals(response.content, 'OK!')
self.assertEqual(len(mail.outbox), 0)
@mock.patch('lms.djangoapps.verify_student.views._send_email')
@mock.patch(
'lms.djangoapps.verify_student.ssencrypt.has_valid_signature',
mock.Mock(side_effect=mocked_has_valid_signature)
)
def test_reverification_on_callback(self, mock_send_email):
"""
Test software secure callback flow for re-verification.
"""
IcrvStatusEmailsConfiguration.objects.create(enabled=True)
# Create the 'edx-reverification-block' in course tree
self.create_reverification_xblock()
# create dummy data for software secure photo verification result callback
data = {
"EdX-ID": self.receipt_id,
"Result": "PASS",
"Reason": "",
"MessageType": "You have been verified."
}
json_data = json.dumps(data)
response = self.client.post(
reverse('verify_student_results_callback'),
data=json_data,
content_type='application/json',
HTTP_AUTHORIZATION='test BBBBBBBBBBBBBBBBBBBB:testing',
HTTP_DATE='testdate'
)
self.assertEqual(response.content, 'OK!')
# now check that '_send_email' method is called on result callback
# with required parameters
subject = "Re-verification Status"
mock_send_email.assert_called_once_with(self.user.id, subject, ANY)
def create_reverification_xblock(self):
"""
Create the reverification XBlock.
"""
# Create the 'edx-reverification-block' in course tree
section = ItemFactory.create(parent=self.course, category='chapter', display_name='Test Section')
subsection = ItemFactory.create(parent=section, category='sequential', display_name='Test Subsection')
vertical = ItemFactory.create(parent=subsection, category='vertical', display_name='Test Unit')
reverification = ItemFactory.create(
parent=vertical,
category='edx-reverification-block',
display_name='Test Verification Block'
)
# Create checkpoint
checkpoint = VerificationCheckpoint(course_id=self.course_id, checkpoint_location=reverification.location)
checkpoint.save()
# Add a re-verification attempt
checkpoint.add_verification_attempt(self.attempt)
# Add a re-verification attempt status for the user
VerificationStatus.add_verification_status(checkpoint, self.user, "submitted")
@attr(shard=2)
class TestReverifyView(TestCase):
@@ -2104,495 +1943,3 @@ class TestReverifyView(TestCase):
"""
response = self._get_reverify_page()
self.assertContains(response, "reverify-blocked")
@attr(shard=2)
class TestInCourseReverifyView(ModuleStoreTestCase):
"""
Tests for the incourse reverification views.
"""
IMAGE_DATA = "abcd,1234"
def build_course(self):
"""
Build up a course tree with a Reverificaiton xBlock.
"""
self.course_key = SlashSeparatedCourseKey("Robot", "999", "Test_Course")
self.course = CourseFactory.create(org='Robot', number='999', display_name='Test Course')
# Create the course modes
for mode in ('audit', 'honor', 'verified'):
min_price = 0 if mode in ["honor", "audit"] else 1
CourseModeFactory.create(mode_slug=mode, course_id=self.course_key, min_price=min_price)
# Create the 'edx-reverification-block' in course tree
section = ItemFactory.create(parent=self.course, category='chapter', display_name='Test Section')
subsection = ItemFactory.create(parent=section, category='sequential', display_name='Test Subsection')
vertical = ItemFactory.create(parent=subsection, category='vertical', display_name='Test Unit')
self.reverification = ItemFactory.create(
parent=vertical,
category='edx-reverification-block',
display_name='Test Verification Block'
)
self.section_location = section.location
self.subsection_location = subsection.location
self.vertical_location = vertical.location
self.reverification_location = unicode(self.reverification.location)
self.reverification_assessment = self.reverification.related_assessment
def setUp(self):
super(TestInCourseReverifyView, self).setUp()
self.build_course()
self.user = UserFactory.create(username="rusty", password="test")
self.client.login(username="rusty", password="test")
# Enroll the user in the default mode (honor) to emulate
CourseEnrollment.enroll(self.user, self.course_key, mode="verified")
# mocking and patching for bi events
analytics_patcher = patch('lms.djangoapps.verify_student.views.analytics')
self.mock_tracker = analytics_patcher.start()
self.addCleanup(analytics_patcher.stop)
@patch.dict(settings.FEATURES, {'AUTOMATIC_VERIFY_STUDENT_IDENTITY_FOR_TESTING': True})
def test_incourse_reverify_invalid_checkpoint_get(self):
# Retrieve a checkpoint that doesn't yet exist
response = self.client.get(self._get_url(self.course_key, "invalid_checkpoint"))
self.assertEqual(response.status_code, 404)
@patch.dict(settings.FEATURES, {'AUTOMATIC_VERIFY_STUDENT_IDENTITY_FOR_TESTING': True})
def test_incourse_reverify_initial_redirect_get(self):
self._create_checkpoint()
response = self.client.get(self._get_url(self.course_key, self.reverification_location))
url = reverse('verify_student_verify_now', kwargs={"course_id": unicode(self.course_key)})
url += u"?{params}".format(params=urllib.urlencode({"checkpoint": self.reverification_location}))
self.assertRedirects(response, url)
@override_settings(LMS_SEGMENT_KEY="foobar")
@patch.dict(settings.FEATURES, {'AUTOMATIC_VERIFY_STUDENT_IDENTITY_FOR_TESTING': True})
def test_incourse_reverify_get(self):
"""
Test incourse reverification.
"""
self._create_checkpoint()
self._create_initial_verification()
response = self.client.get(self._get_url(self.course_key, self.reverification_location))
self.assertEquals(response.status_code, 200)
# verify that Google Analytics event fires after successfully
# submitting the photo verification
self.mock_tracker.track.assert_called_once_with( # pylint: disable=no-member
self.user.id,
'edx.bi.reverify.started',
{
'category': "verification",
'label': unicode(self.course_key),
'checkpoint': self.reverification_assessment
},
context={
'ip': '127.0.0.1',
'Google Analytics':
{'clientId': None}
}
)
self.mock_tracker.reset_mock()
@patch.dict(settings.FEATURES, {'AUTOMATIC_VERIFY_STUDENT_IDENTITY_FOR_TESTING': True})
def test_checkpoint_post(self):
"""Verify that POST requests including an invalid checkpoint location
results in a 400 response.
"""
response = self._submit_photos(self.course_key, self.reverification_location, self.IMAGE_DATA)
self.assertEquals(response.status_code, 400)
@patch.dict(settings.FEATURES, {'AUTOMATIC_VERIFY_STUDENT_IDENTITY_FOR_TESTING': True})
def test_incourse_reverify_id_required_if_no_initial_verification(self):
self._create_checkpoint()
# Since the user has no initial verification and we're not sending the ID photo,
# we should expect a 400 bad request
response = self._submit_photos(self.course_key, self.reverification_location, self.IMAGE_DATA)
self.assertEqual(response.status_code, 400)
@patch.dict(settings.FEATURES, {'AUTOMATIC_VERIFY_STUDENT_IDENTITY_FOR_TESTING': True})
def test_incourse_reverify_index_error_post(self):
self._create_checkpoint()
self._create_initial_verification()
response = self._submit_photos(self.course_key, self.reverification_location, "")
self.assertEqual(response.status_code, 400)
@override_settings(LMS_SEGMENT_KEY="foobar")
@patch.dict(settings.FEATURES, {'AUTOMATIC_VERIFY_STUDENT_IDENTITY_FOR_TESTING': True})
def test_incourse_reverify_post(self):
self._create_checkpoint()
self._create_initial_verification()
response = self._submit_photos(self.course_key, self.reverification_location, self.IMAGE_DATA)
self.assertEqual(response.status_code, 200)
# Check that the checkpoint status has been updated
status = VerificationStatus.get_user_status_at_checkpoint(
self.user, self.course_key, self.reverification_location
)
self.assertEqual(status, "submitted")
# Test that Google Analytics event fires after successfully submitting
# photo verification
self.mock_tracker.track.assert_called_once_with( # pylint: disable=no-member
self.user.id,
'edx.bi.reverify.submitted',
{
'category': "verification",
'label': unicode(self.course_key),
'checkpoint': self.reverification_assessment
},
context={
'ip': '127.0.0.1',
'Google Analytics':
{'clientId': None}
}
)
self.mock_tracker.reset_mock()
def _create_checkpoint(self):
"""
Helper method for creating a reverification checkpoint.
"""
checkpoint = VerificationCheckpoint(course_id=self.course_key, checkpoint_location=self.reverification_location)
checkpoint.save()
def _create_initial_verification(self):
"""
Helper method for initial verification.
"""
attempt = SoftwareSecurePhotoVerification(user=self.user, photo_id_key="dummy_photo_id_key")
attempt.mark_ready()
attempt.save()
attempt.submit()
def _get_url(self, course_key, checkpoint_location):
"""
Construct the reverification url.
Arguments:
course_key (unicode): The ID of the course
checkpoint_location (str): Location of verification checkpoint
Returns:
url
"""
return reverse(
'verify_student_incourse_reverify',
kwargs={
"course_id": unicode(course_key),
"usage_id": checkpoint_location
}
)
def _submit_photos(self, course_key, checkpoint_location, face_image_data):
""" Submit photos for verification. """
url = reverse("verify_student_submit_photos")
data = {
"course_key": unicode(course_key),
"checkpoint": checkpoint_location,
"face_image": face_image_data,
}
return self.client.post(url, data)
@attr(shard=2)
class TestEmailMessageWithCustomICRVBlock(ModuleStoreTestCase):
"""
Test email sending on re-verification
"""
def build_course(self):
"""
Build up a course tree with a Reverificaiton xBlock.
"""
self.course_key = SlashSeparatedCourseKey("Robot", "999", "Test_Course")
self.course = CourseFactory.create(org='Robot', number='999', display_name='Test Course')
self.due_date = datetime.now(pytz.UTC) + timedelta(days=20)
self.allowed_attempts = 1
# Create the course modes
for mode in ('audit', 'honor', 'verified'):
min_price = 0 if mode in ["honor", "audit"] else 1
CourseModeFactory.create(mode_slug=mode, course_id=self.course_key, min_price=min_price)
# Create the 'edx-reverification-block' in course tree
section = ItemFactory.create(parent=self.course, category='chapter', display_name='Test Section')
subsection = ItemFactory.create(parent=section, category='sequential', display_name='Test Subsection')
vertical = ItemFactory.create(parent=subsection, category='vertical', display_name='Test Unit')
self.reverification = ItemFactory.create(
parent=vertical,
category='edx-reverification-block',
display_name='Test Verification Block',
metadata={'attempts': self.allowed_attempts, 'due': self.due_date}
)
self.section_location = section.location
self.subsection_location = subsection.location
self.vertical_location = vertical.location
self.reverification_location = unicode(self.reverification.location)
self.assessment = self.reverification.related_assessment
self.re_verification_link = reverse(
'verify_student_incourse_reverify',
args=(
unicode(self.course_key),
self.reverification_location
)
)
def setUp(self):
"""
Setup method for testing photo verification email messages.
"""
super(TestEmailMessageWithCustomICRVBlock, self).setUp()
self.build_course()
self.check_point = VerificationCheckpoint.objects.create(
course_id=self.course.id, checkpoint_location=self.reverification_location
)
self.check_point.add_verification_attempt(SoftwareSecurePhotoVerification.objects.create(user=self.user))
VerificationStatus.add_verification_status(
checkpoint=self.check_point,
user=self.user,
status='submitted'
)
self.attempt = SoftwareSecurePhotoVerification.objects.filter(user=self.user)
location_id = VerificationStatus.get_location_id(self.attempt)
usage_key = UsageKey.from_string(location_id)
redirect_url = get_redirect_url(self.course_key, usage_key.replace(course_key=self.course_key))
self.request = RequestFactory().get('/url')
self.course_link = self.request.build_absolute_uri(redirect_url)
def test_approved_email_message(self):
subject, body = _compose_message_reverification_email(
self.course.id, self.user.id, self.reverification_location, "approved", self.request
)
self.assertIn(
"We have successfully verified your identity for the {assessment} "
"assessment in the {course_name} course.".format(
assessment=self.assessment,
course_name=self.course.display_name_with_default_escaped
),
body
)
self.check_courseware_link_exists(body)
self.assertIn("Re-verification Status", subject)
def test_denied_email_message_with_valid_due_date_and_attempts_allowed(self):
subject, body = _compose_message_reverification_email(
self.course.id, self.user.id, self.reverification_location, "denied", self.request
)
self.assertIn(
"We could not verify your identity for the {assessment} assessment "
"in the {course_name} course. You have used "
"{used_attempts} out of {allowed_attempts} attempts to "
"verify your identity".format(
course_name=self.course.display_name_with_default_escaped,
assessment=self.assessment,
used_attempts=1,
allowed_attempts=self.allowed_attempts + 1
),
body
)
self.assertIn(
"You must verify your identity before the assessment "
"closes on {due_date}".format(
due_date=get_default_time_display(self.due_date)
),
body
)
reverify_link = self.request.build_absolute_uri(self.re_verification_link)
self.assertIn(
"To try to verify your identity again, select the following link:",
body
)
self.assertIn(reverify_link, body)
self.assertIn("Re-verification Status", subject)
def test_denied_email_message_with_due_date_and_no_attempts(self):
""" Denied email message if due date is still open but user has no
attempts available.
"""
VerificationStatus.add_verification_status(
checkpoint=self.check_point,
user=self.user,
status='submitted'
)
__, body = _compose_message_reverification_email(
self.course.id, self.user.id, self.reverification_location, "denied", self.request
)
self.assertIn(
"We could not verify your identity for the {assessment} assessment "
"in the {course_name} course. You have used "
"{used_attempts} out of {allowed_attempts} attempts to "
"verify your identity, and verification is no longer "
"possible".format(
course_name=self.course.display_name_with_default_escaped,
assessment=self.assessment,
used_attempts=2,
allowed_attempts=self.allowed_attempts + 1
),
body
)
self.check_courseware_link_exists(body)
def test_denied_email_message_with_close_verification_dates(self):
# Due date given and expired
return_value = datetime.now(tz=pytz.UTC) + timedelta(days=22)
with patch.object(timezone, 'now', return_value=return_value):
__, body = _compose_message_reverification_email(
self.course.id, self.user.id, self.reverification_location, "denied", self.request
)
self.assertIn(
"We could not verify your identity for the {assessment} assessment "
"in the {course_name} course. You have used "
"{used_attempts} out of {allowed_attempts} attempts to "
"verify your identity, and verification is no longer "
"possible".format(
course_name=self.course.display_name_with_default_escaped,
assessment=self.assessment,
used_attempts=1,
allowed_attempts=self.allowed_attempts + 1
),
body
)
def test_check_num_queries(self):
# Get the re-verification block to check the call made
with check_mongo_calls(1):
ver_block = modulestore().get_item(self.reverification.location)
# Expect that the verification block is fetched
self.assertIsNotNone(ver_block)
def check_courseware_link_exists(self, body):
"""Checking courseware url and signature information of EDX"""
self.assertIn(
"To go to the courseware, select the following link:",
body
)
self.assertIn(
"{course_link}".format(
course_link=self.course_link
),
body
)
self.assertIn("Thanks,", body)
self.assertIn(
u"The {platform_name} team".format(
platform_name=settings.PLATFORM_NAME
),
body
)
@attr(shard=2)
class TestEmailMessageWithDefaultICRVBlock(ModuleStoreTestCase):
"""
Test for In-course Re-verification
"""
def build_course(self):
"""
Build up a course tree with a Reverificaiton xBlock.
"""
self.course_key = SlashSeparatedCourseKey("Robot", "999", "Test_Course")
self.course = CourseFactory.create(org='Robot', number='999', display_name='Test Course')
# Create the course modes
for mode in ('audit', 'honor', 'verified'):
min_price = 0 if mode in ["honor", "audit"] else 1
CourseModeFactory.create(mode_slug=mode, course_id=self.course_key, min_price=min_price)
# Create the 'edx-reverification-block' in course tree
section = ItemFactory.create(parent=self.course, category='chapter', display_name='Test Section')
subsection = ItemFactory.create(parent=section, category='sequential', display_name='Test Subsection')
vertical = ItemFactory.create(parent=subsection, category='vertical', display_name='Test Unit')
self.reverification = ItemFactory.create(
parent=vertical,
category='edx-reverification-block',
display_name='Test Verification Block'
)
self.section_location = section.location
self.subsection_location = subsection.location
self.vertical_location = vertical.location
self.reverification_location = unicode(self.reverification.location)
self.assessment = self.reverification.related_assessment
self.re_verification_link = reverse(
'verify_student_incourse_reverify',
args=(
unicode(self.course_key),
self.reverification_location
)
)
def setUp(self):
super(TestEmailMessageWithDefaultICRVBlock, self).setUp()
self.build_course()
self.check_point = VerificationCheckpoint.objects.create(
course_id=self.course.id, checkpoint_location=self.reverification_location
)
self.check_point.add_verification_attempt(SoftwareSecurePhotoVerification.objects.create(user=self.user))
self.attempt = SoftwareSecurePhotoVerification.objects.filter(user=self.user)
self.request = RequestFactory().get('/url')
def test_denied_email_message_with_no_attempt_allowed(self):
VerificationStatus.add_verification_status(
checkpoint=self.check_point,
user=self.user,
status='submitted'
)
__, body = _compose_message_reverification_email(
self.course.id, self.user.id, self.reverification_location, "denied", self.request
)
self.assertIn(
"We could not verify your identity for the {assessment} assessment "
"in the {course_name} course. You have used "
"{used_attempts} out of {allowed_attempts} attempts to "
"verify your identity, and verification is no longer "
"possible".format(
course_name=self.course.display_name_with_default_escaped,
assessment=self.assessment,
used_attempts=1,
allowed_attempts=1
),
body
)
def test_error_on_compose_email(self):
resp = _compose_message_reverification_email(
self.course.id, self.user.id, self.reverification_location, "denied", True
)
self.assertIsNone(resp)

12
lms/djangoapps/verify_student/urls.py
View File

@@ -105,18 +105,6 @@ urlpatterns = patterns(
views.ReverifyView.as_view(),
name="verify_student_reverify"
),
# Endpoint for in-course reverification
# Users are sent to this end-point from within courseware
# to re-verify their identities by re-submitting face photos.
url(
r'^reverify/{course_id}/{usage_id}/$'.format(
course_id=settings.COURSE_ID_PATTERN,
usage_id=settings.USAGE_ID_PATTERN
),
views.InCourseReverifyView.as_view(),
name="verify_student_incourse_reverify"
),
)
# Fake response page for incourse reverification ( software secure )

339
lms/djangoapps/verify_student/views.py
View File

@@ -6,7 +6,6 @@ import datetime
import decimal
import json
import logging
import urllib
from pytz import UTC
from ipware.ip import get_ip
@@ -16,9 +15,7 @@ from django.core.mail import send_mail
from django.core.urlresolvers import reverse
from django.db import transaction
from django.http import HttpResponse, HttpResponseBadRequest, Http404
from django.contrib.auth.models import User
from django.shortcuts import redirect
from django.utils import timezone
from django.utils.decorators import method_decorator
from django.utils.translation import ugettext as _, ugettext_lazy
from django.views.decorators.csrf import csrf_exempt
@@ -28,7 +25,7 @@ from django.views.generic.base import View
import analytics
from eventtracking import tracker
from opaque_keys import InvalidKeyError
from opaque_keys.edx.keys import CourseKey, UsageKey
from opaque_keys.edx.keys import CourseKey
from commerce.utils import EcommerceService
from course_modes.models import CourseMode
@@ -40,7 +37,6 @@ from openedx.core.djangoapps.commerce.utils import ecommerce_api_client
from openedx.core.djangoapps.user_api.accounts import NAME_MIN_LENGTH
from openedx.core.djangoapps.user_api.accounts.api import update_account_settings
from openedx.core.djangoapps.user_api.errors import UserNotFound, AccountValidationError
from openedx.core.djangoapps.credit.api import set_credit_requirement_status
from openedx.core.djangoapps.site_configuration import helpers as configuration_helpers
from openedx.core.lib.log_utils import audit_log
from student.models import CourseEnrollment
@@ -52,13 +48,9 @@ from lms.djangoapps.verify_student.ssencrypt import has_valid_signature
from lms.djangoapps.verify_student.models import (
VerificationDeadline,
SoftwareSecurePhotoVerification,
VerificationCheckpoint,
VerificationStatus,
IcrvStatusEmailsConfiguration,
)
from lms.djangoapps.verify_student.image import decode_image_data, InvalidImageData
from util.json_request import JsonResponse
from util.date_utils import get_default_time_display
from util.db import outer_atomic
from xmodule.modulestore.django import modulestore
from django.contrib.staticfiles.storage import staticfiles_storage
@@ -856,9 +848,7 @@ class SubmitPhotosView(View):
"""
# If the user already has an initial verification attempt, we can re-use the photo ID
# the user submitted with the initial attempt. This is useful for the in-course reverification
# case in which users submit only the face photo and have it matched against their ID photos
# submitted with the initial verification.
# the user submitted with the initial attempt.
initial_verification = SoftwareSecurePhotoVerification.get_initial_verification(request.user)
# Validate the POST parameters
@@ -889,35 +879,9 @@ class SubmitPhotosView(View):
# Submit the attempt
attempt = self._submit_attempt(request.user, face_image, photo_id_image, initial_verification)
# If this attempt was submitted at a checkpoint, then associate
# the attempt with the checkpoint.
submitted_at_checkpoint = "checkpoint" in params and "course_key" in params
if submitted_at_checkpoint:
checkpoint = self._associate_attempt_with_checkpoint(
request.user, attempt,
params["course_key"],
params["checkpoint"]
)
# If the submission came from an in-course checkpoint
if initial_verification is not None and submitted_at_checkpoint:
self._fire_event(request.user, "edx.bi.reverify.submitted", {
"category": "verification",
"label": unicode(params["course_key"]),
"checkpoint": checkpoint.checkpoint_name,
})
# Send a URL that the client can redirect to in order
# to return to the checkpoint in the courseware.
redirect_url = get_redirect_url(params["course_key"], params["checkpoint"])
return JsonResponse({"url": redirect_url})
# Otherwise, the submission came from an initial verification flow.
else:
self._fire_event(request.user, "edx.bi.verify.submitted", {"category": "verification"})
self._send_confirmation_email(request.user)
redirect_url = None
return JsonResponse({})
self._fire_event(request.user, "edx.bi.verify.submitted", {"category": "verification"})
self._send_confirmation_email(request.user)
return JsonResponse({})
def _validate_parameters(self, request, has_initial_verification):
"""
@@ -938,7 +902,6 @@ class SubmitPhotosView(View):
"face_image",
"photo_id_image",
"course_key",
"checkpoint",
"full_name"
]
if param_name in request.POST
@@ -974,14 +937,6 @@ class SubmitPhotosView(View):
except InvalidKeyError:
return None, HttpResponseBadRequest(_("Invalid course key"))
if "checkpoint" in params:
try:
params["checkpoint"] = UsageKey.from_string(params["checkpoint"]).replace(
course_key=params["course_key"]
)
except InvalidKeyError:
return None, HttpResponseBadRequest(_("Invalid checkpoint location"))
return params, None
def _update_full_name(self, user, full_name):
@@ -1070,24 +1025,6 @@ class SubmitPhotosView(View):
return attempt
def _associate_attempt_with_checkpoint(self, user, attempt, course_key, usage_id):
"""
Associate the verification attempt with a checkpoint within a course.
Arguments:
user (User): The user making the attempt.
attempt (SoftwareSecurePhotoVerification): The verification attempt.
course_key (CourseKey): The identifier for the course.
usage_key (UsageKey): The location of the checkpoint within the course.
Returns:
VerificationCheckpoint
"""
checkpoint = VerificationCheckpoint.get_or_create_verification_checkpoint(course_key, usage_id)
checkpoint.add_verification_attempt(attempt)
VerificationStatus.add_verification_status(checkpoint, user, "submitted")
return checkpoint
def _send_confirmation_email(self, user):
"""
Send an email confirming that the user submitted photos
@@ -1134,125 +1071,6 @@ class SubmitPhotosView(View):
analytics.track(user.id, event_name, parameters, context=context)
def _compose_message_reverification_email(
course_key, user_id, related_assessment_location, status, request
): # pylint: disable=invalid-name
"""
Compose subject and message for photo reverification email.
Args:
course_key(CourseKey): CourseKey object
user_id(str): User Id
related_assessment_location(str): Location of reverification XBlock
photo_verification(QuerySet): Queryset of SoftwareSecure objects
status(str): Approval status
is_secure(Bool): Is running on secure protocol or not
Returns:
None if any error occurred else Tuple of subject and message strings
"""
try:
usage_key = UsageKey.from_string(related_assessment_location)
reverification_block = modulestore().get_item(usage_key)
course = modulestore().get_course(course_key)
redirect_url = get_redirect_url(course_key, usage_key.replace(course_key=course_key))
subject = "Re-verification Status"
context = {
"status": status,
"course_name": course.display_name_with_default_escaped,
"assessment": reverification_block.related_assessment
}
# Allowed attempts is 1 if not set on verification block
allowed_attempts = reverification_block.attempts + 1
used_attempts = VerificationStatus.get_user_attempts(user_id, course_key, related_assessment_location)
left_attempts = allowed_attempts - used_attempts
is_attempt_allowed = left_attempts > 0
verification_open = True
if reverification_block.due:
verification_open = timezone.now() <= reverification_block.due
context["left_attempts"] = left_attempts
context["is_attempt_allowed"] = is_attempt_allowed
context["verification_open"] = verification_open
context["due_date"] = get_default_time_display(reverification_block.due)
context['platform_name'] = configuration_helpers.get_value('PLATFORM_NAME', settings.PLATFORM_NAME)
context["used_attempts"] = used_attempts
context["allowed_attempts"] = allowed_attempts
context["support_link"] = configuration_helpers.get_value('email_from_address', settings.CONTACT_EMAIL)
re_verification_link = reverse(
'verify_student_incourse_reverify',
args=(
unicode(course_key),
related_assessment_location
)
)
context["course_link"] = request.build_absolute_uri(redirect_url)
context["reverify_link"] = request.build_absolute_uri(re_verification_link)
message = render_to_string('emails/reverification_processed.txt', context)
log.info(
"Sending email to User_Id=%s. Attempts left for this user are %s. "
"Allowed attempts %s. "
"Due Date %s",
str(user_id), left_attempts, allowed_attempts, str(reverification_block.due)
)
return subject, message
# Catch all exception to avoid raising back to view
except: # pylint: disable=bare-except
log.exception("The email for re-verification sending failed for user_id %s", user_id)
def _send_email(user_id, subject, message):
""" Send email to given user
Args:
user_id(str): User Id
subject(str): Subject lines of emails
message(str): Email message body
Returns:
None
"""
from_address = configuration_helpers.get_value(
'email_from_address',
settings.DEFAULT_FROM_EMAIL
)
user = User.objects.get(id=user_id)
user.email_user(subject, message, from_address)
def _set_user_requirement_status(attempt, namespace, status, reason=None):
"""Sets the status of a credit requirement for the user,
based on a verification checkpoint.
"""
checkpoint = None
try:
checkpoint = VerificationCheckpoint.objects.get(photo_verification=attempt)
except VerificationCheckpoint.DoesNotExist:
log.error("Unable to find checkpoint for user with id %d", attempt.user.id)
if checkpoint is not None:
try:
set_credit_requirement_status(
attempt.user,
checkpoint.course_id,
namespace,
checkpoint.checkpoint_location,
status=status,
reason=reason,
)
except Exception: # pylint: disable=broad-except
# Catch exception if unable to add credit requirement
# status for user
log.error("Unable to add Credit requirement status for user with id %d", attempt.user.id)
@require_POST
@csrf_exempt # SS does its own message signing, and their API won't have a cookie value
def results_callback(request):
@@ -1310,15 +1128,11 @@ def results_callback(request):
log.debug("Approving verification for %s", receipt_id)
attempt.approve()
status = "approved"
_set_user_requirement_status(attempt, 'reverification', 'satisfied')
elif result == "FAIL":
log.debug("Denying verification for %s", receipt_id)
attempt.deny(json.dumps(reason), error_code=error_code)
status = "denied"
_set_user_requirement_status(
attempt, 'reverification', 'failed', json.dumps(reason)
)
elif result == "SYSTEM FAIL":
log.debug("System failure for %s -- resetting to must_retry", receipt_id)
attempt.system_error(json.dumps(reason), error_code=error_code)
@@ -1330,22 +1144,6 @@ def results_callback(request):
"Result {} not understood. Known results: PASS, FAIL, SYSTEM FAIL".format(result)
)
checkpoints = VerificationCheckpoint.objects.filter(photo_verification=attempt).all()
VerificationStatus.add_status_from_checkpoints(checkpoints=checkpoints, user=attempt.user, status=status)
# Trigger ICRV email only if ICRV status emails config is enabled
icrv_status_emails = IcrvStatusEmailsConfiguration.current()
if icrv_status_emails.enabled and checkpoints:
user_id = attempt.user.id
course_key = checkpoints[0].course_id
related_assessment_location = checkpoints[0].checkpoint_location
subject, message = _compose_message_reverification_email(
course_key, user_id, related_assessment_location, status, request
)
_send_email(user_id, subject, message)
return HttpResponse("OK!")
@@ -1398,130 +1196,3 @@ class ReverifyView(View):
"status": status
}
return render_to_response("verify_student/reverify_not_allowed.html", context)
class InCourseReverifyView(View):
"""
The in-course reverification view.
In-course reverification occurs while a student is taking a course.
At points in the course, students are prompted to submit face photos,
which are matched against the ID photos the user submitted during their
initial verification.
Students are prompted to enter this flow from an "In Course Reverification"
XBlock (courseware component) that course authors add to the course.
See https://github.com/edx/edx-reverification-block for more details.
"""
@method_decorator(login_required)
def get(self, request, course_id, usage_id):
"""Display the view for face photo submission.
Args:
request(HttpRequest): HttpRequest object
course_id(str): A string of course id
usage_id(str): Location of Reverification XBlock in courseware
Returns:
HttpResponse
"""
user = request.user
course_key = CourseKey.from_string(course_id)
course = modulestore().get_course(course_key)
if course is None:
log.error(u"Could not find course '%s' for in-course reverification.", course_key)
raise Http404
try:
checkpoint = VerificationCheckpoint.objects.get(course_id=course_key, checkpoint_location=usage_id)
except VerificationCheckpoint.DoesNotExist:
log.error(
u"No verification checkpoint exists for the "
u"course '%s' and checkpoint location '%s'.",
course_key, usage_id
)
raise Http404
initial_verification = SoftwareSecurePhotoVerification.get_initial_verification(user)
if not initial_verification:
return self._redirect_to_initial_verification(user, course_key, usage_id)
# emit the reverification event
self._track_reverification_events('edx.bi.reverify.started', user.id, course_id, checkpoint.checkpoint_name)
context = {
'course_key': unicode(course_key),
'course_name': course.display_name_with_default_escaped,
'checkpoint_name': checkpoint.checkpoint_name,
'platform_name': configuration_helpers.get_value('PLATFORM_NAME', settings.PLATFORM_NAME),
'usage_id': usage_id,
'capture_sound': staticfiles_storage.url("audio/camera_capture.wav"),
}
return render_to_response("verify_student/incourse_reverify.html", context)
def _track_reverification_events(self, event_name, user_id, course_id, checkpoint):
"""Track re-verification events for a user against a reverification
checkpoint of a course.
Arguments:
event_name (str): Name of event being tracked
user_id (str): The ID of the user
course_id (unicode): ID associated with the course
checkpoint (str): Checkpoint name
Returns:
None
"""
log.info(
u"In-course reverification: event %s occurred for user '%s' in course '%s' at checkpoint '%s'",
event_name, user_id, course_id, checkpoint
)
if settings.LMS_SEGMENT_KEY:
tracking_context = tracker.get_tracker().resolve_context()
analytics.track(
user_id,
event_name,
{
'category': "verification",
'label': unicode(course_id),
'checkpoint': checkpoint
},
context={
'ip': tracking_context.get('ip'),
'Google Analytics': {
'clientId': tracking_context.get('client_id')
}
}
)
def _redirect_to_initial_verification(self, user, course_key, checkpoint):
"""
Redirect because the user does not have an initial verification.
We will redirect the user to the initial verification flow,
passing the identifier for this checkpoint. When the user
submits a verification attempt, it will count for *both*
the initial and checkpoint verification.
Arguments:
user (User): The user who made the request.
course_key (CourseKey): The identifier for the course for which
the user is attempting to re-verify.
checkpoint (string): Location of the checkpoint in the courseware.
Returns:
HttpResponse
"""
log.info(
u"User %s does not have an initial verification, so "
u"he/she will be redirected to the \"verify later\" flow "
u"for the course %s.",
user.id, course_key
)
base_url = reverse('verify_student_verify_now', kwargs={'course_id': unicode(course_key)})
params = urllib.urlencode({"checkpoint": checkpoint})
full_url = u"{base}?{params}".format(base=base_url, params=params)
return redirect(full_url)

41
openedx/core/djangoapps/credit/api/eligibility.py
View File

@@ -47,12 +47,6 @@ def set_credit_requirements(course_key, requirements):
>>> set_credit_requirements(
"course-v1-edX-DemoX-1T2015",
[
{
"namespace": "reverification",
"name": "i4x://edX/DemoX/edx-reverification-block/assessment_uuid",
"display_name": "Assessment 1",
"criteria": {},
},
{
"namespace": "proctored_exam",
"name": "i4x://edX/DemoX/proctoring-block/final_uuid",
@@ -106,12 +100,6 @@ def get_credit_requirements(course_key, namespace=None):
{
requirements =
[
{
"namespace": "reverification",
"name": "i4x://edX/DemoX/edx-reverification-block/assessment_uuid",
"display_name": "Assessment 1",
"criteria": {},
},
{
"namespace": "proctored_exam",
"name": "i4x://edX/DemoX/proctoring-block/final_uuid",
@@ -216,17 +204,6 @@ def set_credit_requirement_status(user, course_key, req_namespace, req_name, sta
Keyword Arguments:
status (str): Status of the requirement (either "satisfied" or "failed")
reason (dict): Reason of the status
Example:
>>> set_credit_requirement_status(
"staff",
CourseKey.from_string("course-v1-edX-DemoX-1T2015"),
"reverification",
"i4x://edX/DemoX/edx-reverification-block/assessment_uuid",
status="satisfied",
reason={}
)
"""
# Check whether user has credit eligible enrollment.
enrollment_mode, is_active = CourseEnrollment.enrollment_mode_for_user(user, course_key)
@@ -317,14 +294,6 @@ def remove_credit_requirement_status(username, course_key, req_namespace, req_na
req_name (str): Name of the requirement
(e.g. "grade" or the location of the ICRV XBlock)
Example:
>>> remove_credit_requirement_status(
"staff",
CourseKey.from_string("course-v1-edX-DemoX-1T2015"),
"reverification",
"i4x://edX/DemoX/edx-reverification-block/assessment_uuid".
)
"""
# Find the requirement we're trying to remove
@@ -364,16 +333,6 @@ def get_credit_requirement_status(course_key, username, namespace=None, name=Non
>>> get_credit_requirement_status("course-v1-edX-DemoX-1T2015", "john")
[
{
"namespace": "reverification",
"name": "i4x://edX/DemoX/edx-reverification-block/assessment_uuid",
"display_name": "In Course Reverification",
"criteria": {},
"reason": {},
"status": "failed",
"status_date": "2015-06-26 07:49:13",
"order": 0,
},
{
"namespace": "proctored_exam",
"name": "i4x://edX/DemoX/proctoring-block/final_uuid",

136
openedx/core/djangoapps/credit/partition_schemes.py
View File

@@ -1,136 +0,0 @@
"""
Partition scheme for in-course reverification.
This is responsible for placing users into one of two groups,
ALLOW or DENY, for a partition associated with a particular
in-course reverification checkpoint.
NOTE: This really should be defined in the verify_student app,
which owns the verification and reverification process.
It isn't defined there now because (a) we need access to this in both Studio
and the LMS, but verify_student is specific to the LMS, and
(b) in-course reverification checkpoints currently have messaging that's
specific to credit requirements.
"""
import logging
from django.core.cache import cache
from lms.djangoapps.verify_student.models import SkippedReverification, VerificationStatus
from student.models import CourseEnrollment
from xmodule.partitions.partitions import NoSuchUserPartitionGroupError
log = logging.getLogger(__name__)
class VerificationPartitionScheme(object):
"""
Assign users to groups for a particular verification checkpoint.
Users in the ALLOW group can see gated content;
users in the DENY group cannot.
"""
DENY = 0
ALLOW = 1
@classmethod
def get_group_for_user(cls, course_key, user, user_partition, **kwargs): # pylint: disable=unused-argument
"""
Return the user's group depending their enrollment and verification
status.
Args:
course_key (CourseKey): CourseKey
user (User): user object
user_partition (UserPartition): The user partition object.
Returns:
string of allowed access group
"""
checkpoint = user_partition.parameters['location']
# Retrieve all information we need to determine the user's group
# as a multi-get from the cache.
is_verified, has_skipped, has_completed = _get_user_statuses(user, course_key, checkpoint)
# Decide whether the user should have access to content gated by this checkpoint.
# Intuitively, we allow access if the user doesn't need to do anything at the checkpoint,
# either because the user is in a non-verified track or the user has already submitted.
#
# Note that we do NOT wait the user's reverification attempt to be approved,
# since this can take some time and the user might miss an assignment deadline.
partition_group = cls.DENY
if not is_verified or has_skipped or has_completed:
partition_group = cls.ALLOW
# Return matching user partition group if it exists
try:
return user_partition.get_group(partition_group)
except NoSuchUserPartitionGroupError:
log.error(
(
u"Could not find group with ID %s for verified partition "
"with ID %s in course %s. The user will not be assigned a group."
),
partition_group,
user_partition.id,
course_key
)
return None
def _get_user_statuses(user, course_key, checkpoint):
"""
Retrieve all the information we need to determine the user's group.
This will retrieve the information as a multi-get from the cache.
Args:
user (User): User object
course_key (CourseKey): Identifier for the course.
checkpoint (unicode): Location of the checkpoint in the course (serialized usage key)
Returns:
tuple of booleans of the form (is_verified, has_skipped, has_completed)
"""
enrollment_cache_key = CourseEnrollment.cache_key_name(user.id, unicode(course_key))
has_skipped_cache_key = SkippedReverification.cache_key_name(user.id, unicode(course_key))
verification_status_cache_key = VerificationStatus.cache_key_name(user.id, unicode(course_key))
# Try a multi-get from the cache
cache_values = cache.get_many([
enrollment_cache_key,
has_skipped_cache_key,
verification_status_cache_key
])
# Retrieve whether the user is enrolled in a verified mode.
is_verified = cache_values.get(enrollment_cache_key)
if is_verified is None:
is_verified = CourseEnrollment.is_enrolled_as_verified(user, course_key)
cache.set(enrollment_cache_key, is_verified)
# Retrieve whether the user has skipped any checkpoints in this course
has_skipped = cache_values.get(has_skipped_cache_key)
if has_skipped is None:
has_skipped = SkippedReverification.check_user_skipped_reverification_exists(user, course_key)
cache.set(has_skipped_cache_key, has_skipped)
# Retrieve the user's verification status for each checkpoint in the course.
verification_statuses = cache_values.get(verification_status_cache_key)
if verification_statuses is None:
verification_statuses = VerificationStatus.get_all_checkpoints(user.id, course_key)
cache.set(verification_status_cache_key, verification_statuses)
# Check whether the user has completed this checkpoint
# "Completion" here means *any* submission, regardless of its status
# since we want to show the user the content if they've submitted
# photos.
checkpoint = verification_statuses.get(checkpoint)
has_completed_check = bool(checkpoint)
return (is_verified, has_skipped, has_completed_check)

20
openedx/core/djangoapps/credit/signals.py
View File

@@ -9,7 +9,6 @@ from django.utils import timezone
from opaque_keys.edx.keys import CourseKey
from xmodule.modulestore.django import SignalHandler
from openedx.core.djangoapps.credit.verification_access import update_verification_partitions
from openedx.core.djangoapps.signals.signals import COURSE_GRADE_CHANGED
log = logging.getLogger(__name__)
@@ -33,25 +32,6 @@ def on_course_publish(course_key):
log.info(u'Added task to update credit requirements for course "%s" to the task queue', course_key)
@receiver(SignalHandler.pre_publish)
def on_pre_publish(sender, course_key, **kwargs): # pylint: disable=unused-argument
"""
Create user partitions for verification checkpoints.
This is a pre-publish step since we need to write to the course descriptor.
"""
from openedx.core.djangoapps.credit import api
if api.is_credit_course(course_key):
# For now, we are tagging content with in-course-reverification access groups
# only in credit courses on publish. In the long run, this is not where we want to put this.
# This really should be a transformation on the course structure performed as a pre-processing
# step by the LMS, and the transformation should be owned by the verify_student app.
# Since none of that infrastructure currently exists, we're doing it this way instead.
log.info(u"Starting to update in-course reverification access rules")
update_verification_partitions(course_key)
log.info(u"Finished updating in-course reverification access rules")
@receiver(COURSE_GRADE_CHANGED)
def listen_for_grade_calculation(sender, user, course_grade, course_key, deadline, **kwargs): # pylint: disable=unused-argument
"""Receive 'MIN_GRADE_REQUIREMENT_STATUS' signal and update minimum grade requirement status.

86
openedx/core/djangoapps/credit/tasks.py
View File

@@ -18,12 +18,6 @@ from xmodule.modulestore.exceptions import ItemNotFoundError
LOGGER = get_task_logger(__name__)
# XBlocks that can be added as credit requirements
CREDIT_REQUIREMENT_XBLOCK_CATEGORIES = [
"edx-reverification-block",
]
# pylint: disable=not-callable
@task(default_retry_delay=settings.CREDIT_TASK_DEFAULT_RETRY_DELAY, max_retries=settings.CREDIT_TASK_MAX_RETRIES)
def update_credit_course_requirements(course_id): # pylint: disable=invalid-name
@@ -67,18 +61,14 @@ def _get_course_credit_requirements(course_key):
List of credit requirements (dictionaries)
"""
credit_xblock_requirements = _get_credit_course_requirement_xblocks(course_key)
min_grade_requirement = _get_min_grade_requirement(course_key)
proctored_exams_requirements = _get_proctoring_requirements(course_key)
block_requirements = credit_xblock_requirements + proctored_exams_requirements
# sort credit requirements list based on start date and put all the
# requirements with no start date at the end of requirement list.
sorted_block_requirements = sorted(
block_requirements, key=lambda x: (x['start_date'] is None, x['start_date'], x['display_name'])
sorted_exam_requirements = sorted(
proctored_exams_requirements, key=lambda x: (x['start_date'] is None, x['start_date'], x['display_name'])
)
credit_requirements = (
min_grade_requirement + sorted_block_requirements
min_grade_requirement + sorted_exam_requirements
)
return credit_requirements
@@ -112,76 +102,6 @@ def _get_min_grade_requirement(course_key):
return []
def _get_credit_course_requirement_xblocks(course_key): # pylint: disable=invalid-name
"""Generate a course structure dictionary for the specified course.
Args:
course_key (CourseKey): Identifier for the course.
Returns:
The list of credit requirements xblocks dicts
"""
requirements = []
# Retrieve all XBlocks from the course that we know to be credit requirements.
# For performance reasons, we look these up by their "category" to avoid
# loading and searching the entire course tree.
for category in CREDIT_REQUIREMENT_XBLOCK_CATEGORIES:
requirements.extend([
{
"namespace": block.get_credit_requirement_namespace(),
"name": block.get_credit_requirement_name(),
"display_name": block.get_credit_requirement_display_name(),
'start_date': block.start,
"criteria": {},
}
for block in _get_xblocks(course_key, category)
if _is_credit_requirement(block)
])
return requirements
def _get_xblocks(course_key, category):
"""
Retrieve all XBlocks in the course for a particular category.
Returns only XBlocks that are published and haven't been deleted.
"""
xblocks = get_course_blocks(course_key, category)
return xblocks
def _is_credit_requirement(xblock):
"""
Check if the given XBlock is a credit requirement.
Args:
xblock(XBlock): The given XBlock object
Returns:
True if XBlock is a credit requirement else False
"""
required_methods = [
"get_credit_requirement_namespace",
"get_credit_requirement_name",
"get_credit_requirement_display_name"
]
for method_name in required_methods:
if not callable(getattr(xblock, method_name, None)):
LOGGER.error(
"XBlock %s is marked as a credit requirement but does not "
"implement %s", unicode(xblock), method_name
)
return False
return True
def _get_proctoring_requirements(course_key):
"""
Will return list of requirements regarding any exams that have been

34
openedx/core/djangoapps/credit/tests/test_api.py
View File

@@ -287,7 +287,7 @@ class CreditRequirementApiTests(CreditApiTestBase):
# Set initial requirements
requirements = [
{
"namespace": "reverification",
"namespace": "grade",
"name": "midterm",
"display_name": "Midterm",
"criteria": {},
@@ -328,8 +328,8 @@ class CreditRequirementApiTests(CreditApiTestBase):
requirements = [
{
"namespace": "reverification",
"name": "i4x://edX/DemoX/edx-reverification-block/assessment_uuid",
"namespace": "grade",
"name": "other_grade",
"display_name": "Assessment 1",
"criteria": {},
}
@@ -457,8 +457,8 @@ class CreditRequirementApiTests(CreditApiTestBase):
},
},
{
"namespace": "reverification",
"name": "i4x://edX/DemoX/edx-reverification-block/assessment_uuid",
"namespace": "grade",
"name": "other_grade",
"display_name": "Assessment 1",
"criteria": {},
}
@@ -499,15 +499,15 @@ class CreditRequirementApiTests(CreditApiTestBase):
# Set the requirement to "declined" and check that it's actually set
api.set_credit_requirement_status(
self.user, self.course_key,
"reverification",
"i4x://edX/DemoX/edx-reverification-block/assessment_uuid",
"grade",
"other_grade",
status="declined"
)
req_status = api.get_credit_requirement_status(
self.course_key,
username,
namespace="reverification",
name="i4x://edX/DemoX/edx-reverification-block/assessment_uuid"
namespace="grade",
name="other_grade"
)
self.assertEqual(req_status[0]["status"], "declined")
@@ -528,8 +528,8 @@ class CreditRequirementApiTests(CreditApiTestBase):
},
},
{
"namespace": "reverification",
"name": "i4x://edX/DemoX/edx-reverification-block/assessment_uuid",
"namespace": "grade",
"name": "other_grade",
"display_name": "Assessment 1",
"criteria": {},
}
@@ -600,8 +600,8 @@ class CreditRequirementApiTests(CreditApiTestBase):
},
},
{
"namespace": "reverification",
"name": "i4x://edX/DemoX/edx-reverification-block/assessment_uuid",
"namespace": "grade",
"name": "other_grade",
"display_name": "Assessment 1",
"criteria": {},
}
@@ -727,8 +727,8 @@ class CreditRequirementApiTests(CreditApiTestBase):
},
},
{
"namespace": "reverification",
"name": "i4x://edX/DemoX/edx-reverification-block/assessment_uuid",
"namespace": "grade",
"name": "other_grade",
"display_name": "Assessment 1",
"criteria": {},
}
@@ -790,8 +790,8 @@ class CreditRequirementApiTests(CreditApiTestBase):
},
},
{
"namespace": "reverification",
"name": "i4x://edX/DemoX/edx-reverification-block/assessment_uuid",
"namespace": "grade",
"name": "other_grade",
"display_name": "Assessment 1",
"criteria": {},
}

6
openedx/core/djangoapps/credit/tests/test_models.py
View File

@@ -61,9 +61,9 @@ class CreditEligibilityModelTests(TestCase):
self.assertEqual(created, True)
requirement = {
"namespace": "reverification",
"name": "i4x://edX/DemoX/edx-reverification-block/assessment_uuid",
"display_name": "Assessment 1",
"namespace": "new_grade",
"name": "new_grade",
"display_name": "New Grade",
"criteria": {},
}
credit_req, created = CreditRequirement.add_or_update_course_requirement(credit_course, requirement, 1)

182
openedx/core/djangoapps/credit/tests/test_partition.py
View File

@@ -1,182 +0,0 @@
# -*- coding: utf-8 -*-
"""
Tests for In-Course Reverification Access Control Partition scheme
"""
import ddt
from nose.plugins.attrib import attr
from lms.djangoapps.verify_student.models import (
VerificationCheckpoint,
VerificationStatus,
SkippedReverification,
)
from openedx.core.djangoapps.credit.partition_schemes import VerificationPartitionScheme
from openedx.core.djangolib.testing.utils import skip_unless_lms
from student.models import CourseEnrollment
from student.tests.factories import UserFactory
from xmodule.partitions.partitions import UserPartition, Group
from xmodule.modulestore.tests.django_utils import ModuleStoreTestCase
from xmodule.modulestore.tests.factories import CourseFactory
@attr(shard=2)
@ddt.ddt
@skip_unless_lms
class ReverificationPartitionTest(ModuleStoreTestCase):
"""Tests for the Reverification Partition Scheme. """
SUBMITTED = "submitted"
APPROVED = "approved"
DENIED = "denied"
ENABLED_CACHES = ['default', 'mongo_metadata_inheritance', 'loc_cache']
def setUp(self):
super(ReverificationPartitionTest, self).setUp()
# creating course, checkpoint location and user partition mock object.
self.course = CourseFactory.create()
self.checkpoint_location = u'i4x://{org}/{course}/edx-reverification-block/first_uuid'.format(
org=self.course.id.org, course=self.course.id.course
)
scheme = UserPartition.get_scheme("verification")
self.user_partition = UserPartition(
id=0,
name=u"Verification Checkpoint",
description=u"Verification Checkpoint",
scheme=scheme,
parameters={"location": self.checkpoint_location},
groups=[
Group(scheme.ALLOW, "Allow access to content"),
Group(scheme.DENY, "Deny access to content"),
]
)
self.first_checkpoint = VerificationCheckpoint.objects.create(
course_id=self.course.id,
checkpoint_location=self.checkpoint_location
)
def create_user_and_enroll(self, enrollment_type):
"""Create and enroll users with provided enrollment type."""
user = UserFactory.create()
CourseEnrollment.objects.create(
user=user,
course_id=self.course.id,
mode=enrollment_type,
is_active=True
)
return user
def add_verification_status(self, user, status):
"""Adding the verification status for a user."""
VerificationStatus.add_status_from_checkpoints(
checkpoints=[self.first_checkpoint],
user=user,
status=status
)
@ddt.data(
("verified", SUBMITTED, VerificationPartitionScheme.ALLOW),
("verified", APPROVED, VerificationPartitionScheme.ALLOW),
("verified", DENIED, VerificationPartitionScheme.ALLOW),
("verified", None, VerificationPartitionScheme.DENY),
("honor", None, VerificationPartitionScheme.ALLOW),
)
@ddt.unpack
def test_get_group_for_user(self, enrollment_type, verification_status, expected_group):
# creating user and enroll them.
user = self.create_user_and_enroll(enrollment_type)
if verification_status:
self.add_verification_status(user, verification_status)
self._assert_group_assignment(user, expected_group)
def test_get_group_for_user_with_skipped(self):
# Check that a user is in verified allow group if that user has skipped
# any ICRV block.
user = self.create_user_and_enroll('verified')
SkippedReverification.add_skipped_reverification_attempt(
checkpoint=self.first_checkpoint,
user_id=user.id,
course_id=self.course.id
)
self._assert_group_assignment(user, VerificationPartitionScheme.ALLOW)
def test_cache_with_skipped_icrv(self):
# Check that a user is in verified allow group if that user has skipped
# any ICRV block.
user = self.create_user_and_enroll('verified')
SkippedReverification.add_skipped_reverification_attempt(
checkpoint=self.first_checkpoint,
user_id=user.id,
course_id=self.course.id
)
# this will warm the cache.
with self.assertNumQueries(3):
self._assert_group_assignment(user, VerificationPartitionScheme.ALLOW)
# no db queries this time.
with self.assertNumQueries(0):
self._assert_group_assignment(user, VerificationPartitionScheme.ALLOW)
def test_cache_with_submitted_status(self):
# Check that a user is in verified allow group if that user has approved status at
# any ICRV block.
user = self.create_user_and_enroll('verified')
self.add_verification_status(user, VerificationStatus.APPROVED_STATUS)
# this will warm the cache.
with self.assertNumQueries(4):
self._assert_group_assignment(user, VerificationPartitionScheme.ALLOW)
# no db queries this time.
with self.assertNumQueries(0):
self._assert_group_assignment(user, VerificationPartitionScheme.ALLOW)
def test_cache_with_denied_status(self):
# Check that a user is in verified allow group if that user has denied at
# any ICRV block.
user = self.create_user_and_enroll('verified')
self.add_verification_status(user, VerificationStatus.DENIED_STATUS)
# this will warm the cache.
with self.assertNumQueries(4):
self._assert_group_assignment(user, VerificationPartitionScheme.ALLOW)
# no db queries this time.
with self.assertNumQueries(0):
self._assert_group_assignment(user, VerificationPartitionScheme.ALLOW)
def test_cache_with_honor(self):
# Check that a user is in honor mode.
# any ICRV block.
user = self.create_user_and_enroll('honor')
# this will warm the cache.
with self.assertNumQueries(3):
self._assert_group_assignment(user, VerificationPartitionScheme.ALLOW)
# no db queries this time.
with self.assertNumQueries(0):
self._assert_group_assignment(user, VerificationPartitionScheme.ALLOW)
def test_cache_with_verified_deny_group(self):
# Check that a user is in verified mode. But not perform any action
user = self.create_user_and_enroll('verified')
# this will warm the cache.
with self.assertNumQueries(3):
self._assert_group_assignment(user, VerificationPartitionScheme.DENY)
# no db queries this time.
with self.assertNumQueries(0):
self._assert_group_assignment(user, VerificationPartitionScheme.DENY)
def _assert_group_assignment(self, user, expected_group_id):
"""Check that the user was assigned to a group. """
actual_group = VerificationPartitionScheme.get_group_for_user(self.course.id, user, self.user_partition)
self.assertEqual(actual_group.id, expected_group_id)

114
openedx/core/djangoapps/credit/tests/test_tasks.py
View File

@@ -4,16 +4,14 @@ Tests for credit course tasks.
import mock
from nose.plugins.attrib import attr
from datetime import datetime, timedelta
from datetime import datetime
from pytz import UTC
from openedx.core.djangoapps.credit.api import get_credit_requirements
from openedx.core.djangoapps.credit.exceptions import InvalidCreditRequirements
from openedx.core.djangoapps.credit.models import CreditCourse
from openedx.core.djangoapps.credit.signals import on_course_publish
from xmodule.modulestore import ModuleStoreEnum
from xmodule.modulestore.tests.django_utils import ModuleStoreTestCase
from xmodule.modulestore.tests.factories import CourseFactory, ItemFactory, check_mongo_calls_range
from xmodule.modulestore.tests.factories import CourseFactory, ItemFactory
from edx_proctoring.api import create_exam
@@ -34,25 +32,6 @@ class TestTaskExecution(ModuleStoreTestCase):
"""
raise InvalidCreditRequirements
def add_icrv_xblock(self, related_assessment_name=None, start_date=None):
""" Create the 'edx-reverification-block' in course tree """
block = ItemFactory.create(
parent=self.vertical,
category='edx-reverification-block',
)
if related_assessment_name is not None:
block.related_assessment = related_assessment_name
block.start = start_date
self.store.update_item(block, ModuleStoreEnum.UserID.test)
with self.store.branch_setting(ModuleStoreEnum.Branch.draft_preferred, self.course.id):
self.store.publish(block.location, ModuleStoreEnum.UserID.test)
return block
def setUp(self):
super(TestTaskExecution, self).setUp()
@@ -86,19 +65,6 @@ class TestTaskExecution(ModuleStoreTestCase):
requirements = get_credit_requirements(self.course.id)
self.assertEqual(len(requirements), 1)
def test_task_adding_icrv_requirements(self):
"""Make sure that the receiver correctly fires off the task when
invoked by signal.
"""
self.add_credit_course(self.course.id)
self.add_icrv_xblock()
requirements = get_credit_requirements(self.course.id)
self.assertEqual(len(requirements), 0)
on_course_publish(self.course.id)
requirements = get_credit_requirements(self.course.id)
self.assertEqual(len(requirements), 2)
def test_proctored_exam_requirements(self):
"""
Make sure that proctored exams are being registered as requirements
@@ -202,71 +168,6 @@ class TestTaskExecution(ModuleStoreTestCase):
if requirement['namespace'] == 'proctored_exam'
])
def test_query_counts(self):
self.add_credit_course(self.course.id)
self.add_icrv_xblock()
with check_mongo_calls_range(max_finds=11):
on_course_publish(self.course.id)
def test_remove_icrv_requirement(self):
self.add_credit_course(self.course.id)
self.add_icrv_xblock()
on_course_publish(self.course.id)
# There should be one ICRV requirement
requirements = get_credit_requirements(self.course.id, namespace="reverification")
self.assertEqual(len(requirements), 1)
# Delete the parent section containing the ICRV block
with self.store.branch_setting(ModuleStoreEnum.Branch.draft_preferred, self.course.id):
self.store.delete_item(self.subsection.location, ModuleStoreEnum.UserID.test)
# Check that the ICRV block is no longer visible in the requirements
on_course_publish(self.course.id)
requirements = get_credit_requirements(self.course.id, namespace="reverification")
self.assertEqual(len(requirements), 0)
def test_icrv_requirement_ordering(self):
self.add_credit_course(self.course.id)
# Create multiple ICRV blocks
start = datetime.now(UTC)
self.add_icrv_xblock(related_assessment_name="Midterm A", start_date=start)
start = start - timedelta(days=1)
self.add_icrv_xblock(related_assessment_name="Midterm B", start_date=start)
# Primary sort is based on start date
on_course_publish(self.course.id)
requirements = get_credit_requirements(self.course.id, namespace="reverification")
self.assertEqual(len(requirements), 2)
self.assertEqual(requirements[0]["display_name"], "Midterm B")
self.assertEqual(requirements[1]["display_name"], "Midterm A")
# Add two additional ICRV blocks that have no start date
# and the same name.
start = datetime.now(UTC)
first_block = self.add_icrv_xblock(related_assessment_name="Midterm Start Date")
start = start + timedelta(days=1)
second_block = self.add_icrv_xblock(related_assessment_name="Midterm Start Date")
on_course_publish(self.course.id)
requirements = get_credit_requirements(self.course.id, namespace="reverification")
self.assertEqual(len(requirements), 4)
# Since we are now primarily sorting on start_date and display_name if
# start_date is present otherwise we are just sorting on display_name.
self.assertEqual(requirements[0]["display_name"], "Midterm B")
self.assertEqual(requirements[1]["display_name"], "Midterm A")
self.assertEqual(requirements[2]["display_name"], "Midterm Start Date")
self.assertEqual(requirements[3]["display_name"], "Midterm Start Date")
# Since the last two requirements have the same display name,
# we need to also check that their internal names (locations) are the same.
self.assertEqual(requirements[2]["name"], first_block.get_credit_requirement_name())
self.assertEqual(requirements[3]["name"], second_block.get_credit_requirement_name())
@mock.patch(
'openedx.core.djangoapps.credit.tasks.set_credit_requirements',
mock.Mock(
@@ -315,24 +216,15 @@ class TestTaskExecution(ModuleStoreTestCase):
self.assertEqual(requirements[1]['display_name'], 'A Proctored Exam')
self.assertEqual(requirements[1]['criteria'], {})
# Create multiple ICRV blocks
start = datetime.now(UTC)
self.add_icrv_xblock(related_assessment_name="Midterm A", start_date=start)
start = start - timedelta(days=1)
self.add_icrv_xblock(related_assessment_name="Midterm B", start_date=start)
# Primary sort is based on start date
on_course_publish(self.course.id)
requirements = get_credit_requirements(self.course.id)
# grade requirement is added on publish of the requirements
self.assertEqual(len(requirements), 4)
self.assertEqual(len(requirements), 2)
# check requirements are added in the desired order
# 1st Minimum grade then the blocks with start date than other blocks
self.assertEqual(requirements[0]["display_name"], "Minimum Grade")
self.assertEqual(requirements[1]["display_name"], "A Proctored Exam")
self.assertEqual(requirements[2]["display_name"], "Midterm B")
self.assertEqual(requirements[3]["display_name"], "Midterm A")
def add_credit_course(self, course_key):
"""Add the course as a credit.

275
openedx/core/djangoapps/credit/tests/test_verification_access.py
View File

@@ -1,275 +0,0 @@
"""
Tests for in-course reverification user partition creation.
This should really belong to the verify_student app,
but we can't move it there because it's in the LMS and we're
currently applying these rules on publish from Studio.
In the future, this functionality should be a course transformation
defined in the verify_student app, and these tests should be moved
into verify_student.
"""
from mock import patch
from nose.plugins.attrib import attr
from django.conf import settings
from openedx.core.djangoapps.credit.models import CreditCourse
from openedx.core.djangoapps.credit.partition_schemes import VerificationPartitionScheme
from openedx.core.djangoapps.credit.verification_access import update_verification_partitions
from openedx.core.djangoapps.credit.signals import on_pre_publish
from xmodule.modulestore import ModuleStoreEnum
from xmodule.modulestore.django import SignalHandler
from xmodule.modulestore.exceptions import ItemNotFoundError
from xmodule.modulestore.tests.django_utils import ModuleStoreTestCase, TEST_DATA_SPLIT_MODULESTORE
from xmodule.modulestore.tests.factories import CourseFactory, ItemFactory, check_mongo_calls_range
from xmodule.partitions.partitions import Group, UserPartition
@attr(shard=2)
class CreateVerificationPartitionTest(ModuleStoreTestCase):
"""
Tests for applying verification access rules.
"""
# Run the tests in split modulestore
# While verification access will work in old-Mongo, it's not something
# we're committed to supporting, since this feature is meant for use
# in new courses.
MODULESTORE = TEST_DATA_SPLIT_MODULESTORE
@patch.dict(settings.FEATURES, {"ENABLE_COURSEWARE_INDEX": False})
def setUp(self):
super(CreateVerificationPartitionTest, self).setUp()
# Disconnect the signal receiver -- we'll invoke the update code ourselves
SignalHandler.pre_publish.disconnect(receiver=on_pre_publish)
self.addCleanup(SignalHandler.pre_publish.connect, receiver=on_pre_publish)
# Create a dummy course with a single verification checkpoint
# Because we need to check "exam" content surrounding the ICRV checkpoint,
# we need to create a fairly large course structure, with multiple sections,
# subsections, verticals, units, and items.
self.course = CourseFactory()
self.sections = [
ItemFactory.create(parent=self.course, category='chapter', display_name='Test Section A'),
ItemFactory.create(parent=self.course, category='chapter', display_name='Test Section B'),
]
self.subsections = [
ItemFactory.create(parent=self.sections[0], category='sequential', display_name='Test Subsection A 1'),
ItemFactory.create(parent=self.sections[0], category='sequential', display_name='Test Subsection A 2'),
ItemFactory.create(parent=self.sections[1], category='sequential', display_name='Test Subsection B 1'),
ItemFactory.create(parent=self.sections[1], category='sequential', display_name='Test Subsection B 2'),
]
self.verticals = [
ItemFactory.create(parent=self.subsections[0], category='vertical', display_name='Test Unit A 1 a'),
ItemFactory.create(parent=self.subsections[0], category='vertical', display_name='Test Unit A 1 b'),
ItemFactory.create(parent=self.subsections[1], category='vertical', display_name='Test Unit A 2 a'),
ItemFactory.create(parent=self.subsections[1], category='vertical', display_name='Test Unit A 2 b'),
ItemFactory.create(parent=self.subsections[2], category='vertical', display_name='Test Unit B 1 a'),
ItemFactory.create(parent=self.subsections[2], category='vertical', display_name='Test Unit B 1 b'),
ItemFactory.create(parent=self.subsections[3], category='vertical', display_name='Test Unit B 2 a'),
ItemFactory.create(parent=self.subsections[3], category='vertical', display_name='Test Unit B 2 b'),
]
self.icrv = ItemFactory.create(parent=self.verticals[0], category='edx-reverification-block')
self.sibling_problem = ItemFactory.create(parent=self.verticals[0], category='problem')
def test_creates_user_partitions(self):
self._update_partitions()
# Check that a new user partition was created for the ICRV block
self.assertEqual(len(self.course.user_partitions), 1)
partition = self.course.user_partitions[0]
self.assertEqual(partition.scheme.name, "verification")
self.assertEqual(partition.parameters["location"], unicode(self.icrv.location))
# Check that the groups for the partition were created correctly
self.assertEqual(len(partition.groups), 2)
self.assertItemsEqual(
[g.id for g in partition.groups],
[
VerificationPartitionScheme.ALLOW,
VerificationPartitionScheme.DENY,
]
)
@patch.dict(settings.FEATURES, {"ENABLE_COURSEWARE_INDEX": False})
def test_removes_deleted_user_partitions(self):
self._update_partitions()
# Delete the reverification block, then update the partitions
self.store.delete_item(
self.icrv.location,
ModuleStoreEnum.UserID.test,
revision=ModuleStoreEnum.RevisionOption.published_only
)
self._update_partitions()
# Check that the user partition was marked as inactive
self.assertEqual(len(self.course.user_partitions), 1)
partition = self.course.user_partitions[0]
self.assertFalse(partition.active)
self.assertEqual(partition.scheme.name, "verification")
@patch.dict(settings.FEATURES, {"ENABLE_COURSEWARE_INDEX": False})
def test_preserves_partition_id_for_verified_partitions(self):
self._update_partitions()
partition_id = self.course.user_partitions[0].id
self._update_partitions()
new_partition_id = self.course.user_partitions[0].id
self.assertEqual(partition_id, new_partition_id)
@patch.dict(settings.FEATURES, {"ENABLE_COURSEWARE_INDEX": False})
def test_preserves_existing_user_partitions(self):
# Add other, non-verified partition to the course
self.course.user_partitions = [
UserPartition(
id=0,
name='Cohort user partition',
scheme=UserPartition.get_scheme('cohort'),
description='Cohorted user partition',
groups=[
Group(id=0, name="Group A"),
Group(id=1, name="Group B"),
],
),
UserPartition(
id=1,
name='Random user partition',
scheme=UserPartition.get_scheme('random'),
description='Random user partition',
groups=[
Group(id=0, name="Group A"),
Group(id=1, name="Group B"),
],
),
]
self.course = self.store.update_item(self.course, ModuleStoreEnum.UserID.test)
# Update the verification partitions.
# The existing partitions should still be available
self._update_partitions()
partition_ids = [p.id for p in self.course.user_partitions]
self.assertEqual(len(partition_ids), 3)
self.assertIn(0, partition_ids)
self.assertIn(1, partition_ids)
def test_multiple_reverification_blocks(self):
# Add an additional ICRV block in another section
other_icrv = ItemFactory.create(parent=self.verticals[3], category='edx-reverification-block')
self._update_partitions()
# Expect that both ICRV blocks have corresponding partitions
self.assertEqual(len(self.course.user_partitions), 2)
partition_locations = [p.parameters.get("location") for p in self.course.user_partitions]
self.assertIn(unicode(self.icrv.location), partition_locations)
self.assertIn(unicode(other_icrv.location), partition_locations)
# Delete the first ICRV block and update partitions
icrv_location = self.icrv.location
self.store.delete_item(
self.icrv.location,
ModuleStoreEnum.UserID.test,
revision=ModuleStoreEnum.RevisionOption.published_only
)
self._update_partitions()
# Expect that the correct partition is marked as inactive
self.assertEqual(len(self.course.user_partitions), 2)
partitions_by_loc = {
p.parameters["location"]: p
for p in self.course.user_partitions
}
self.assertFalse(partitions_by_loc[unicode(icrv_location)].active)
self.assertTrue(partitions_by_loc[unicode(other_icrv.location)].active)
def test_query_counts_with_no_reverification_blocks(self):
# Delete the ICRV block, so the number of ICRV blocks is zero
self.store.delete_item(
self.icrv.location,
ModuleStoreEnum.UserID.test,
revision=ModuleStoreEnum.RevisionOption.published_only
)
# 2 calls: get the course (definitions + structures)
# 2 calls: look up ICRV blocks in the course (definitions + structures)
with check_mongo_calls_range(max_finds=4, max_sends=2):
self._update_partitions(reload_items=False)
def test_query_counts_with_one_reverification_block(self):
# One ICRV block created in the setup method
# Additional call to load the ICRV block
with check_mongo_calls_range(max_finds=5, max_sends=3):
self._update_partitions(reload_items=False)
def test_query_counts_with_multiple_reverification_blocks(self):
# Total of two ICRV blocks (one created in setup method)
# Additional call to load each ICRV block
ItemFactory.create(parent=self.verticals[3], category='edx-reverification-block')
with check_mongo_calls_range(max_finds=6, max_sends=3):
self._update_partitions(reload_items=False)
def _update_partitions(self, reload_items=True):
"""Update user partitions in the course descriptor, then reload the content. """
update_verification_partitions(self.course.id) # pylint: disable=no-member
# Reload each component so we can see the changes
if reload_items:
self.course = self.store.get_course(self.course.id) # pylint: disable=no-member
self.sections = [self._reload_item(section.location) for section in self.sections]
self.subsections = [self._reload_item(subsection.location) for subsection in self.subsections]
self.verticals = [self._reload_item(vertical.location) for vertical in self.verticals]
self.icrv = self._reload_item(self.icrv.location)
self.sibling_problem = self._reload_item(self.sibling_problem.location)
def _reload_item(self, location):
"""Safely reload an item from the moduelstore. """
try:
return self.store.get_item(location)
except ItemNotFoundError:
return None
@attr(shard=2)
class WriteOnPublishTest(ModuleStoreTestCase):
"""
Verify that updates to the course descriptor's
user partitions are written automatically on publish.
"""
MODULESTORE = TEST_DATA_SPLIT_MODULESTORE
ENABLED_SIGNALS = ['course_published', 'pre_publish']
@patch.dict(settings.FEATURES, {"ENABLE_COURSEWARE_INDEX": False})
def setUp(self):
super(WriteOnPublishTest, self).setUp()
# Create a dummy course with an ICRV block
self.course = CourseFactory()
self.section = ItemFactory.create(parent=self.course, category='chapter', display_name='Test Section')
self.subsection = ItemFactory.create(parent=self.section, category='sequential', display_name='Test Subsection')
self.vertical = ItemFactory.create(parent=self.subsection, category='vertical', display_name='Test Unit')
self.icrv = ItemFactory.create(parent=self.vertical, category='edx-reverification-block')
# Mark the course as credit
CreditCourse.objects.create(course_key=self.course.id, enabled=True) # pylint: disable=no-member
@patch.dict(settings.FEATURES, {"ENABLE_COURSEWARE_INDEX": False})
def test_can_write_on_publish_signal(self):
# Sanity check -- initially user partitions should be empty
self.assertEqual(self.course.user_partitions, [])
# Make and publish a change to a block, which should trigger the publish signal
with self.store.bulk_operations(self.course.id): # pylint: disable=no-member
self.icrv.display_name = "Updated display name"
self.store.update_item(self.icrv, ModuleStoreEnum.UserID.test)
self.store.publish(self.icrv.location, ModuleStoreEnum.UserID.test)
# Within the test, the course pre-publish signal should have fired synchronously
# Since the course is marked as credit, the in-course verification partitions
# should have been created.
# We need to verify that these changes were actually persisted to the modulestore.
retrieved_course = self.store.get_course(self.course.id) # pylint: disable=no-member
self.assertEqual(len(retrieved_course.user_partitions), 1)

187
openedx/core/djangoapps/credit/verification_access.py
View File

@@ -1,187 +0,0 @@
"""
Create in-course reverification access groups in a course.
We model the rules as a set of user partitions, one for each
verification checkpoint in a course.
For example, suppose that a course has two verification checkpoints,
one at midterm A and one at the midterm B.
Then the user partitions would look like this:
Midterm A: |-- ALLOW --|-- DENY --|
Midterm B: |-- ALLOW --|-- DENY --|
where the groups are defined as:
* ALLOW: The user has access to content gated by the checkpoint.
* DENY: The user does not have access to content gated by the checkpoint.
"""
import logging
from util.db import generate_int_id
from openedx.core.djangoapps.credit.utils import get_course_blocks
from xmodule.modulestore.django import modulestore
from xmodule.modulestore import ModuleStoreEnum
from xmodule.partitions.partitions import Group, UserPartition
log = logging.getLogger(__name__)
VERIFICATION_SCHEME_NAME = "verification"
VERIFICATION_BLOCK_CATEGORY = "edx-reverification-block"
def update_verification_partitions(course_key):
"""
Create a user partition for each verification checkpoint in the course.
This will modify the published version of the course descriptor.
It ensures that any in-course reverification XBlocks in the course
have an associated user partition. Other user partitions (e.g. cohorts)
will be preserved. Partitions associated with deleted reverification checkpoints
will be marked as inactive and will not be used to restrict access.
Arguments:
course_key (CourseKey): identifier for the course.
Returns:
None
"""
# Batch all the queries we're about to do and suppress
# the "publish" signal to avoid an infinite call loop.
with modulestore().bulk_operations(course_key, emit_signals=False):
# Retrieve all in-course reverification blocks in the course
icrv_blocks = get_course_blocks(course_key, VERIFICATION_BLOCK_CATEGORY)
# Update the verification definitions in the course descriptor
# This will also clean out old verification partitions if checkpoints
# have been deleted.
_set_verification_partitions(course_key, icrv_blocks)
def _unique_partition_id(course):
"""Return a unique user partition ID for the course. """
# Exclude all previously used IDs, even for partitions that have been disabled
# (e.g. if the course author deleted an in-course reverifification block but
# there are courseware components that reference the disabled partition).
used_ids = set(p.id for p in course.user_partitions)
return generate_int_id(used_ids=used_ids)
def _other_partitions(verified_partitions, exclude_partitions, course_key):
"""
Retrieve all partitions NOT associated with the current set of ICRV blocks.
Any partition associated with a deleted ICRV block will be marked as inactive
so its access rules will no longer be enforced.
Arguments:
all_partitions (list of UserPartition): All verified partitions defined in the course.
exclude_partitions (list of UserPartition): Partitions to exclude (e.g. the ICRV partitions already added)
course_key (CourseKey): Identifier for the course (used for logging).
Returns: list of `UserPartition`s
"""
results = []
partition_by_id = {
p.id: p for p in verified_partitions
}
other_partition_ids = set(p.id for p in verified_partitions) - set(p.id for p in exclude_partitions)
for pid in other_partition_ids:
partition = partition_by_id[pid]
results.append(
UserPartition(
id=partition.id,
name=partition.name,
description=partition.description,
scheme=partition.scheme,
parameters=partition.parameters,
groups=partition.groups,
active=False,
)
)
log.info(
(
"Disabled partition %s in course %s because the "
"associated in-course-reverification checkpoint does not exist."
),
partition.id, course_key
)
return results
def _set_verification_partitions(course_key, icrv_blocks):
"""
Create or update user partitions in the course.
Ensures that each ICRV block in the course has an associated user partition
with the groups ALLOW and DENY.
Arguments:
course_key (CourseKey): Identifier for the course.
icrv_blocks (list of XBlock): In-course reverification blocks, e.g. reverification checkpoints.
Returns:
list of UserPartition
"""
scheme = UserPartition.get_scheme(VERIFICATION_SCHEME_NAME)
if scheme is None:
log.error("Could not retrieve user partition scheme with ID %s", VERIFICATION_SCHEME_NAME)
return []
course = modulestore().get_course(course_key)
if course is None:
log.error("Could not find course %s", course_key)
return []
verified_partitions = course.get_user_partitions_for_scheme(scheme)
partition_id_for_location = {
p.parameters["location"]: p.id
for p in verified_partitions
if "location" in p.parameters
}
partitions = []
for block in icrv_blocks:
partition = UserPartition(
id=partition_id_for_location.get(
unicode(block.location),
_unique_partition_id(course)
),
name=block.related_assessment,
description=u"Verification checkpoint at {}".format(block.related_assessment),
scheme=scheme,
parameters={"location": unicode(block.location)},
groups=[
Group(scheme.ALLOW, "Completed verification at {}".format(block.related_assessment)),
Group(scheme.DENY, "Did not complete verification at {}".format(block.related_assessment)),
]
)
partitions.append(partition)
log.info(
(
"Configured partition %s for course %s using a verified partition scheme "
"for the in-course-reverification checkpoint at location %s"
),
partition.id,
course_key,
partition.parameters["location"]
)
# Preserve existing, non-verified partitions from the course
# Mark partitions for deleted in-course reverification as disabled.
partitions += _other_partitions(verified_partitions, partitions, course_key)
course.set_user_partitions_for_scheme(partitions, scheme)
modulestore().update_item(course, ModuleStoreEnum.UserID.system)
log.info("Saved updated partitions for the course %s", course_key)
return partitions

14
openedx/core/djangolib/model_mixins.py Normal file
View File

@@ -0,0 +1,14 @@
"""
Custom Django Model mixins.
"""
class DeprecatedModelMixin(object):
"""
Used to make a class unusable in practice, but leave database tables intact.
"""
def __init__(self, *args, **kwargs): # pylint: disable=unused-argument
"""
Override to kill usage of this model.
"""
raise TypeError("This model has been deprecated and should not be used.")

1
requirements/edx/github.txt
View File

@@ -86,7 +86,6 @@ git+https://github.com/solashirai/crowdsourcehinter.git@518605f0a95190949fe77bd3
git+https://github.com/edx/edx-milestones.git@v0.1.10#egg=edx-milestones==0.1.10
git+https://github.com/edx/xblock-utils.git@v1.0.3#egg=xblock-utils==1.0.3
-e git+https://github.com/edx-solutions/xblock-google-drive.git@138e6fa0bf3a2013e904a085b9fed77dab7f3f21#egg=xblock-google-drive
-e git+https://github.com/edx/edx-reverification-block.git@0.0.5#egg=edx-reverification-block==0.0.5
git+https://github.com/edx/edx-user-state-client.git@1.0.1#egg=edx-user-state-client==1.0.1
git+https://github.com/edx/xblock-lti-consumer.git@v1.1.2#egg=lti_consumer-xblock==1.1.2
git+https://github.com/edx/edx-proctoring.git@0.18.0#egg=edx-proctoring==0.18.0

1
setup.py
View File

@@ -41,7 +41,6 @@ setup(
"openedx.user_partition_scheme": [
"random = openedx.core.djangoapps.user_api.partition_schemes:RandomUserPartitionScheme",
"cohort = openedx.core.djangoapps.course_groups.partition_scheme:CohortPartitionScheme",
"verification = openedx.core.djangoapps.credit.partition_schemes:VerificationPartitionScheme",
],
"openedx.block_structure_transformer": [
"library_content = lms.djangoapps.course_blocks.transformers.library_content:ContentLibraryTransformer",