verification_access.py 6.59 KB
Newer Older
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187
"""
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