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

Merge pull request #15170 from edx/jeskew/PLAT_1316_partitions_inheritance

Browse Source 
Move fields.py, inheritance.py, and partitions to openedx/core.
This commit is contained in:
John Eskew
2017-05-25 14:21:03 -04:00
committed by GitHub
parent 800bcd8e20 76389d8678
commit f97f052591
94 changed files with 1098 additions and 1126 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
openedx/core/djangoapps/content/course_overviews/models.py
View File

@@ -20,10 +20,11 @@ from lms.djangoapps import django_comment_client
from openedx.core.djangoapps.models.course_details import CourseDetails
from static_replace.models import AssetBaseUrlConfig
from xmodule import course_metadata_utils, block_metadata_utils
from xmodule.course_module import CourseDescriptor, DEFAULT_START_DATE
from xmodule.course_module import CourseDescriptor
from xmodule.error_module import ErrorDescriptor
from xmodule.modulestore.django import modulestore
from openedx.core.djangoapps.xmodule_django.models import CourseKeyField, UsageKeyField
from openedx.core.lib.xblock_fields.inherited_fields import DEFAULT_START_DATE
log = logging.getLogger(__name__)

2
openedx/core/djangoapps/content/course_overviews/tests.py
View File

@@ -19,11 +19,11 @@ from PIL import Image
from lms.djangoapps.certificates.api import get_active_web_certificate
from openedx.core.djangoapps.models.course_details import CourseDetails
from openedx.core.lib.courses import course_image_url
from openedx.core.lib.xblock_fields.inherited_fields import DEFAULT_START_DATE
from static_replace.models import AssetBaseUrlConfig
from xmodule.assetstore.assetmgr import AssetManager
from xmodule.contentstore.django import contentstore
from xmodule.contentstore.content import StaticContent
from xmodule.course_metadata_utils import DEFAULT_START_DATE
from xmodule.course_module import (
CATALOG_VISIBILITY_CATALOG_AND_ABOUT,
CATALOG_VISIBILITY_ABOUT,

3
openedx/core/djangoapps/course_groups/partition_scheme.py
View File

@@ -8,11 +8,10 @@ from courseware.masquerade import ( # pylint: disable=import-error
get_masquerading_user_group,
is_masquerading_as_specific_student
)
from xmodule.partitions.partitions import NoSuchUserPartitionGroupError
from openedx.core.lib.partitions.partitions import NoSuchUserPartitionGroupError
from .cohorts import get_cohort, get_group_info_for_cohort
log = logging.getLogger(__name__)

19
openedx/core/djangoapps/course_groups/tests/test_partition_scheme.py
View File

@@ -3,22 +3,21 @@ Test the partitions and partitions service
"""
import django.test
from courseware.tests.test_masquerade import StaffMasqueradeTestCase
from mock import patch
from nose.plugins.attrib import attr
from courseware.tests.test_masquerade import StaffMasqueradeTestCase
from student.tests.factories import UserFactory
from xmodule.partitions.partitions import Group, UserPartition, UserPartitionError
from xmodule.modulestore.django import modulestore
from xmodule.modulestore.tests.django_utils import ModuleStoreTestCase, TEST_DATA_MIXED_MODULESTORE
from xmodule.modulestore.tests.factories import ToyCourseFactory
from openedx.core.djangoapps.user_api.partition_schemes import RandomUserPartitionScheme
from openedx.core.djangolib.testing.utils import skip_unless_lms
from ..partition_scheme import CohortPartitionScheme, get_cohorted_user_partition
from openedx.core.lib.partitions.partitions import Group, UserPartition, UserPartitionError
from student.tests.factories import UserFactory
from xmodule.modulestore.django import modulestore
from xmodule.modulestore.tests.django_utils import TEST_DATA_MIXED_MODULESTORE, ModuleStoreTestCase
from xmodule.modulestore.tests.factories import ToyCourseFactory
from ..cohorts import add_user_to_cohort, get_course_cohorts, remove_user_from_cohort
from ..models import CourseUserGroupPartitionGroup
from ..partition_scheme import CohortPartitionScheme, get_cohorted_user_partition
from ..views import link_cohort_to_partition_group, unlink_cohort_partition_group
from ..cohorts import add_user_to_cohort, remove_user_from_cohort, get_course_cohorts
from .helpers import CohortFactory, config_course_cohorts

8
openedx/core/djangoapps/models/course_details.py
View File

@@ -1,17 +1,15 @@
"""
CourseDetails
"""
import re
import logging
import re
from django.conf import settings
from xmodule.fields import Date
from xmodule.modulestore.exceptions import ItemNotFoundError
from openedx.core.djangoapps.self_paced.models import SelfPacedConfiguration
from openedx.core.lib.courses import course_image_url
from openedx.core.lib.xblock_fields.fields import Date
from xmodule.modulestore.django import modulestore
from xmodule.modulestore.exceptions import ItemNotFoundError
# This list represents the attribute keys for a course's 'about' info.
# Note: The 'video' attribute is intentionally excluded as it must be

4
openedx/core/djangoapps/user_api/partition_schemes.py
View File

@@ -3,9 +3,9 @@ Provides partition support to the user service.
"""
import logging
import random
import course_tag.api as course_tag_api
from xmodule.partitions.partitions import UserPartitionError, NoSuchUserPartitionGroupError
import course_tag.api as course_tag_api
from openedx.core.lib.partitions.partitions import NoSuchUserPartitionGroupError, UserPartitionError
log = logging.getLogger(__name__)

6
openedx/core/djangoapps/user_api/tests/test_partition_schemes.py
View File

@@ -2,13 +2,13 @@
Test the user api's partition extensions.
"""
from collections import defaultdict
from mock import patch
from unittest import TestCase
from mock import patch
from openedx.core.djangoapps.user_api.partition_schemes import RandomUserPartitionScheme, UserPartitionError
from openedx.core.lib.partitions.partitions import Group, UserPartition
from openedx.core.lib.partitions.tests.test_partitions import PartitionTestCase
from student.tests.factories import UserFactory
from xmodule.partitions.partitions import Group, UserPartition
from xmodule.partitions.tests.test_partitions import PartitionTestCase
class MemoryCourseTagAPI(object):

8
openedx/core/djangoapps/util/testing.py
View File

@@ -1,15 +1,15 @@
""" Mixins for setting up particular course structures (such as split tests or cohorted content) """
from datetime import datetime
from pytz import UTC
from openedx.core.djangoapps.course_groups.models import CourseUserGroupPartitionGroup
from openedx.core.djangoapps.course_groups.tests.helpers import CohortFactory
from openedx.core.djangoapps.user_api.tests.factories import UserCourseTagFactory
from xmodule.modulestore.tests.factories import CourseFactory, ItemFactory
from xmodule.modulestore.tests.django_utils import ModuleStoreTestCase
from xmodule.partitions.partitions import UserPartition, Group
from openedx.core.lib.partitions.partitions import Group, UserPartition
from pytz import UTC
from student.tests.factories import CourseEnrollmentFactory, UserFactory
from xmodule.modulestore.tests.django_utils import ModuleStoreTestCase
from xmodule.modulestore.tests.factories import CourseFactory, ItemFactory
class ContentGroupTestCase(ModuleStoreTestCase):

10
openedx/core/djangoapps/verified_track_content/partition_scheme.py
View File

@@ -1,19 +1,17 @@
"""
UserPartitionScheme for enrollment tracks.
"""
from django.conf import settings
from course_modes.models import CourseMode
from courseware.masquerade import (
get_course_masquerade,
get_masquerading_user_group,
is_masquerading_as_specific_student
)
from course_modes.models import CourseMode
from student.models import CourseEnrollment
from django.conf import settings
from opaque_keys.edx.keys import CourseKey
from openedx.core.djangoapps.verified_track_content.models import VerifiedTrackCohortedCourse
from xmodule.partitions.partitions import NoSuchUserPartitionGroupError, Group, UserPartition
from openedx.core.lib.partitions.partitions import Group, NoSuchUserPartitionGroupError, UserPartition
from student.models import CourseEnrollment
# These IDs must be less than 100 so that they do not overlap with Groups in
# CohortUserPartition or RandomUserPartitionScheme

10
openedx/core/djangoapps/verified_track_content/tests/test_partition_scheme.py
View File

@@ -2,17 +2,17 @@
Tests for verified_track_content/partition_scheme.py.
"""
from datetime import datetime, timedelta
import pytz
from ..partition_scheme import EnrollmentTrackPartitionScheme, EnrollmentTrackUserPartition, ENROLLMENT_GROUP_IDS
from ..models import VerifiedTrackCohortedCourse
from course_modes.models import CourseMode
from openedx.core.lib.partitions.partitions import MINIMUM_STATIC_PARTITION_ID, UserPartition
from student.models import CourseEnrollment
from student.tests.factories import UserFactory
from xmodule.modulestore.tests.django_utils import SharedModuleStoreTestCase
from xmodule.modulestore.tests.factories import CourseFactory
from xmodule.partitions.partitions import UserPartition, MINIMUM_STATIC_PARTITION_ID
from ..models import VerifiedTrackCohortedCourse
from ..partition_scheme import ENROLLMENT_GROUP_IDS, EnrollmentTrackPartitionScheme, EnrollmentTrackUserPartition
class EnrollmentTrackUserPartitionTest(SharedModuleStoreTestCase):

0
openedx/core/lib/partitions/__init__.py Normal file
View File

240
openedx/core/lib/partitions/partitions.py Normal file
View File

@@ -0,0 +1,240 @@
"""Defines ``Group`` and ``UserPartition`` models for partitioning"""
from collections import namedtuple
from stevedore.extension import ExtensionManager
# We use ``id`` in this file as the IDs of our Groups and UserPartitions,
# which Pylint disapproves of.
# pylint: disable=redefined-builtin
# UserPartition IDs must be unique. The Cohort and Random UserPartitions (when they are
# created via Studio) choose an unused ID in the range of 100 (historical) to MAX_INT. Therefore the
# dynamic UserPartitionIDs must be under 100, and they have to be hard-coded to ensure
# they are always the same whenever the dynamic partition is added (since the UserPartition
# ID is stored in the xblock group_access dict).
ENROLLMENT_TRACK_PARTITION_ID = 50
MINIMUM_STATIC_PARTITION_ID = 100
class UserPartitionError(Exception):
"""
Base Exception for when an error was found regarding user partitions.
"""
pass
class NoSuchUserPartitionError(UserPartitionError):
"""
Exception to be raised when looking up a UserPartition by its ID fails.
"""
pass
class NoSuchUserPartitionGroupError(UserPartitionError):
"""
Exception to be raised when looking up a UserPartition Group by its ID fails.
"""
pass
class Group(namedtuple("Group", "id name")):
"""
An id and name for a group of students. The id should be unique
within the UserPartition this group appears in.
"""
# in case we want to add to this class, a version will be handy
# for deserializing old versions. (This will be serialized in courses)
VERSION = 1
def __new__(cls, id, name):
return super(Group, cls).__new__(cls, int(id), name)
def to_json(self):
"""
'Serialize' to a json-serializable representation.
Returns:
a dictionary with keys for the properties of the group.
"""
return {
"id": self.id,
"name": self.name,
"version": Group.VERSION
}
@staticmethod
def from_json(value):
"""
Deserialize a Group from a json-like representation.
Args:
value: a dictionary with keys for the properties of the group.
Raises TypeError if the value doesn't have the right keys.
"""
if isinstance(value, Group):
return value
for key in ("id", "name", "version"):
if key not in value:
raise TypeError("Group dict {0} missing value key '{1}'".format(
value, key))
if value["version"] != Group.VERSION:
raise TypeError("Group dict {0} has unexpected version".format(
value))
return Group(value["id"], value["name"])
# The Stevedore extension point namespace for user partition scheme plugins.
USER_PARTITION_SCHEME_NAMESPACE = 'openedx.user_partition_scheme'
class UserPartition(namedtuple("UserPartition", "id name description groups scheme parameters active")):
"""A named way to partition users into groups, primarily intended for
running experiments. It is expected that each user will be in at most one
group in a partition.
A Partition has an id, name, scheme, description, parameters, and a list
of groups. The id is intended to be unique within the context where these
are used. (e.g., for partitions of users within a course, the ids should
be unique per-course). The scheme is used to assign users into groups.
The parameters field is used to save extra parameters e.g., location of
the block in case of VerificationPartitionScheme.
Partitions can be marked as inactive by setting the "active" flag to False.
Any group access rule referencing inactive partitions will be ignored
when performing access checks.
"""
VERSION = 3
# The collection of user partition scheme extensions.
scheme_extensions = None
# The default scheme to be used when upgrading version 1 partitions.
VERSION_1_SCHEME = "random"
def __new__(cls, id, name, description, groups, scheme=None, parameters=None, active=True, scheme_id=VERSION_1_SCHEME): # pylint: disable=line-too-long
if not scheme:
scheme = UserPartition.get_scheme(scheme_id)
if parameters is None:
parameters = {}
return super(UserPartition, cls).__new__(cls, int(id), name, description, groups, scheme, parameters, active)
@staticmethod
def get_scheme(name):
"""
Returns the user partition scheme with the given name.
"""
# Note: we're creating the extension manager lazily to ensure that the Python path
# has been correctly set up. Trying to create this statically will fail, unfortunately.
if not UserPartition.scheme_extensions:
UserPartition.scheme_extensions = ExtensionManager(namespace=USER_PARTITION_SCHEME_NAMESPACE)
try:
scheme = UserPartition.scheme_extensions[name].plugin
except KeyError:
raise UserPartitionError("Unrecognized scheme '{0}'".format(name))
scheme.name = name
return scheme
def to_json(self):
"""
'Serialize' to a json-serializable representation.
Returns:
a dictionary with keys for the properties of the partition.
"""
return {
"id": self.id,
"name": self.name,
"scheme": self.scheme.name,
"description": self.description,
"parameters": self.parameters,
"groups": [g.to_json() for g in self.groups],
"active": bool(self.active),
"version": UserPartition.VERSION
}
@staticmethod
def from_json(value):
"""
Deserialize a Group from a json-like representation.
Args:
value: a dictionary with keys for the properties of the group.
Raises TypeError if the value doesn't have the right keys.
"""
if isinstance(value, UserPartition):
return value
for key in ("id", "name", "description", "version", "groups"):
if key not in value:
raise TypeError("UserPartition dict {0} missing value key '{1}'".format(value, key))
if value["version"] == 1:
# If no scheme was provided, set it to the default ('random')
scheme_id = UserPartition.VERSION_1_SCHEME
# Version changes should be backwards compatible in case the code
# gets rolled back. If we see a version number greater than the current
# version, we should try to read it rather than raising an exception.
elif value["version"] >= 2:
if "scheme" not in value:
raise TypeError("UserPartition dict {0} missing value key 'scheme'".format(value))
scheme_id = value["scheme"]
else:
raise TypeError("UserPartition dict {0} has unexpected version".format(value))
parameters = value.get("parameters", {})
active = value.get("active", True)
groups = [Group.from_json(g) for g in value["groups"]]
scheme = UserPartition.get_scheme(scheme_id)
if not scheme:
raise TypeError("UserPartition dict {0} has unrecognized scheme {1}".format(value, scheme_id))
if hasattr(scheme, "create_user_partition"):
return scheme.create_user_partition(
value["id"],
value["name"],
value["description"],
groups,
parameters,
active,
)
else:
return UserPartition(
value["id"],
value["name"],
value["description"],
groups,
scheme,
parameters,
active,
)
def get_group(self, group_id):
"""
Returns the group with the specified id.
Arguments:
group_id (int): ID of the partition group.
Raises:
NoSuchUserPartitionGroupError: The specified group could not be found.
"""
for group in self.groups:
if group.id == group_id:
return group
raise NoSuchUserPartitionGroupError(
"Could not find a Group with ID [{group_id}] in UserPartition [{partition_id}].".format(
group_id=group_id, partition_id=self.id
)
)

183
openedx/core/lib/partitions/partitions_service.py Normal file
View File

@@ -0,0 +1,183 @@
"""
This is a service-like API that assigns tracks which groups users are in for various
user partitions. It uses the user_service key/value store provided by the LMS runtime to
persist the assignments.
"""
from django.conf import settings
from django.utils.translation import ugettext_lazy as _
import logging
from .partitions import UserPartition, UserPartitionError, ENROLLMENT_TRACK_PARTITION_ID
from xmodule.modulestore.django import modulestore
log = logging.getLogger(__name__)
# settings will not be available when running nosetests.
FEATURES = getattr(settings, 'FEATURES', {})
def get_all_partitions_for_course(course, active_only=False):
"""
A method that returns all `UserPartitions` associated with a course, as a List.
This will include the ones defined in course.user_partitions, but it may also
include dynamically included partitions (such as the `EnrollmentTrackUserPartition`).
Args:
course: the course for which user partitions should be returned.
active_only: if `True`, only partitions with `active` set to True will be returned.
Returns:
A List of UserPartitions associated with the course.
"""
all_partitions = course.user_partitions + _get_dynamic_partitions(course)
if active_only:
all_partitions = [partition for partition in all_partitions if partition.active]
return all_partitions
def _get_dynamic_partitions(course):
"""
Return the dynamic user partitions for this course.
If none exists, returns an empty array.
"""
enrollment_partition = _create_enrollment_track_partition(course)
return [enrollment_partition] if enrollment_partition else []
def _create_enrollment_track_partition(course):
"""
Create and return the dynamic enrollment track user partition.
If it cannot be created, None is returned.
"""
if not FEATURES.get('ENABLE_ENROLLMENT_TRACK_USER_PARTITION'):
return None
try:
enrollment_track_scheme = UserPartition.get_scheme("enrollment_track")
except UserPartitionError:
log.warning("No 'enrollment_track' scheme registered, EnrollmentTrackUserPartition will not be created.")
return None
used_ids = set(p.id for p in course.user_partitions)
if ENROLLMENT_TRACK_PARTITION_ID in used_ids:
# TODO: change to Exception after this has been in production for awhile, see TNL-6796.
log.warning(
"Can't add 'enrollment_track' partition, as ID {id} is assigned to {partition} in course {course}.".format(
id=ENROLLMENT_TRACK_PARTITION_ID,
partition=_get_partition_from_id(course.user_partitions, ENROLLMENT_TRACK_PARTITION_ID).name,
course=unicode(course.id)
)
)
return None
partition = enrollment_track_scheme.create_user_partition(
id=ENROLLMENT_TRACK_PARTITION_ID,
name=_(u"Enrollment Track Groups"),
description=_(u"Partition for segmenting users by enrollment track"),
parameters={"course_id": unicode(course.id)}
)
return partition
class PartitionService(object):
"""
This is an XBlock service that returns information about the user partitions associated
with a given course.
"""
def __init__(self, course_id, track_function=None, cache=None):
self._course_id = course_id
self._track_function = track_function
self._cache = cache
def get_course(self):
"""
Return the course instance associated with this PartitionService.
This default implementation looks up the course from the modulestore.
"""
return modulestore().get_course(self._course_id)
@property
def course_partitions(self):
"""
Return the set of partitions assigned to self._course_id (both those set directly on the course
through course.user_partitions, and any dynamic partitions that exist). Note: this returns
both active and inactive partitions.
"""
return get_all_partitions_for_course(self.get_course())
def get_user_group_id_for_partition(self, user, user_partition_id):
"""
If the user is already assigned to a group in user_partition_id, return the
group_id.
If not, assign them to one of the groups, persist that decision, and
return the group_id.
Args:
user_partition_id -- an id of a partition that's hopefully in the
runtime.user_partitions list.
Returns:
The id of one of the groups in the specified user_partition_id (as a string).
Raises:
ValueError if the user_partition_id isn't found.
"""
cache_key = "PartitionService.ugidfp.{}.{}.{}".format(
user.id, self._course_id, user_partition_id
)
if self._cache and (cache_key in self._cache):
return self._cache[cache_key]
user_partition = self._get_user_partition(user_partition_id)
if user_partition is None:
raise ValueError(
"Configuration problem! No user_partition with id {0} "
"in course {1}".format(user_partition_id, self._course_id)
)
group = self.get_group(user, user_partition)
group_id = group.id if group else None
if self._cache is not None:
self._cache[cache_key] = group_id
return group_id
def _get_user_partition(self, user_partition_id):
"""
Look for a user partition with a matching id in the course's partitions.
Note that this method can return an inactive user partition.
Returns:
A UserPartition, or None if not found.
"""
return _get_partition_from_id(self.course_partitions, user_partition_id)
def get_group(self, user, user_partition, assign=True):
"""
Returns the group from the specified user partition to which the user is assigned.
If the user has not yet been assigned, a group will be chosen for them based upon
the partition's scheme.
"""
return user_partition.scheme.get_group_for_user(
self._course_id, user, user_partition, assign=assign, track_function=self._track_function
)
def _get_partition_from_id(partitions, user_partition_id):
"""
Look for a user partition with a matching id in the provided list of partitions.
Returns:
A UserPartition, or None if not found.
"""
for partition in partitions:
if partition.id == user_partition_id:
return partition
return None

0
openedx/core/lib/partitions/tests/__init__.py Normal file
View File

620
openedx/core/lib/partitions/tests/test_partitions.py Normal file
View File

@@ -0,0 +1,620 @@
"""
Test the partitions and partitions service
"""
from unittest import TestCase
from mock import Mock
from opaque_keys.edx.locator import CourseLocator
from stevedore.extension import Extension, ExtensionManager
from openedx.core.lib.partitions.partitions import (
Group, UserPartition, UserPartitionError, NoSuchUserPartitionGroupError,
USER_PARTITION_SCHEME_NAMESPACE, ENROLLMENT_TRACK_PARTITION_ID
)
from openedx.core.lib.partitions.partitions_service import (
PartitionService, get_all_partitions_for_course, FEATURES
)
class TestGroup(TestCase):
"""Test constructing groups"""
def test_construct(self):
test_id = 10
name = "Grendel"
group = Group(test_id, name)
self.assertEqual(group.id, test_id)
self.assertEqual(group.name, name)
def test_string_id(self):
test_id = "10"
name = "Grendel"
group = Group(test_id, name)
self.assertEqual(group.id, 10)
def test_to_json(self):
test_id = 10
name = "Grendel"
group = Group(test_id, name)
jsonified = group.to_json()
act_jsonified = {
"id": test_id,
"name": name,
"version": group.VERSION
}
self.assertEqual(jsonified, act_jsonified)
def test_from_json(self):
test_id = 5
name = "Grendel"
jsonified = {
"id": test_id,
"name": name,
"version": Group.VERSION
}
group = Group.from_json(jsonified)
self.assertEqual(group.id, test_id)
self.assertEqual(group.name, name)
def test_from_json_broken(self):
test_id = 5
name = "Grendel"
# Bad version
jsonified = {
"id": test_id,
"name": name,
"version": -1,
}
with self.assertRaisesRegexp(TypeError, "has unexpected version"):
Group.from_json(jsonified)
# Missing key "id"
jsonified = {
"name": name,
"version": Group.VERSION
}
with self.assertRaisesRegexp(TypeError, "missing value key 'id'"):
Group.from_json(jsonified)
# Has extra key - should not be a problem
jsonified = {
"id": test_id,
"name": name,
"version": Group.VERSION,
"programmer": "Cale"
}
group = Group.from_json(jsonified)
self.assertNotIn("programmer", group.to_json())
class MockUserPartitionScheme(object):
"""
Mock user partition scheme
"""
def __init__(self, name="mock", current_group=None, **kwargs):
super(MockUserPartitionScheme, self).__init__(**kwargs)
self.name = name
self.current_group = current_group
def get_group_for_user(self, course_id, user, user_partition, assign=True, track_function=None): # pylint: disable=unused-argument
"""
Returns the current group if set, else the first group from the specified user partition.
"""
if self.current_group:
return self.current_group
groups = user_partition.groups
if not groups or len(groups) == 0:
return None
return groups[0]
class MockEnrollmentTrackUserPartitionScheme(MockUserPartitionScheme):
def create_user_partition(self, id, name, description, groups=None, parameters=None, active=True): # pylint: disable=redefined-builtin, invalid-name, unused-argument
"""
The EnrollmentTrackPartitionScheme provides this method to return a subclass of UserPartition.
"""
return UserPartition(id, name, description, groups, self, parameters, active)
class PartitionTestCase(TestCase):
"""Base class for test cases that require partitions"""
TEST_ID = 0
TEST_NAME = "Mock Partition"
TEST_DESCRIPTION = "for testing purposes"
TEST_PARAMETERS = {"location": "block-v1:edX+DemoX+Demo+type@block@uuid"}
TEST_GROUPS = [Group(0, 'Group 1'), Group(1, 'Group 2')]
TEST_SCHEME_NAME = "mock"
ENROLLMENT_TRACK_SCHEME_NAME = "enrollment_track"
def setUp(self):
super(PartitionTestCase, self).setUp()
# Set up two user partition schemes: mock and random
self.non_random_scheme = MockUserPartitionScheme(self.TEST_SCHEME_NAME)
self.random_scheme = MockUserPartitionScheme("random")
self.enrollment_track_scheme = MockEnrollmentTrackUserPartitionScheme(self.ENROLLMENT_TRACK_SCHEME_NAME)
extensions = [
Extension(
self.non_random_scheme.name, USER_PARTITION_SCHEME_NAMESPACE, self.non_random_scheme, None
),
Extension(
self.random_scheme.name, USER_PARTITION_SCHEME_NAMESPACE, self.random_scheme, None
),
Extension(
self.enrollment_track_scheme.name, USER_PARTITION_SCHEME_NAMESPACE, self.enrollment_track_scheme, None
),
]
UserPartition.scheme_extensions = ExtensionManager.make_test_instance(
extensions, namespace=USER_PARTITION_SCHEME_NAMESPACE
)
# Be sure to clean up the global scheme_extensions after the test.
self.addCleanup(self.cleanup_scheme_extensions)
# Create a test partition
self.user_partition = UserPartition(
self.TEST_ID,
self.TEST_NAME,
self.TEST_DESCRIPTION,
self.TEST_GROUPS,
extensions[0].plugin,
self.TEST_PARAMETERS,
)
# Make sure the names are set on the schemes (which happens normally in code, but may not happen in tests).
self.user_partition.get_scheme(self.non_random_scheme.name)
self.user_partition.get_scheme(self.random_scheme.name)
def cleanup_scheme_extensions(self):
"""
Unset the UserPartition.scheme_extensions cache.
"""
UserPartition.scheme_extensions = None
class TestUserPartition(PartitionTestCase):
"""Test constructing UserPartitions"""
def test_construct(self):
user_partition = UserPartition(
self.TEST_ID,
self.TEST_NAME,
self.TEST_DESCRIPTION,
self.TEST_GROUPS,
MockUserPartitionScheme(),
self.TEST_PARAMETERS,
)
self.assertEqual(user_partition.id, self.TEST_ID)
self.assertEqual(user_partition.name, self.TEST_NAME)
self.assertEqual(user_partition.description, self.TEST_DESCRIPTION)
self.assertEqual(user_partition.groups, self.TEST_GROUPS)
self.assertEquals(user_partition.scheme.name, self.TEST_SCHEME_NAME)
self.assertEquals(user_partition.parameters, self.TEST_PARAMETERS)
def test_string_id(self):
user_partition = UserPartition(
"70",
self.TEST_NAME,
self.TEST_DESCRIPTION,
self.TEST_GROUPS,
MockUserPartitionScheme(),
self.TEST_PARAMETERS,
)
self.assertEqual(user_partition.id, 70)
def test_to_json(self):
jsonified = self.user_partition.to_json()
act_jsonified = {
"id": self.TEST_ID,
"name": self.TEST_NAME,
"description": self.TEST_DESCRIPTION,
"parameters": self.TEST_PARAMETERS,
"groups": [group.to_json() for group in self.TEST_GROUPS],
"version": self.user_partition.VERSION,
"scheme": self.TEST_SCHEME_NAME,
"active": True,
}
self.assertEqual(jsonified, act_jsonified)
def test_from_json(self):
jsonified = {
"id": self.TEST_ID,
"name": self.TEST_NAME,
"description": self.TEST_DESCRIPTION,
"parameters": self.TEST_PARAMETERS,
"groups": [group.to_json() for group in self.TEST_GROUPS],
"version": UserPartition.VERSION,
"scheme": "mock",
}
user_partition = UserPartition.from_json(jsonified)
self.assertEqual(user_partition.id, self.TEST_ID)
self.assertEqual(user_partition.name, self.TEST_NAME)
self.assertEqual(user_partition.description, self.TEST_DESCRIPTION)
self.assertEqual(user_partition.parameters, self.TEST_PARAMETERS)
for act_group in user_partition.groups:
self.assertIn(act_group.id, [0, 1])
exp_group = self.TEST_GROUPS[act_group.id]
self.assertEqual(exp_group.id, act_group.id)
self.assertEqual(exp_group.name, act_group.name)
def test_version_upgrade(self):
# Test that version 1 partitions did not have a scheme specified
# and have empty parameters
jsonified = {
"id": self.TEST_ID,
"name": self.TEST_NAME,
"description": self.TEST_DESCRIPTION,
"groups": [group.to_json() for group in self.TEST_GROUPS],
"version": 1,
}
user_partition = UserPartition.from_json(jsonified)
self.assertEqual(user_partition.scheme.name, "random")
self.assertEqual(user_partition.parameters, {})
self.assertTrue(user_partition.active)
def test_version_upgrade_2_to_3(self):
# Test that version 3 user partition raises error if 'scheme' field is
# not provided (same behavior as version 2)
jsonified = {
'id': self.TEST_ID,
"name": self.TEST_NAME,
"description": self.TEST_DESCRIPTION,
"parameters": self.TEST_PARAMETERS,
"groups": [group.to_json() for group in self.TEST_GROUPS],
"version": 2,
}
with self.assertRaisesRegexp(TypeError, "missing value key 'scheme'"):
UserPartition.from_json(jsonified)
# Test that version 3 partitions have a scheme specified
# and a field 'parameters' (optional while setting user partition but
# always present in response)
jsonified = {
"id": self.TEST_ID,
"name": self.TEST_NAME,
"description": self.TEST_DESCRIPTION,
"groups": [group.to_json() for group in self.TEST_GROUPS],
"version": 2,
"scheme": self.TEST_SCHEME_NAME,
}
user_partition = UserPartition.from_json(jsonified)
self.assertEqual(user_partition.scheme.name, self.TEST_SCHEME_NAME)
self.assertEqual(user_partition.parameters, {})
self.assertTrue(user_partition.active)
# now test that parameters dict is present in response with same value
# as provided
jsonified = {
"id": self.TEST_ID,
"name": self.TEST_NAME,
"description": self.TEST_DESCRIPTION,
"groups": [group.to_json() for group in self.TEST_GROUPS],
"parameters": self.TEST_PARAMETERS,
"version": 3,
"scheme": self.TEST_SCHEME_NAME,
}
user_partition = UserPartition.from_json(jsonified)
self.assertEqual(user_partition.parameters, self.TEST_PARAMETERS)
self.assertTrue(user_partition.active)
def test_from_json_broken(self):
# Missing field
jsonified = {
"name": self.TEST_NAME,
"description": self.TEST_DESCRIPTION,
"parameters": self.TEST_PARAMETERS,
"groups": [group.to_json() for group in self.TEST_GROUPS],
"version": UserPartition.VERSION,
"scheme": self.TEST_SCHEME_NAME,
}
with self.assertRaisesRegexp(TypeError, "missing value key 'id'"):
UserPartition.from_json(jsonified)
# Missing scheme
jsonified = {
'id': self.TEST_ID,
"name": self.TEST_NAME,
"description": self.TEST_DESCRIPTION,
"parameters": self.TEST_PARAMETERS,
"groups": [group.to_json() for group in self.TEST_GROUPS],
"version": UserPartition.VERSION,
}
with self.assertRaisesRegexp(TypeError, "missing value key 'scheme'"):
UserPartition.from_json(jsonified)
# Invalid scheme
jsonified = {
'id': self.TEST_ID,
"name": self.TEST_NAME,
"description": self.TEST_DESCRIPTION,
"parameters": self.TEST_PARAMETERS,
"groups": [group.to_json() for group in self.TEST_GROUPS],
"version": UserPartition.VERSION,
"scheme": "no_such_scheme",
}
with self.assertRaisesRegexp(UserPartitionError, "Unrecognized scheme"):
UserPartition.from_json(jsonified)
# Wrong version
jsonified = {
'id': self.TEST_ID,
"name": self.TEST_NAME,
"description": self.TEST_DESCRIPTION,
"parameters": self.TEST_PARAMETERS,
"groups": [group.to_json() for group in self.TEST_GROUPS],
"version": -1,
"scheme": self.TEST_SCHEME_NAME,
}
with self.assertRaisesRegexp(TypeError, "has unexpected version"):
UserPartition.from_json(jsonified)
# Has extra key - should not be a problem
jsonified = {
'id': self.TEST_ID,
"name": self.TEST_NAME,
"description": self.TEST_DESCRIPTION,
"parameters": self.TEST_PARAMETERS,
"groups": [group.to_json() for group in self.TEST_GROUPS],
"version": UserPartition.VERSION,
"scheme": "mock",
"programmer": "Cale",
}
user_partition = UserPartition.from_json(jsonified)
self.assertNotIn("programmer", user_partition.to_json())
# No error on missing parameters key (which is optional)
jsonified = {
'id': self.TEST_ID,
"name": self.TEST_NAME,
"description": self.TEST_DESCRIPTION,
"groups": [group.to_json() for group in self.TEST_GROUPS],
"version": UserPartition.VERSION,
"scheme": "mock",
}
user_partition = UserPartition.from_json(jsonified)
self.assertEqual(user_partition.parameters, {})
def test_get_group(self):
"""
UserPartition.get_group correctly returns the group referenced by the
`group_id` parameter, or raises NoSuchUserPartitionGroupError when
the lookup fails.
"""
self.assertEqual(
self.user_partition.get_group(self.TEST_GROUPS[0].id),
self.TEST_GROUPS[0]
)
self.assertEqual(
self.user_partition.get_group(self.TEST_GROUPS[1].id),
self.TEST_GROUPS[1]
)
with self.assertRaises(NoSuchUserPartitionGroupError):
self.user_partition.get_group(3)
def test_forward_compatibility(self):
# If the user partition version is updated in a release,
# then the release is rolled back, courses might contain
# version numbers greater than the currently deployed
# version number.
newer_version_json = {
"id": self.TEST_ID,
"name": self.TEST_NAME,
"description": self.TEST_DESCRIPTION,
"groups": [group.to_json() for group in self.TEST_GROUPS],
"version": UserPartition.VERSION + 1,
"scheme": "mock",
"additional_new_field": "foo",
}
partition = UserPartition.from_json(newer_version_json)
self.assertEqual(partition.id, self.TEST_ID)
self.assertEqual(partition.name, self.TEST_NAME)
class MockPartitionService(PartitionService):
"""
Mock PartitionService for testing.
"""
def __init__(self, course, **kwargs):
super(MockPartitionService, self).__init__(**kwargs)
self._course = course
def get_course(self):
return self._course
class PartitionServiceBaseClass(PartitionTestCase):
"""
Base test class for testing the PartitionService.
"""
def setUp(self):
super(PartitionServiceBaseClass, self).setUp()
self.course = Mock(id=CourseLocator('org_0', 'course_0', 'run_0'))
self.partition_service = self._create_service("ma")
def _create_service(self, username, cache=None):
"""Convenience method to generate a MockPartitionService for a user."""
# Derive a "user_id" from the username, just so we don't have to add an
# extra param to this method. Just has to be unique per user.
user_id = abs(hash(username))
self.user = Mock(
username=username, email='{}@edx.org'.format(username), is_staff=False, is_active=True, id=user_id
)
self.course.user_partitions = [self.user_partition]
return MockPartitionService(
self.course,
course_id=self.course.id,
track_function=Mock(),
cache=cache
)
class TestPartitionService(PartitionServiceBaseClass):
"""
Test getting a user's group out of a partition
"""
def test_get_user_group_id_for_partition(self):
# assign the first group to be returned
user_partition_id = self.user_partition.id
groups = self.user_partition.groups
self.user_partition.scheme.current_group = groups[0]
# get a group assigned to the user
group1_id = self.partition_service.get_user_group_id_for_partition(self.user, user_partition_id)
self.assertEqual(group1_id, groups[0].id)
# switch to the second group and verify that it is returned for the user
self.user_partition.scheme.current_group = groups[1]
group2_id = self.partition_service.get_user_group_id_for_partition(self.user, user_partition_id)
self.assertEqual(group2_id, groups[1].id)
def test_caching(self):
username = "psvc_cache_user"
user_partition_id = self.user_partition.id
shared_cache = {}
# Two MockPartitionService objects that share the same cache:
ps_shared_cache_1 = self._create_service(username, shared_cache)
ps_shared_cache_2 = self._create_service(username, shared_cache)
# A MockPartitionService with its own local cache
ps_diff_cache = self._create_service(username, {})
# A MockPartitionService that never uses caching.
ps_uncached = self._create_service(username)
# Set the group we expect users to be placed into
first_group = self.user_partition.groups[0]
self.user_partition.scheme.current_group = first_group
# Make sure our partition services all return the right thing, but skip
# ps_shared_cache_2 so we can see if its cache got updated anyway.
for part_svc in [ps_shared_cache_1, ps_diff_cache, ps_uncached]:
self.assertEqual(
first_group.id,
part_svc.get_user_group_id_for_partition(self.user, user_partition_id)
)
# Now select a new target group
second_group = self.user_partition.groups[1]
self.user_partition.scheme.current_group = second_group
# Both of the shared cache entries should return the old value, even
# ps_shared_cache_2, which was never asked for the value the first time
# Likewise, our separately cached piece should return the original answer
for part_svc in [ps_shared_cache_1, ps_shared_cache_2, ps_diff_cache]:
self.assertEqual(
first_group.id,
part_svc.get_user_group_id_for_partition(self.user, user_partition_id)
)
# Our uncached service should be accurate.
self.assertEqual(
second_group.id,
ps_uncached.get_user_group_id_for_partition(self.user, user_partition_id)
)
# And a newly created service should see the right thing
ps_new_cache = self._create_service(username, {})
self.assertEqual(
second_group.id,
ps_new_cache.get_user_group_id_for_partition(self.user, user_partition_id)
)
def test_get_group(self):
"""
Test that a partition group is assigned to a user.
"""
groups = self.user_partition.groups
# assign first group and verify that it is returned for the user
self.user_partition.scheme.current_group = groups[0]
group1 = self.partition_service.get_group(self.user, self.user_partition)
self.assertEqual(group1, groups[0])
# switch to the second group and verify that it is returned for the user
self.user_partition.scheme.current_group = groups[1]
group2 = self.partition_service.get_group(self.user, self.user_partition)
self.assertEqual(group2, groups[1])
class TestGetCourseUserPartitions(PartitionServiceBaseClass):
"""
Test the helper method get_all_partitions_for_course.
"""
def setUp(self):
super(TestGetCourseUserPartitions, self).setUp()
# django.conf.settings is not available when nosetests are run
TestGetCourseUserPartitions._enable_enrollment_track_partition(True)
@staticmethod
def _enable_enrollment_track_partition(enable):
"""
Enable or disable the feature flag for the enrollment track user partition.
"""
FEATURES['ENABLE_ENROLLMENT_TRACK_USER_PARTITION'] = enable
def test_enrollment_track_partition_added(self):
"""
Test that the dynamic enrollment track scheme is added if there is no conflict with the user partition ID.
"""
all_partitions = get_all_partitions_for_course(self.course)
self.assertEqual(2, len(all_partitions))
self.assertEqual(self.TEST_SCHEME_NAME, all_partitions[0].scheme.name)
enrollment_track_partition = all_partitions[1]
self.assertEqual(self.ENROLLMENT_TRACK_SCHEME_NAME, enrollment_track_partition.scheme.name)
self.assertEqual(unicode(self.course.id), enrollment_track_partition.parameters['course_id'])
self.assertEqual(ENROLLMENT_TRACK_PARTITION_ID, enrollment_track_partition.id)
def test_enrollment_track_partition_not_added_if_conflict(self):
"""
Test that the dynamic enrollment track scheme is NOT added if a UserPartition exists with that ID.
"""
self.user_partition = UserPartition(
ENROLLMENT_TRACK_PARTITION_ID,
self.TEST_NAME,
self.TEST_DESCRIPTION,
self.TEST_GROUPS,
self.non_random_scheme,
self.TEST_PARAMETERS,
)
self.course.user_partitions = [self.user_partition]
all_partitions = get_all_partitions_for_course(self.course)
self.assertEqual(1, len(all_partitions))
self.assertEqual(self.TEST_SCHEME_NAME, all_partitions[0].scheme.name)
def test_enrollment_track_partition_not_added_if_disabled(self):
"""
Test that the dynamic enrollment track scheme is NOT added if the settings FEATURE flag is disabled.
"""
TestGetCourseUserPartitions._enable_enrollment_track_partition(False)
all_partitions = get_all_partitions_for_course(self.course)
self.assertEqual(1, len(all_partitions))
self.assertEqual(self.TEST_SCHEME_NAME, all_partitions[0].scheme.name)
def test_filter_inactive_user_partitions(self):
"""
Tests supplying the `active_only` parameter.
"""
self.user_partition = UserPartition(
self.TEST_ID,
self.TEST_NAME,
self.TEST_DESCRIPTION,
self.TEST_GROUPS,
self.non_random_scheme,
self.TEST_PARAMETERS,
active=False
)
self.course.user_partitions = [self.user_partition]
all_partitions = get_all_partitions_for_course(self.course, active_only=True)
self.assertEqual(1, len(all_partitions))
self.assertEqual(self.ENROLLMENT_TRACK_SCHEME_NAME, all_partitions[0].scheme.name)
all_partitions = get_all_partitions_for_course(self.course, active_only=False)
self.assertEqual(2, len(all_partitions))
self.assertEqual(self.TEST_SCHEME_NAME, all_partitions[0].scheme.name)
self.assertEqual(self.ENROLLMENT_TRACK_SCHEME_NAME, all_partitions[1].scheme.name)

0
openedx/core/lib/xblock_fields/__init__.py Normal file
View File

290
openedx/core/lib/xblock_fields/fields.py Normal file
View File

@@ -0,0 +1,290 @@
import time
import logging
import re
from xblock.fields import JSONField
import datetime
import dateutil.parser
from pytz import UTC
log = logging.getLogger(__name__)
class Date(JSONField):
'''
Date fields know how to parse and produce json (iso) compatible formats. Converts to tz aware datetimes.
'''
# See note below about not defaulting these
CURRENT_YEAR = datetime.datetime.now(UTC).year
PREVENT_DEFAULT_DAY_MON_SEED1 = datetime.datetime(CURRENT_YEAR, 1, 1, tzinfo=UTC)
PREVENT_DEFAULT_DAY_MON_SEED2 = datetime.datetime(CURRENT_YEAR, 2, 2, tzinfo=UTC)
MUTABLE = False
def _parse_date_wo_default_month_day(self, field):
"""
Parse the field as an iso string but prevent dateutils from defaulting the day or month while
allowing it to default the other fields.
"""
# It's not trivial to replace dateutil b/c parsing timezones as Z, +03:30, -400 is hard in python
# however, we don't want dateutil to default the month or day (but some tests at least expect
# us to default year); so, we'll see if dateutil uses the defaults for these the hard way
result = dateutil.parser.parse(field, default=self.PREVENT_DEFAULT_DAY_MON_SEED1)
result_other = dateutil.parser.parse(field, default=self.PREVENT_DEFAULT_DAY_MON_SEED2)
if result != result_other:
log.warning("Field {0} is missing month or day".format(self.name))
return None
if result.tzinfo is None:
result = result.replace(tzinfo=UTC)
return result
def from_json(self, field):
"""
Parse an optional metadata key containing a time: if present, complain
if it doesn't parse.
Return None if not present or invalid.
"""
if field is None:
return field
elif field is "":
return None
elif isinstance(field, basestring):
return self._parse_date_wo_default_month_day(field)
elif isinstance(field, (int, long, float)):
return datetime.datetime.fromtimestamp(field / 1000, UTC)
elif isinstance(field, time.struct_time):
return datetime.datetime.fromtimestamp(time.mktime(field), UTC)
elif isinstance(field, datetime.datetime):
return field
else:
msg = "Field {0} has bad value '{1}'".format(
self.name, field)
raise TypeError(msg)
def to_json(self, value):
"""
Convert a time struct to a string
"""
if value is None:
return None
if isinstance(value, time.struct_time):
# struct_times are always utc
return time.strftime('%Y-%m-%dT%H:%M:%SZ', value)
elif isinstance(value, datetime.datetime):
if value.tzinfo is None or value.utcoffset().total_seconds() == 0:
if value.year < 1900:
# strftime doesn't work for pre-1900 dates, so use
# isoformat instead
return value.isoformat()
# isoformat adds +00:00 rather than Z
return value.strftime('%Y-%m-%dT%H:%M:%SZ')
else:
return value.isoformat()
else:
raise TypeError("Cannot convert {!r} to json".format(value))
enforce_type = from_json
TIMEDELTA_REGEX = re.compile(r'^((?P<days>\d+?) day(?:s?))?(\s)?((?P<hours>\d+?) hour(?:s?))?(\s)?((?P<minutes>\d+?) minute(?:s)?)?(\s)?((?P<seconds>\d+?) second(?:s)?)?$')
class Timedelta(JSONField):
# Timedeltas are immutable, see http://docs.python.org/2/library/datetime.html#available-types
MUTABLE = False
def from_json(self, time_str):
"""
time_str: A string with the following components:
<D> day[s] (optional)
<H> hour[s] (optional)
<M> minute[s] (optional)
<S> second[s] (optional)
Returns a datetime.timedelta parsed from the string
"""
if time_str is None:
return None
if isinstance(time_str, datetime.timedelta):
return time_str
parts = TIMEDELTA_REGEX.match(time_str)
if not parts:
return
parts = parts.groupdict()
time_params = {}
for (name, param) in parts.iteritems():
if param:
time_params[name] = int(param)
return datetime.timedelta(**time_params)
def to_json(self, value):
if value is None:
return None
values = []
for attr in ('days', 'hours', 'minutes', 'seconds'):
cur_value = getattr(value, attr, 0)
if cur_value > 0:
values.append("%d %s" % (cur_value, attr))
return ' '.join(values)
def enforce_type(self, value):
"""
Ensure that when set explicitly the Field is set to a timedelta
"""
if isinstance(value, datetime.timedelta) or value is None:
return value
return self.from_json(value)
class RelativeTime(JSONField):
"""
Field for start_time and end_time video module properties.
It was decided, that python representation of start_time and end_time
should be python datetime.timedelta object, to be consistent with
common time representation.
At the same time, serialized representation should be "HH:MM:SS"
This format is convenient to use in XML (and it is used now),
and also it is used in frond-end studio editor of video module as format
for start and end time fields.
In database we previously had float type for start_time and end_time fields,
so we are checking it also.
Python object of RelativeTime is datetime.timedelta.
JSONed representation of RelativeTime is "HH:MM:SS"
"""
# Timedeltas are immutable, see http://docs.python.org/2/library/datetime.html#available-types
MUTABLE = False
@classmethod
def isotime_to_timedelta(cls, value):
"""
Validate that value in "HH:MM:SS" format and convert to timedelta.
Validate that user, that edits XML, sets proper format, and
that max value that can be used by user is "23:59:59".
"""
try:
obj_time = time.strptime(value, '%H:%M:%S')
except ValueError as e:
raise ValueError(
"Incorrect RelativeTime value {!r} was set in XML or serialized. "
"Original parse message is {}".format(value, e.message)
)
return datetime.timedelta(
hours=obj_time.tm_hour,
minutes=obj_time.tm_min,
seconds=obj_time.tm_sec
)
def from_json(self, value):
"""
Convert value is in 'HH:MM:SS' format to datetime.timedelta.
If not value, returns 0.
If value is float (backward compatibility issue), convert to timedelta.
"""
if not value:
return datetime.timedelta(seconds=0)
if isinstance(value, datetime.timedelta):
return value
# We've seen serialized versions of float in this field
if isinstance(value, float):
return datetime.timedelta(seconds=value)
if isinstance(value, basestring):
return self.isotime_to_timedelta(value)
msg = "RelativeTime Field {0} has bad value '{1!r}'".format(self.name, value)
raise TypeError(msg)
def to_json(self, value):
"""
Convert datetime.timedelta to "HH:MM:SS" format.
If not value, return "00:00:00"
Backward compatibility: check if value is float, and convert it. No exceptions here.
If value is not float, but is exceed 23:59:59, raise exception.
"""
if not value:
return "00:00:00"
if isinstance(value, float): # backward compatibility
value = min(value, 86400)
return self.timedelta_to_string(datetime.timedelta(seconds=value))
if isinstance(value, datetime.timedelta):
if value.total_seconds() > 86400: # sanity check
raise ValueError(
"RelativeTime max value is 23:59:59=86400.0 seconds, "
"but {} seconds is passed".format(value.total_seconds())
)
return self.timedelta_to_string(value)
raise TypeError("RelativeTime: cannot convert {!r} to json".format(value))
def timedelta_to_string(self, value):
"""
Makes first 'H' in str representation non-optional.
str(timedelta) has [H]H:MM:SS format, which is not suitable
for front-end (and ISO time standard), so we force HH:MM:SS format.
"""
stringified = str(value)
if len(stringified) == 7:
stringified = '0' + stringified
return stringified
def enforce_type(self, value):
"""
Ensure that when set explicitly the Field is set to a timedelta
"""
if isinstance(value, datetime.timedelta) or value is None:
return value
return self.from_json(value)
class TimeInfo(object):
"""
This is a simple object that calculates and stores datetime information for an XModule
based on the due date and the grace period string
So far it parses out three different pieces of time information:
self.display_due_date - the 'official' due date that gets displayed to students
self.grace_period - the length of the grace period
self.close_date - the real due date
"""
_delta_standin = Timedelta()
def __init__(self, due_date, grace_period_string_or_timedelta):
if due_date is not None:
self.display_due_date = due_date
else:
self.display_due_date = None
if grace_period_string_or_timedelta is not None and self.display_due_date:
if isinstance(grace_period_string_or_timedelta, basestring):
try:
self.grace_period = TimeInfo._delta_standin.from_json(grace_period_string_or_timedelta)
except:
log.error("Error parsing the grace period {0}".format(grace_period_string_or_timedelta))
raise
else:
self.grace_period = grace_period_string_or_timedelta
self.close_date = self.display_due_date + self.grace_period
else:
self.grace_period = None
self.close_date = self.display_due_date

239
openedx/core/lib/xblock_fields/inherited_fields.py Normal file
View File

@@ -0,0 +1,239 @@
"""
Inherited fields for all XBlocks.
"""
from __future__ import absolute_import
from datetime import datetime
from django.conf import settings
from openedx.core.lib.partitions.partitions import UserPartition
from pytz import utc
from xblock.fields import Boolean, Dict, Float, Integer, List, Scope, String, XBlockMixin
from .fields import Date, Timedelta
DEFAULT_START_DATE = datetime(2030, 1, 1, tzinfo=utc)
# Make '_' a no-op so we can scrape strings
# Using lambda instead of `django.utils.translation.ugettext_noop` because Django cannot be imported in this file
_ = lambda text: text
class UserPartitionList(List):
"""Special List class for listing UserPartitions"""
def from_json(self, values):
return [UserPartition.from_json(v) for v in values]
def to_json(self, values):
return [user_partition.to_json()
for user_partition in values]
class InheritanceMixin(XBlockMixin):
"""Field definitions for inheritable fields."""
graded = Boolean(
help="Whether this module contributes to the final course grade",
scope=Scope.settings,
default=False,
)
start = Date(
help="Start time when this module is visible",
default=DEFAULT_START_DATE,
scope=Scope.settings
)
due = Date(
display_name=_("Due Date"),
help=_("Enter the default date by which problems are due."),
scope=Scope.settings,
)
visible_to_staff_only = Boolean(
help=_("If true, can be seen only by course staff, regardless of start date."),
default=False,
scope=Scope.settings,
)
course_edit_method = String(
display_name=_("Course Editor"),
help=_("Enter the method by which this course is edited (\"XML\" or \"Studio\")."),
default="Studio",
scope=Scope.settings,
deprecated=True # Deprecated because user would not change away from Studio within Studio.
)
giturl = String(
display_name=_("GIT URL"),
help=_("Enter the URL for the course data GIT repository."),
scope=Scope.settings
)
xqa_key = String(
display_name=_("XQA Key"),
help=_("This setting is not currently supported."), scope=Scope.settings,
deprecated=True
)
annotation_storage_url = String(
help=_("Enter the location of the annotation storage server. The textannotation, videoannotation, and imageannotation advanced modules require this setting."),
scope=Scope.settings,
default="http://your_annotation_storage.com",
display_name=_("URL for Annotation Storage")
)
annotation_token_secret = String(
help=_("Enter the secret string for annotation storage. The textannotation, videoannotation, and imageannotation advanced modules require this string."),
scope=Scope.settings,
default="xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx",
display_name=_("Secret Token String for Annotation")
)
graceperiod = Timedelta(
help="Amount of time after the due date that submissions will be accepted",
scope=Scope.settings,
)
group_access = Dict(
help=_("Enter the ids for the content groups this problem belongs to."),
scope=Scope.settings,
)
showanswer = String(
display_name=_("Show Answer"),
help=_(
# Translators: DO NOT translate the words in quotes here, they are
# specific words for the acceptable values.
'Specify when the Show Answer button appears for each problem. '
'Valid values are "always", "answered", "attempted", "closed", '
'"finished", "past_due", "correct_or_past_due", and "never".'
),
scope=Scope.settings,
default="finished",
)
show_correctness = String(
display_name=_("Show Results"),
help=_(
# Translators: DO NOT translate the words in quotes here, they are
# specific words for the acceptable values.
'Specify when to show answer correctness and score to learners. '
'Valid values are "always", "never", and "past_due".'
),
scope=Scope.settings,
default="always",
)
rerandomize = String(
display_name=_("Randomization"),
help=_(
# Translators: DO NOT translate the words in quotes here, they are
# specific words for the acceptable values.
'Specify the default for how often variable values in a problem are randomized. '
'This setting should be set to "never" unless you plan to provide a Python '
'script to identify and randomize values in most of the problems in your course. '
'Valid values are "always", "onreset", "never", and "per_student".'
),
scope=Scope.settings,
default="never",
)
days_early_for_beta = Float(
display_name=_("Days Early for Beta Users"),
help=_("Enter the number of days before the start date that beta users can access the course."),
scope=Scope.settings,
default=None,
)
static_asset_path = String(
display_name=_("Static Asset Path"),
help=_("Enter the path to use for files on the Files & Uploads page. This value overrides the Studio default, c4x://."),
scope=Scope.settings,
default='',
)
use_latex_compiler = Boolean(
display_name=_("Enable LaTeX Compiler"),
help=_("Enter true or false. If true, you can use the LaTeX templates for HTML components and advanced Problem components."),
default=False,
scope=Scope.settings
)
max_attempts = Integer(
display_name=_("Maximum Attempts"),
help=_("Enter the maximum number of times a student can try to answer problems. By default, Maximum Attempts is set to null, meaning that students have an unlimited number of attempts for problems. You can override this course-wide setting for individual problems. However, if the course-wide setting is a specific number, you cannot set the Maximum Attempts for individual problems to unlimited."),
values={"min": 0}, scope=Scope.settings
)
matlab_api_key = String(
display_name=_("Matlab API key"),
help=_("Enter the API key provided by MathWorks for accessing the MATLAB Hosted Service. "
"This key is granted for exclusive use in this course for the specified duration. "
"Do not share the API key with other courses. Notify MathWorks immediately "
"if you believe the key is exposed or compromised. To obtain a key for your course, "
"or to report an issue, please contact moocsupport@mathworks.com"),
scope=Scope.settings
)
# This is should be scoped to content, but since it's defined in the policy
# file, it is currently scoped to settings.
user_partitions = UserPartitionList(
display_name=_("Group Configurations"),
help=_("Enter the configurations that govern how students are grouped together."),
default=[],
scope=Scope.settings
)
video_speed_optimizations = Boolean(
display_name=_("Enable video caching system"),
help=_("Enter true or false. If true, video caching will be used for HTML5 videos."),
default=True,
scope=Scope.settings
)
video_bumper = Dict(
display_name=_("Video Pre-Roll"),
help=_(
"Identify a video, 5-10 seconds in length, to play before course videos. Enter the video ID from "
"the Video Uploads page and one or more transcript files in the following format: {format}. "
"For example, an entry for a video with two transcripts looks like this: {example}"
).format(
format='{"video_id": "ID", "transcripts": {"language": "/static/filename.srt"}}',
example=(
'{'
'"video_id": "77cef264-d6f5-4cf2-ad9d-0178ab8c77be", '
'"transcripts": {"en": "/static/DemoX-D01_1.srt", "uk": "/static/DemoX-D01_1_uk.srt"}'
'}'
),
),
scope=Scope.settings
)
# TODO: Remove this Django dependency! It really doesn't belong here.
reset_key = "DEFAULT_SHOW_RESET_BUTTON"
default_reset_button = getattr(settings, reset_key) if hasattr(settings, reset_key) else False
show_reset_button = Boolean(
display_name=_("Show Reset Button for Problems"),
help=_(
"Enter true or false. If true, problems in the course default to always displaying a 'Reset' button. "
"You can override this in each problem's settings. All existing problems are affected when "
"this course-wide setting is changed."
),
scope=Scope.settings,
default=default_reset_button
)
edxnotes = Boolean(
display_name=_("Enable Student Notes"),
help=_("Enter true or false. If true, students can use the Student Notes feature."),
default=False,
scope=Scope.settings
)
edxnotes_visibility = Boolean(
display_name="Student Notes Visibility",
help=_("Indicates whether Student Notes are visible in the course. "
"Students can also show or hide their notes in the courseware."),
default=True,
scope=Scope.user_info
)
in_entrance_exam = Boolean(
display_name=_("Tag this module as part of an Entrance Exam section"),
help=_("Enter true or false. If true, answer submissions for problem modules will be "
"considered in the Entrance Exam scoring/gating algorithm."),
scope=Scope.settings,
default=False
)
self_paced = Boolean(
display_name=_('Self Paced'),
help=_(
'Set this to "true" to mark this course as self-paced. Self-paced courses do not have '
'due dates for assignments, and students can progress through the course at any rate before '
'the course ends.'
),
default=False,
scope=Scope.settings
)